agentmesh_lightning

Agent Lightning — RL Training Governance

[!IMPORTANT] Public Preview — The agentmesh-lightning package on PyPI is a public preview release. APIs may change before GA.

Train AI agents with RL while maintaining 0% policy violations.

Part of the Agent Governance Toolkit

🎯 Overview

This package provides governed RL training integration:

• Agent-Lightning = Training/Optimization (the "brains")
• Agent-OS = Governance/Safety (the "guardrails")

Result: Agents learn to be smart AND safe from the start.

Note: This package was extracted from agent_os.integrations.agent_lightning. The old import path still works via a backward-compatibility shim but new code should import from agent_lightning_gov directly.

🚀 Quick Start

pip install agentmesh-lightning
# Optional: pip install agent-os-kernel  # for kernel integration

from agent_lightning_gov import GovernedRunner, PolicyReward
from agent_os import KernelSpace
from agent_os.policies import SQLPolicy, CostControlPolicy

# 1. Create governed kernel
kernel = KernelSpace(policy=[
    SQLPolicy(deny=["DROP", "DELETE"]),
    CostControlPolicy(max_cost_usd=100)
])

# 2. Create governed runner
runner = GovernedRunner(kernel)

# 3. Create policy-aware reward function
def base_accuracy(rollout):
    return rollout.task_output.accuracy if rollout.success else 0.0

reward_fn = PolicyReward(kernel, base_reward_fn=base_accuracy)

# 4. Train with Agent-Lightning
from agentlightning import Trainer
trainer = Trainer(
    runner=runner,
    reward_fn=reward_fn,
    algorithm="GRPO"
)

trainer.train(num_epochs=100)

📊 Key Benefits

🔧 Components

GovernedRunner

Agent-Lightning runner that enforces policies during execution:

from agent_lightning_gov import GovernedRunner

runner = GovernedRunner(
    kernel,
    fail_on_violation=False,   # Continue but penalize
    log_violations=True,        # Log all violations
)

# Execute a task
rollout = await runner.step(task_input)
print(f"Violations: {len(rollout.violations)}")
print(f"Total penalty: {rollout.total_penalty}")

PolicyReward

Converts policy violations to RL penalties:

from agent_lightning_gov import PolicyReward, RewardConfig

config = RewardConfig(
    critical_penalty=-100.0,  # Harsh penalty for critical violations
    high_penalty=-50.0,
    medium_penalty=-10.0,
    low_penalty=-1.0,
    clean_bonus=5.0,          # Bonus for no violations
)

reward_fn = PolicyReward(kernel, config=config)

# Calculate reward
reward = reward_fn(rollout)  # Base reward + policy penalties

GovernedEnvironment

Gym-compatible training environment:

from agent_lightning_gov import GovernedEnvironment

env = GovernedEnvironment(
    kernel,
    config=EnvironmentConfig(
        max_steps=100,
        terminate_on_critical=True,
    )
)

# Standard Gym interface
state, info = env.reset()
while not env.terminated:
    action = agent.get_action(state)
    state, reward, terminated, truncated, info = env.step(action)

FlightRecorderEmitter

Export audit logs to LightningStore:

from agent_os import FlightRecorder
from agent_lightning_gov import FlightRecorderEmitter

recorder = FlightRecorder()
emitter = FlightRecorderEmitter(recorder)

# Export to LightningStore
emitter.emit_to_store(lightning_store)

# Or export to file for analysis
emitter.export_to_file("training_audit.json")

# Get violation summary
summary = emitter.get_violation_summary()
print(f"Violation rate: {summary['violation_rate']:.1%}")

Ecosystem

Agent Lightning is one of 7 packages in the Agent Governance Toolkit:

📋 License

MIT — see LICENSE.