基于意图的授权（Intent-Based Authorization）实战：在 agent-os 中声明计划、检测漂移并治理多 Agent 行为

基于意图的授权Intent-Based Authorization实战在 agent-os 中声明计划、检测漂移并治理多 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 将做什么再让它去做一旦偏离声明系统自动察觉。Intent-Based Authorization基于意图的授权是 AI Agent 治理领域的一种前置控制范式它作为一个治理层横亘在 Agent 与其动作之间。在执行开始之前Agent 必须先声明一份计划declare_intent由人工审查者或策略门禁批准approve_intent随后每一个动作在执行前都要与计划比对check_action会话结束时系统将计划 vs 实际进行核对verify_intent并暴露任何偏差drift。本教程基于 agent-osAgent Governance Toolkit 的 Python 运行时内核核心实现 与配套 示例脚本带你完整走通声明—批准—执行—校验生命周期掌握三种漂移策略的取舍、多 Agent 场景下的子意图作用域收窄以及如何把意图检查接入无状态内核StatelessKernel进行生产化落地。本文覆盖四个主题核心生命周期Declare → Approve → Execute → Verify漂移检测策略SOFT_BLOCKvsHARD_BLOCKvsRE_DECLARE多 Agent 场景下的子意图作用域收窄Child Intent Scope Narrowing运行完整 Demo 与源码级原理剖析前置条件pip install agent-os-kernel需要 Python 3.9 及以上版本。开发与测试阶段可以使用进程内后端MemoryBackend生产环境请替换为RedisBackend以便在多个 Agent 副本之间共享意图状态详见文末生产化部署一节。1. 核心生命周期声明、批准、执行、校验Intent-Based Authorization 的四步生命周期可以概括为一张图DECLARED --approve-- APPROVED --first action-- EXECUTING | ---------------- v v COMPLETED VIOLATED (no drift) (drift found) Any state --ttl expired-- EXPIRED下面逐步用代码走通整条链路。Step 1声明意图Declare IntentAgent 先宣布自己的行动计划此时没有任何动作真正执行import asyncio from agent_os.intent import DriftPolicy, IntentAction, IntentManager from agent_os.stateless import MemoryBackend async def main(): manager IntentManager(backendMemoryBackend()) intent await manager.declare_intent( agent_idpayment-agent, planned_actions[ IntentAction(actionread_balance), IntentAction(actiontransfer_funds, params_schema{max_amount: 1000}), ], drift_policyDriftPolicy.SOFT_BLOCK, ttl_seconds300, ) print(fIntent ID: {intent.intent_id}) print(fState: {intent.state.value}) # declared print(fActions: {intent.planned_action_names})参数说明IntentAction(action, params_schemaNone)声明一个计划动作。action是动作名如read_balance、transfer_fundsparams_schema是可选的参数约束字典键为参数名、值为允许的值或模式动作必须满足约束才视为命中计划。ttl_seconds设置意图的过期时间。若批准或执行发生在过期之后意图会被拒绝。drift_policy声明漂移处理策略默认DriftPolicy.SOFT_BLOCK下文第 2 节详解。从源码看declare_intent会生成形如intent:12位hex的唯一 ID并根据ttl_seconds计算expires_at如果提供了parent_intent_id还会先执行子意图作用域校验见第 3 节。存储层写入时使用的后端 TTL 是声明 TTL 的两倍ttl_seconds * 2为批准与执行留出缓冲窗口。Step 2批准意图Approve Intent批准将意图从declared推进到approved表示人工审查者或自动化策略门禁已经签字放行intent await manager.approve_intent(intent.intent_id) print(fState: {intent.state.value}) # approved在生产环境中这一步通常会接入 human-in-the-loop 审批队列或外部策略引擎。底层实现会记录approved_at时间戳供后续校验计算会话时长并拒绝批准已过期或非declared状态的意图抛出IntentStateError。Step 3运行时检查动作Check Actions在每个动作真正执行前调用check_action。管理器会记录检查结果并在第一次调用时把意图推进到executing状态# Planned action - allowed check await manager.check_action( intent.intent_id, read_balance, {}, payment-agent, req-001, ) print(fread_balance: {ALLOWED if check.allowed else BLOCKED}) print(f was_planned: {check.was_planned}) # Unplanned action - drift detected check await manager.check_action( intent.intent_id, delete_account, {}, payment-agent, req-002, ) print(fdelete_account: {ALLOWED if check.allowed else BLOCKED}) if check.drift_policy_applied: print(f policy: {check.drift_policy_applied.value}) print(f penalty: -{check.trust_penalty} trust points)check_action的签名是check_action(intent_id, action, params, agent_id, request_id)params本次动作的实际参数会与计划动作的params_schema比对agent_id发起动作的 Agent 标识request_id请求关联 ID用于跨系统追踪。IntentCheckResult字段说明字段类型描述allowedbool动作是否允许执行was_plannedbool动作是否在声明计划内drift_policy_appliedDriftPolicy \| None检测到漂移时触发的策略trust_penaltyfloat信任分扣减默认 50.0reasonstr人类可读的原因说明重要的安全细节check_action会校验agent_id与意图声明者是否一致。源码中明确注释——如果没有这个检查一个恶意或被攻陷的 Agent 只要拿到别人的intent_id就能搭便车借对方的批准去执行高危动作跨 Agent 意图复用Cross-agent intent reuse。这种场景会被直接拒绝对应的测试用例test_check_action_rejects_cross_agent_intent_reuse覆盖了所有者可执行、攻击者被拒绝的正反两条路径。另外每次动作执行时参数会通过_hash_params生成一个确定性的 SHA-256 前 16 位哈希params_hash随执行记录一并落库供审计时比对实际调用参数。Step 4校验意图Verify Intentverify_intent关闭会话产出一份计划 vs 实际的结构化审计报告verification await manager.verify_intent(intent.intent_id) print(fFinal state: {verification.state.value}) # violated or completed print(fPlanned: {verification.planned_actions}) print(fExecuted: {verification.executed_actions}) print(fUnplanned: {verification.unplanned_actions}) print(fMissed: {verification.missed_actions}) print(fDrift events: {verification.total_drift_events}) print(fTrust penalty: {verification.total_trust_penalty}) asyncio.run(main())没有漂移事件时意图进入completed检测到漂移则进入violated。IntentVerification字段说明字段描述planned_actions执行前声明的动作列表executed_actions实际运行且成功的动作列表unplanned_actions执行了但不在计划内的动作missed_actions计划了但从未执行的动作total_drift_events漂移检测次数total_trust_penalty累计扣减的信任分duration_seconds从批准到校验的秒数一个容易踩坑的约束源码规定verify_intent只能在executing状态调用。如果意图还停留在approved一次check_action都没调用过校验会抛出IntentStateError——因为此时没有任何执行记录可以拿来与计划比对。测试test_cannot_verify_completed_intent、test_cannot_check_unapproved_intent都验证了这类非法状态转换被拦截。2. 漂移检测策略DriftPolicy枚举共有三个取值在declare_intent时设定。它决定了计划外动作出现时系统如何响应。SOFT_BLOCK默认计划外动作允许放行但会被标记。系统施加信任分惩罚并记录一条DriftEvent。适用于连续性优先于严格管控的场景——比如支付链路中途不能因为一个通知动作而整体中断intent await manager.declare_intent( agent_idpayment-agent, planned_actions[IntentAction(actionread_balance)], drift_policyDriftPolicy.SOFT_BLOCK, ) intent await manager.approve_intent(intent.intent_id) check await manager.check_action( intent.intent_id, send_notification, {}, payment-agent, req-003 ) print(check.allowed) # True - action proceeds print(check.trust_penalty) # 50.0 - penalty recordedHARD_BLOCK计划外动作直接拒绝。适用于合规关键型 Agent——任何偏离都不可接受如金融报告、医疗数据访问from agent_os.intent import DriftPolicy, IntentAction, IntentManager from agent_os.stateless import MemoryBackend async def hard_block_demo(): manager IntentManager(backendMemoryBackend()) intent await manager.declare_intent( agent_idcompliance-agent, planned_actions[IntentAction(actiongenerate_report)], drift_policyDriftPolicy.HARD_BLOCK, ) intent await manager.approve_intent(intent.intent_id) # Planned - allowed check await manager.check_action( intent.intent_id, generate_report, {}, compliance-agent, req-010 ) print(fgenerate_report: {ALLOWED if check.allowed else BLOCKED}) # ALLOWED # Unplanned - blocked check await manager.check_action( intent.intent_id, send_email, {}, compliance-agent, req-011 ) print(fsend_email: {ALLOWED if check.allowed else BLOCKED}) # BLOCKEDRE_DECLARE计划外动作被拒绝Agent 必须重新声明一个新意图才能继续。适用于作用域变化需要完整复审周期的场景intent await manager.declare_intent( agent_idmy-agent, planned_actions[IntentAction(actionread_config)], drift_policyDriftPolicy.RE_DECLARE, )从实现上看三种策略在check_action中的落地路径是清晰的命中计划 → 记录outcomeexecuted放行未命中计划 → 按策略分别记allowed_with_penalty放行扣分、blocked拒绝、re_declare_required拒绝并要求重新声明同时生成DriftEvent和ExecutionRecord双份结构化审计记录。信任分惩罚值默认 50.0可通过IntentManager(backend..., drift_penalty...)覆盖。3. 子意图作用域收窄多 Agent 编排在多 Agent 编排中编排者orchestrator声明一个宽泛的意图再把子任务委托给专精 Agent。子意图必须是父意图计划动作的子集绝不能扩大作用域from agent_os.intent import IntentAction, IntentManager, IntentScopeError from agent_os.stateless import MemoryBackend async def multi_agent_demo(): manager IntentManager(backendMemoryBackend()) # Orchestrator declares the full scope parent await manager.declare_intent( agent_idorchestrator, planned_actions[ IntentAction(actionread_balance), IntentAction(actiontransfer_funds), IntentAction(actiongenerate_report), ], ) parent await manager.approve_intent(parent.intent_id) print(fParent scope: {parent.planned_action_names}) # Sub-agent gets only the actions it needs child await manager.declare_intent( agent_idreport-agent, planned_actions[IntentAction(actiongenerate_report)], parent_intent_idparent.intent_id, ) print(fChild scope: {child.planned_action_names}) # {generate_report} # A rogue agent trying to exceed parent scope is rejected try: await manager.declare_intent( agent_idrogue-agent, planned_actions[IntentAction(actiondelete_everything)], parent_intent_idparent.intent_id, ) except IntentScopeError as e: print(fScope violation blocked: {e})关键安全属性IntentScopeError在declare_intent阶段、意图落库之前就被抛出。源码中的校验逻辑先加载父意图、计算父作用域集合再用子动作集合做差集只要存在超出父作用域的动作excess非空立即抛异常。因此不存在子意图带着超范围动作短暂存在的窗口期。也可以使用便捷方法create_child_intent它默认继承父意图的漂移策略child await manager.create_child_intent( parent_intent_idparent.intent_id, agent_idreport-agent, planned_actions[IntentAction(actiongenerate_report)], )create_child_intent支持通过drift_policy参数覆盖继承策略传入None时继承父策略。测试套件test_child_inherits_drift_policy与test_child_can_override_drift_policy分别验证了继承与覆盖两条路径。4. 运行完整 Demo仓库附带了一个完整可运行的演示脚本 examples/intent-auth/intent_auth_demo.py它把上面所有环节串成一次端到端演示pip install agent-os-kernel python examples/intent-auth/intent_auth_demo.py预期输出 Intent-Based Authorization Demo --- Step 1: Declare Intent --- Intent ID: intent:... State: declared Actions: {read_balance, transfer_funds} --- Step 2: Approve Intent --- State: approved --- Step 3: Execute Actions --- read_balance: ALLOWED (planned) delete_account: ALLOWED (DRIFT!) policy: soft_block penalty: -50.0 trust points --- Step 4: Verify Intent --- Final state: violated Planned: [read_balance, transfer_funds] Executed: [read_balance, delete_account] Unplanned: [delete_account] Missed: [transfer_funds] Drift events: 1 Trust penalty: 50.0 ...脚本还演示了HARD_BLOCKgenerate_report放行、send_email被拒以及子意图作用域收窄report-agent获得合法的子集、rogue-agent试图执行delete_everything被IntentScopeError拦截。可以对照脚本 intent_auth_demo.py 的源码逐步跟踪。5. 源码级原理IntentManager 如何保证正确性5.1 显式状态机与非法转换拦截agent_os.intent中定义了一张显式的状态转换表_TRANSITIONSdeclared→approved/expiredapproved→executing/expiredexecuting→completed/violated/expiredcompleted、violated、expired是终态一旦进入不可再迁移。所有迁移都经过_transition校验非法转换如重复批准、对未批准意图执行动作、对已完成意图再次校验统一抛出IntentStateError对应的测试覆盖见 test_intent.py 的TestStateMachine类。5.2 乐观并发与多实例一致性ExecutionIntent携带version字段每次保存时递增。IntentManager在写入前用expected_version做乐观并发检查一旦发现后端已有版本与期望版本不符就抛出IntentVersionConflict——这保证了多个IntentManager实例可以共享同一个后端而不会互相覆盖。加固测试 test_intent_hardened.py 中的TestVersionConflict类专门验证了陈旧批准引发版本冲突与并发check_action不丢记录两个场景。5.3 与 StatelessKernel 的集成意图检查并不是孤岛StatelessKernel可以接收一个intent_manager并把ExecutionContext.intent_id透传进每次执行。集成测试TestKernelIntegration验证了计划内动作经内核执行后成功元数据intent_drift为NoneSOFT_BLOCK漂移时动作仍成功但元数据带intent_drift: True与trust_penalty: 50.0HARD_BLOCK漂移时内核返回successFalse且信号为SIGKILL策略检查先于意图检查动作被策略如read_only拒绝时不会产生任何意图执行记录test_policy_denied_before_intent_check。这与 stateless 架构的定位一致内核不保存会话状态所有上下文随请求传递stateless.py 的模块说明明确每请求携带 ExecutionContext含 Agent 身份、策略列表与历史意图状态则统一放在可插拔的StateBackend中。5.4 关键类速查类 / 函数用途IntentManager(backend)主入口所有方法均为异步构造参数drift_penalty可调IntentAction(action, params_schema)声明一个计划动作DriftPolicy枚举SOFT_BLOCK、HARD_BLOCK、RE_DECLAREIntentCheckResultcheck_action的返回值IntentVerificationverify_intent的返回值IntentScopeError子意图超出父作用域时抛出ExecutionRecord/DriftEvent结构化审计记录动作、结果、策略、时间戳、参数哈希MemoryBackend进程内后端用于开发与测试6. 生产化部署从 MemoryBackend 到 RedisBackend教程开头的所有示例都使用MemoryBackend它适合开发与测试但状态只存在于当前进程。生产环境建议替换为 stateless.py 中提供的RedisBackend它带连接池、可配置超时与可选RedisConfig支持在多个 Agent 副本之间共享意图状态。因为意图数据以 JSON 序列化并带命名空间前缀默认agent-os:存储在外部后端任意无状态内核实例都能加载任意请求对应的意图天然支持水平扩展N 个副本 负载均衡无需粘性会话。值得强调的是StateBackend采用typing.Protocol结构子类型而非抽象基类——只要对象实现了get/set/delete方法就满足契约因此可以低成本适配 DynamoDB、Cosmos DB 等第三方客户端作为后端。7. 下一步5 分钟快速上手 — 搭建你的第一个受治理 Agent30 分钟深入剖析 — 信任评分、策略引擎与执行收据receipts自定义工具教程 — 注册运行时受意图检查约束的工具意图即契约让 Agent 在动手前亮明计划让每一次偏离都留下可审计的痕迹这正是 Intent-Based Authorization 在零信任 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），仅供参考