用 Agent OS 治理强化学习训练基于 Agent-Lightning 打造天生安全的 SQL Agent【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit在 Agent Governance Toolkit 中Agent-Lightning 训练示例 展示了一种关键理念安全不应该在训练完成之后才补课而应该作为奖励信号直接内化进强化学习RL训练过程。本文以示例自带的 SQL Agent 训练脚本 sql_agent.py 为主线讲解如何用GovernedRunner、PolicyReward、GovernedEnvironment三件套把 Agent OS 的策略引擎接入 Agent-Lightning 训练循环让智能体在学会生成准确 SQL 的同时从一开始就绝不触碰DROP/DELETE等危险操作、不超出成本上限。读完本文你将掌握训练即治理的完整落地路径从策略初始化、受治理的 rollout 执行到把策略违规折算为负向奖励、再到训练统计的可观测化。为什么要对 RL 训练本身做治理常规的治理思路是运行时拦截Agent 做出危险动作时由策略引擎在推理/执行阶段将其阻断。但 ADR-0024: RL Training Governance with Violation Penalties 指出了单纯依赖运行时拦截在训练场景下的三个盲区奖励错配如果违规行为能带来更高的任务完成度RL 优化器反而会奖励违规行为让智能体学会绕过治理规避学习智能体可能摸索出规避治理检查的路径而不是真正理解策略约束数据污染训练轨迹中混入不安全的动作序列会污染整个训练数据集。因此 ADR-0024 的决策是把治理下沉到奖励信号本身——每次策略违规都从奖励中扣除惩罚严重程度映射到惩罚量级从而让智能体在梯度更新的过程中自发地学会违规 高代价。对应的核心实现位于 agent-governance-python/agent-lightning 包PyPI 包名为agentmesh-lightning公开预览阶段API 在正式发布前可能调整。该包的角色定位是Agent-Lightning 负责训练与优化大脑Agent OS 负责治理与安全护栏两者结合得到既聪明又安全的 Agent。整体架构三个组件各司其职agent_lightning_gov模块见init.py对外暴露四个核心构件其中训练示例用到三个组件职责源码位置GovernedRunner用内核KernelSpace包裹 Agent 执行收集违规记录兼容 Agent-Lightning Trainerrunner.pyPolicyReward把策略违规折算成负向 RL 奖励包装任意基础奖励函数reward.pyGovernedEnvironmentGym/Gymnasium 风格的训练环境逐步执行动作、施加违规惩罚、支持关键违规终止environment.pyFlightRecorderEmitter把训练审计跨度span导出到 LightningStore 或文件emitter.py示例文档将工作流程归纳为四步这正是本文要展开的骨架GovernedRunner用策略检查包裹 Agent 执行PolicyReward把违规转换为负向 RL 奖励Agent 在训练中学会避开策略违规最终产出从第一天起就安全的 Agent。环境准备按照示例文档的 README 要求安装两个依赖pip install agent-os-kernel agentlightning其中agent-os-kernel提供策略内核KernelSpace 与 SQLPolicy 等策略类agentlightning提供 RL 训练框架。若想直接使用治理集成包也可以安装agentmesh-lightning并从agent_lightning_gov导入治理组件该包从agent_os.integrations.agent_lightning抽取而来旧导入路径仍通过兼容垫片可用新代码应直接使用agent_lightning_gov。运行示例python sql_agent.py逐行拆解 sql_agent.py一个受治理的训练循环sql_agent.py 是完整的可运行示例。虽然其中MockKernelSpace、MockSQLPolicy、MockCostControlPolicy是为了演示而简化的替身文件头部注释明确写着 MOCK COMPONENTS (Replace with real implementations)但它完整呈现了与真实内核一致的调用契约可以直接替换为 Agent OS 的真实实现。第 1 步初始化带策略的内核kernel MockKernelSpace(policy[ MockSQLPolicy( allow[SELECT, INSERT, UPDATE], deny[DROP, DELETE, TRUNCATE], ), MockCostControlPolicy(max_cost_usd100), ])这里定义了本示例的两条治理规则SQLPolicy允许SELECT/INSERT/UPDATE拒绝DROP/DELETE/TRUNCATE——即只读为主、危险 DDL/DML 一律阻断CostControlPolicy单次查询成本上限 100 美元即成本治理。从真实实现的文档字符串看与GovernedRunner配套的典型内核构造方式是from agent_os import KernelSpace from agent_os.policies import SQLPolicy, CostControlPolicy kernel KernelSpace(policy[ SQLPolicy(deny[DROP, DELETE]), CostControlPolicy(max_cost_usd100), ])内核之所以是关键是因为 GovernedRunner.step() 会优先调用kernel.execute_async()回退到kernel.execute()把每个动作都送进策略引擎做确定性评估——这正是 ADR-0024 中确定性动作评估的设计延续。第 2 步创建 GovernedRunnerrunner GovernedRunner( kernel, fail_on_violationFalse, log_violationsTrue, )构造函数签名见 runner.py支持三个关键参数参数默认值作用kernel必填已加载策略的 Agent OS KernelSpacefail_on_violationFalse为True时违规且被阻断直接抛PolicyViolationError训练终止False则继续训练、仅施加惩罚log_violationsTrue是否将每次违规写入日志violation_callbackNone每次违规触发的可选回调可用于外部审计接入随后示例执行runner.init(MockAgent()) runner.init_worker(0, None)init()负责挂载内核钩子如果内核暴露on_policy_violation与on_signalRunner 会注册_handle_violation和_handle_signal回调从而把内核在执行期间的违规/信号实时捕获进 rollout 记录。第 3 步创建策略感知的奖励函数def accuracy_reward(rollout): if rollout.success and rollout.task_output: return rollout.task_output.get(accuracy, 0.0) return 0.0 reward_fn PolicyReward(kernel, base_reward_fnaccuracy_reward)PolicyReward的设计是包装任意基础奖励函数见 reward.pybase_reward_fn负责奖励任务本身的质量这里用accuracy字段衡量 SQL 生成准确度PolicyReward在它之上叠加策略惩罚。最终奖励计算公式为final_reward base_reward penalty(违规惩罚恒为负值)若启用multiplicative模式则改为final_reward base_reward * multiplicative_factor默认 0.5。此外还有两个边界保护min_reward默认 -100与max_reward默认 100把奖励钳制在合理区间防止极端负值破坏训练稳定性。默认的惩罚配置RewardConfig与PolicyViolation的严重度映射完全对齐见 runner.py 与 reward.py严重度违规记录惩罚PolicyViolationRL 奖励惩罚RewardConfigcritical100.0-100.0high50.0-50.0medium10.0-10.0low1.0-1.0一个值得注意的细节PolicyViolation的penalty字段在构造时由严重度表自动推导但调用方显式传入penalty时以调用方为准——这一设计允许对个别违规单独加权避免覆盖调用方意图见 runner.py 的注释。PolicyReward还内置了干净执行奖励当一次 rollout 没有任何违规时在最终奖励上追加clean_bonus默认 5.0形成违规扣分、守规加分的双向激励。第 4 步模拟训练回合示例用 6 条代表性查询模拟了训练过程test_queries [ SELECT * FROM users WHERE id 1, INSERT INTO logs (msg) VALUES (hello), DROP TABLE users, # Should be blocked! UPDATE users SET name John WHERE id 1, DELETE FROM users WHERE id 1, # Should be blocked! SELECT COUNT(*) FROM orders, ]每回合的核心循环只有三行rollout await runner.step(query) # 受治理的执行 reward reward_fn(rollout, emitFalse) # 计算带惩罚的奖励 violations_count len(rollout.violations) # 统计违规GovernedRunner.step()返回GovernedRollout见 runner.py其中携带success、violationsPolicyViolation列表、signals_sent、total_penalty各违规惩罚之和与execution_time_ms等治理元数据。这里有一个值得关注的并发设计step()通过contextvars.ContextVar把每次调用的违规/信号缓冲绑定到当前 asyncio 上下文保证同一个 Runner 上并发执行多个step()时各 rollout 的违规记录不会互相串扰见 runner.py。PolicyReward.__call__在emitTrue默认时还会通过agentlightning.emitter.emit_reward上报多维奖励final/base/policy_penalty并附带agent_os.violation_count、agent_os.policy_compliant属性供训练可视化与审计使用。第 5 步输出训练统计训练结束后示例分别拉取两套统计stats runner.get_stats() reward_stats reward_fn.get_stats()GovernedRunner.get_stats()见 runner.py返回total_rollouts、total_violations、violation_ratePolicyReward.get_stats()见 reward.py返回total_rewards、total_penalties、avg_penalty、violation_rate、clean_rate。最后示例用一句直白的话点明训练的关键洞察Agent 会学到 DROP/DELETE → 负奖励训练之后Agent 将主动回避危险 SQL 操作——这正是训练即治理的最终效果。另一种集成形态Gym 风格的 GovernedEnvironment除了 Runner 形态示例还演示了 demo_environment()把内核包装成标准的 Gymnasium 五元组接口环境。config EnvironmentConfig( max_steps10, violation_penalty-10.0, terminate_on_criticalTrue, ) env GovernedEnvironment(kernel, configconfig) state, info env.reset() state, reward, terminated, truncated, info env.step(action)EnvironmentConfig的全部可调参数见 environment.py参数默认值作用max_steps100每回合最大步数超出后truncatedTrueviolation_penalty-10.0单次违规的基础惩罚critical放大 10 倍-100、high放大 5 倍-50terminate_on_criticalTrue出现 critical 违规立即终止本回合阻止不安全探索继续step_penalty-0.1每步小幅扣分鼓励高效完成任务success_bonus10.0无违规且成功完成时的一次性奖励reset_kernel_stateTrue重置回合时是否同步重置内核状态这种形态可以直接对接 Agent-Lightning Trainer、OpenAI Gym/Gymnasium、Stable Baselines3 等任何遵循step/reset接口的框架。环境内置了与 Runner 一致的违规钩子机制若内核支持on_policy_violation回调则注册钩子否则回退到轮询kernel.get_recent_violations()见 environment.py确保不暴露回调 API 的内核同样能被正确计量违规。环境还会持续累计total_episodes、total_steps、total_violations、success_rate、violations_per_episode、steps_per_episode等指标通过env.get_metrics()随时取用。预期输出解读按照示例文档的 Expected Output运行后会看到类似如下输出SQL Agent Training with Agent-Lightning Agent OS ✓ Kernel initialized with policies ✓ GovernedRunner initialized ✓ PolicyReward function created Episode 1: SELECT * FROM users... Status: ✅ SUCCESS Violations: 0 Reward: 5.85 Episode 3: DROP TABLE users... Status: ❌ BLOCKED Violations: 1 ⚠️ SQLPolicy: Dangerous SQL operation blocked Reward: -100.00 Training Summary: Violation rate: 33.3% Clean rate: 66.7%这组数据印证了几个要点合规动作SELECT/INSERT/UPDATE正常执行并拿到正奖励基础 accuracy 奖励 clean_bonus如 Episode 1 的 5.85危险动作DROP/DELETE被策略阻断状态为BLOCKED、违规计数为 1、奖励被扣到 -100critical 惩罚 无基础奖励6 条查询中 2 条违规因此违规率 33.3%、守规率 66.7%与示例文档完全一致。从示例走向真实训练接入 Agent-Lightning Trainer示例文档强调示例中的 Mock 组件应替换为真实实现。完整接入 Agent-Lightning 的形态在 agent-lightning 包 README 中有规范示范from agent_lightning_gov import GovernedRunner, PolicyReward from agent_os import KernelSpace from agent_os.policies import SQLPolicy, CostControlPolicy # 1. 创建受治理内核 kernel KernelSpace(policy[ SQLPolicy(deny[DROP, DELETE]), CostControlPolicy(max_cost_usd100), ]) # 2. 创建受治理 Runner runner GovernedRunner(kernel) # 3. 创建策略感知奖励函数 def base_accuracy(rollout): return rollout.task_output.accuracy if rollout.success else 0.0 reward_fn PolicyReward(kernel, base_reward_fnbase_accuracy) # 4. 交给 Agent-Lightning Trainer 训练如 GRPO 算法 from agentlightning import Trainer trainer Trainer( runnerrunner, reward_fnreward_fn, algorithmGRPO, ) trainer.train(num_epochs100)在整个训练过程中GovernedRunner还会尝试通过agentlightning.emitter.emit_annotation上报agent_os.violations、agent_os.total_penalty、agent_os.policies_violated等治理跨度见 runner.py让每一次违规都进入可审计的训练轨迹——对应 ADR-0024 中违规记录包含 policy、description、severity、blocked、step、timestamp与训练审计跨度携带agent_os.*前缀属性的约定。若需要更完整的审计导出还可使用FlightRecorderEmitter将记录导出到 LightningStore 或 JSON 文件做事后分析。小结从 训练示例 README 出发本文完整还原了治理内化进奖励信号的实现路径Runner 层GovernedRunner让每一次 rollout 都经过策略内核违规与信号被完整捕获奖励层PolicyReward按严重度折算负向惩罚、对干净执行发放奖励把治理约束变成梯度信号环境层GovernedEnvironment提供 Gym 兼容接口支持关键违规即时终止阻止不安全探索可观测层统计接口与agent_os.*审计跨度让训练过程全程可查、可复盘。与纯运行时拦截相比这种做法的本质区别在于智能体不是被阻止去做危险操作而是被教会危险操作没有价值。这也正是 ADR-0024 所确立的违规惩罚内化于奖励、关键违规终止探索、全量违规留痕审计三大设计原则的工程化落地。示例中以 SQL Agent 为场景但 Runner Reward Environment 的组合对任何存在策略约束的 Agent 训练任务成本治理、内容安全、工具调用白名单等同样适用。【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 Agent OS 治理强化学习训练：基于 Agent-Lightning 打造“天生安全“的 SQL Agent
用 Agent OS 治理强化学习训练基于 Agent-Lightning 打造天生安全的 SQL Agent【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit在 Agent Governance Toolkit 中Agent-Lightning 训练示例 展示了一种关键理念安全不应该在训练完成之后才补课而应该作为奖励信号直接内化进强化学习RL训练过程。本文以示例自带的 SQL Agent 训练脚本 sql_agent.py 为主线讲解如何用GovernedRunner、PolicyReward、GovernedEnvironment三件套把 Agent OS 的策略引擎接入 Agent-Lightning 训练循环让智能体在学会生成准确 SQL 的同时从一开始就绝不触碰DROP/DELETE等危险操作、不超出成本上限。读完本文你将掌握训练即治理的完整落地路径从策略初始化、受治理的 rollout 执行到把策略违规折算为负向奖励、再到训练统计的可观测化。为什么要对 RL 训练本身做治理常规的治理思路是运行时拦截Agent 做出危险动作时由策略引擎在推理/执行阶段将其阻断。但 ADR-0024: RL Training Governance with Violation Penalties 指出了单纯依赖运行时拦截在训练场景下的三个盲区奖励错配如果违规行为能带来更高的任务完成度RL 优化器反而会奖励违规行为让智能体学会绕过治理规避学习智能体可能摸索出规避治理检查的路径而不是真正理解策略约束数据污染训练轨迹中混入不安全的动作序列会污染整个训练数据集。因此 ADR-0024 的决策是把治理下沉到奖励信号本身——每次策略违规都从奖励中扣除惩罚严重程度映射到惩罚量级从而让智能体在梯度更新的过程中自发地学会违规 高代价。对应的核心实现位于 agent-governance-python/agent-lightning 包PyPI 包名为agentmesh-lightning公开预览阶段API 在正式发布前可能调整。该包的角色定位是Agent-Lightning 负责训练与优化大脑Agent OS 负责治理与安全护栏两者结合得到既聪明又安全的 Agent。整体架构三个组件各司其职agent_lightning_gov模块见init.py对外暴露四个核心构件其中训练示例用到三个组件职责源码位置GovernedRunner用内核KernelSpace包裹 Agent 执行收集违规记录兼容 Agent-Lightning Trainerrunner.pyPolicyReward把策略违规折算成负向 RL 奖励包装任意基础奖励函数reward.pyGovernedEnvironmentGym/Gymnasium 风格的训练环境逐步执行动作、施加违规惩罚、支持关键违规终止environment.pyFlightRecorderEmitter把训练审计跨度span导出到 LightningStore 或文件emitter.py示例文档将工作流程归纳为四步这正是本文要展开的骨架GovernedRunner用策略检查包裹 Agent 执行PolicyReward把违规转换为负向 RL 奖励Agent 在训练中学会避开策略违规最终产出从第一天起就安全的 Agent。环境准备按照示例文档的 README 要求安装两个依赖pip install agent-os-kernel agentlightning其中agent-os-kernel提供策略内核KernelSpace 与 SQLPolicy 等策略类agentlightning提供 RL 训练框架。若想直接使用治理集成包也可以安装agentmesh-lightning并从agent_lightning_gov导入治理组件该包从agent_os.integrations.agent_lightning抽取而来旧导入路径仍通过兼容垫片可用新代码应直接使用agent_lightning_gov。运行示例python sql_agent.py逐行拆解 sql_agent.py一个受治理的训练循环sql_agent.py 是完整的可运行示例。虽然其中MockKernelSpace、MockSQLPolicy、MockCostControlPolicy是为了演示而简化的替身文件头部注释明确写着 MOCK COMPONENTS (Replace with real implementations)但它完整呈现了与真实内核一致的调用契约可以直接替换为 Agent OS 的真实实现。第 1 步初始化带策略的内核kernel MockKernelSpace(policy[ MockSQLPolicy( allow[SELECT, INSERT, UPDATE], deny[DROP, DELETE, TRUNCATE], ), MockCostControlPolicy(max_cost_usd100), ])这里定义了本示例的两条治理规则SQLPolicy允许SELECT/INSERT/UPDATE拒绝DROP/DELETE/TRUNCATE——即只读为主、危险 DDL/DML 一律阻断CostControlPolicy单次查询成本上限 100 美元即成本治理。从真实实现的文档字符串看与GovernedRunner配套的典型内核构造方式是from agent_os import KernelSpace from agent_os.policies import SQLPolicy, CostControlPolicy kernel KernelSpace(policy[ SQLPolicy(deny[DROP, DELETE]), CostControlPolicy(max_cost_usd100), ])内核之所以是关键是因为 GovernedRunner.step() 会优先调用kernel.execute_async()回退到kernel.execute()把每个动作都送进策略引擎做确定性评估——这正是 ADR-0024 中确定性动作评估的设计延续。第 2 步创建 GovernedRunnerrunner GovernedRunner( kernel, fail_on_violationFalse, log_violationsTrue, )构造函数签名见 runner.py支持三个关键参数参数默认值作用kernel必填已加载策略的 Agent OS KernelSpacefail_on_violationFalse为True时违规且被阻断直接抛PolicyViolationError训练终止False则继续训练、仅施加惩罚log_violationsTrue是否将每次违规写入日志violation_callbackNone每次违规触发的可选回调可用于外部审计接入随后示例执行runner.init(MockAgent()) runner.init_worker(0, None)init()负责挂载内核钩子如果内核暴露on_policy_violation与on_signalRunner 会注册_handle_violation和_handle_signal回调从而把内核在执行期间的违规/信号实时捕获进 rollout 记录。第 3 步创建策略感知的奖励函数def accuracy_reward(rollout): if rollout.success and rollout.task_output: return rollout.task_output.get(accuracy, 0.0) return 0.0 reward_fn PolicyReward(kernel, base_reward_fnaccuracy_reward)PolicyReward的设计是包装任意基础奖励函数见 reward.pybase_reward_fn负责奖励任务本身的质量这里用accuracy字段衡量 SQL 生成准确度PolicyReward在它之上叠加策略惩罚。最终奖励计算公式为final_reward base_reward penalty(违规惩罚恒为负值)若启用multiplicative模式则改为final_reward base_reward * multiplicative_factor默认 0.5。此外还有两个边界保护min_reward默认 -100与max_reward默认 100把奖励钳制在合理区间防止极端负值破坏训练稳定性。默认的惩罚配置RewardConfig与PolicyViolation的严重度映射完全对齐见 runner.py 与 reward.py严重度违规记录惩罚PolicyViolationRL 奖励惩罚RewardConfigcritical100.0-100.0high50.0-50.0medium10.0-10.0low1.0-1.0一个值得注意的细节PolicyViolation的penalty字段在构造时由严重度表自动推导但调用方显式传入penalty时以调用方为准——这一设计允许对个别违规单独加权避免覆盖调用方意图见 runner.py 的注释。PolicyReward还内置了干净执行奖励当一次 rollout 没有任何违规时在最终奖励上追加clean_bonus默认 5.0形成违规扣分、守规加分的双向激励。第 4 步模拟训练回合示例用 6 条代表性查询模拟了训练过程test_queries [ SELECT * FROM users WHERE id 1, INSERT INTO logs (msg) VALUES (hello), DROP TABLE users, # Should be blocked! UPDATE users SET name John WHERE id 1, DELETE FROM users WHERE id 1, # Should be blocked! SELECT COUNT(*) FROM orders, ]每回合的核心循环只有三行rollout await runner.step(query) # 受治理的执行 reward reward_fn(rollout, emitFalse) # 计算带惩罚的奖励 violations_count len(rollout.violations) # 统计违规GovernedRunner.step()返回GovernedRollout见 runner.py其中携带success、violationsPolicyViolation列表、signals_sent、total_penalty各违规惩罚之和与execution_time_ms等治理元数据。这里有一个值得关注的并发设计step()通过contextvars.ContextVar把每次调用的违规/信号缓冲绑定到当前 asyncio 上下文保证同一个 Runner 上并发执行多个step()时各 rollout 的违规记录不会互相串扰见 runner.py。PolicyReward.__call__在emitTrue默认时还会通过agentlightning.emitter.emit_reward上报多维奖励final/base/policy_penalty并附带agent_os.violation_count、agent_os.policy_compliant属性供训练可视化与审计使用。第 5 步输出训练统计训练结束后示例分别拉取两套统计stats runner.get_stats() reward_stats reward_fn.get_stats()GovernedRunner.get_stats()见 runner.py返回total_rollouts、total_violations、violation_ratePolicyReward.get_stats()见 reward.py返回total_rewards、total_penalties、avg_penalty、violation_rate、clean_rate。最后示例用一句直白的话点明训练的关键洞察Agent 会学到 DROP/DELETE → 负奖励训练之后Agent 将主动回避危险 SQL 操作——这正是训练即治理的最终效果。另一种集成形态Gym 风格的 GovernedEnvironment除了 Runner 形态示例还演示了 demo_environment()把内核包装成标准的 Gymnasium 五元组接口环境。config EnvironmentConfig( max_steps10, violation_penalty-10.0, terminate_on_criticalTrue, ) env GovernedEnvironment(kernel, configconfig) state, info env.reset() state, reward, terminated, truncated, info env.step(action)EnvironmentConfig的全部可调参数见 environment.py参数默认值作用max_steps100每回合最大步数超出后truncatedTrueviolation_penalty-10.0单次违规的基础惩罚critical放大 10 倍-100、high放大 5 倍-50terminate_on_criticalTrue出现 critical 违规立即终止本回合阻止不安全探索继续step_penalty-0.1每步小幅扣分鼓励高效完成任务success_bonus10.0无违规且成功完成时的一次性奖励reset_kernel_stateTrue重置回合时是否同步重置内核状态这种形态可以直接对接 Agent-Lightning Trainer、OpenAI Gym/Gymnasium、Stable Baselines3 等任何遵循step/reset接口的框架。环境内置了与 Runner 一致的违规钩子机制若内核支持on_policy_violation回调则注册钩子否则回退到轮询kernel.get_recent_violations()见 environment.py确保不暴露回调 API 的内核同样能被正确计量违规。环境还会持续累计total_episodes、total_steps、total_violations、success_rate、violations_per_episode、steps_per_episode等指标通过env.get_metrics()随时取用。预期输出解读按照示例文档的 Expected Output运行后会看到类似如下输出SQL Agent Training with Agent-Lightning Agent OS ✓ Kernel initialized with policies ✓ GovernedRunner initialized ✓ PolicyReward function created Episode 1: SELECT * FROM users... Status: ✅ SUCCESS Violations: 0 Reward: 5.85 Episode 3: DROP TABLE users... Status: ❌ BLOCKED Violations: 1 ⚠️ SQLPolicy: Dangerous SQL operation blocked Reward: -100.00 Training Summary: Violation rate: 33.3% Clean rate: 66.7%这组数据印证了几个要点合规动作SELECT/INSERT/UPDATE正常执行并拿到正奖励基础 accuracy 奖励 clean_bonus如 Episode 1 的 5.85危险动作DROP/DELETE被策略阻断状态为BLOCKED、违规计数为 1、奖励被扣到 -100critical 惩罚 无基础奖励6 条查询中 2 条违规因此违规率 33.3%、守规率 66.7%与示例文档完全一致。从示例走向真实训练接入 Agent-Lightning Trainer示例文档强调示例中的 Mock 组件应替换为真实实现。完整接入 Agent-Lightning 的形态在 agent-lightning 包 README 中有规范示范from agent_lightning_gov import GovernedRunner, PolicyReward from agent_os import KernelSpace from agent_os.policies import SQLPolicy, CostControlPolicy # 1. 创建受治理内核 kernel KernelSpace(policy[ SQLPolicy(deny[DROP, DELETE]), CostControlPolicy(max_cost_usd100), ]) # 2. 创建受治理 Runner runner GovernedRunner(kernel) # 3. 创建策略感知奖励函数 def base_accuracy(rollout): return rollout.task_output.accuracy if rollout.success else 0.0 reward_fn PolicyReward(kernel, base_reward_fnbase_accuracy) # 4. 交给 Agent-Lightning Trainer 训练如 GRPO 算法 from agentlightning import Trainer trainer Trainer( runnerrunner, reward_fnreward_fn, algorithmGRPO, ) trainer.train(num_epochs100)在整个训练过程中GovernedRunner还会尝试通过agentlightning.emitter.emit_annotation上报agent_os.violations、agent_os.total_penalty、agent_os.policies_violated等治理跨度见 runner.py让每一次违规都进入可审计的训练轨迹——对应 ADR-0024 中违规记录包含 policy、description、severity、blocked、step、timestamp与训练审计跨度携带agent_os.*前缀属性的约定。若需要更完整的审计导出还可使用FlightRecorderEmitter将记录导出到 LightningStore 或 JSON 文件做事后分析。小结从 训练示例 README 出发本文完整还原了治理内化进奖励信号的实现路径Runner 层GovernedRunner让每一次 rollout 都经过策略内核违规与信号被完整捕获奖励层PolicyReward按严重度折算负向惩罚、对干净执行发放奖励把治理约束变成梯度信号环境层GovernedEnvironment提供 Gym 兼容接口支持关键违规即时终止阻止不安全探索可观测层统计接口与agent_os.*审计跨度让训练过程全程可查、可复盘。与纯运行时拦截相比这种做法的本质区别在于智能体不是被阻止去做危险操作而是被教会危险操作没有价值。这也正是 ADR-0024 所确立的违规惩罚内化于奖励、关键违规终止探索、全量违规留痕审计三大设计原则的工程化落地。示例中以 SQL Agent 为场景但 Runner Reward Environment 的组合对任何存在策略约束的 Agent 训练任务成本治理、内容安全、工具调用白名单等同样适用。【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考