LiteLLM Proxy 接入 ACS 护栏钩子用 Agent Control Specification 为 OpenAI 兼容流量构建策略执行层【免费下载链接】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-toolkitAgent Control SpecificationACS为自主 Agent 的完整循环Input → Model → Tool Call → Tool Result → Output提供无状态、确定性、失败即拒绝fail-closed的策略决策运行时。当你的 Agent 流量经由 LiteLLM Proxy 统一路由到各模型提供商时ACS 提供了一对官方集成入口AgentControlLiteLLMGuardrail通过guardrails:YAML 零代码注册的 guardrail hook与guard_litellm_proxy()ASGI 中间件形态的进程内接入。本文基于 LiteLLM Proxy guardrail hook 文档 展开结合 Python SDK 适配器源码 与 测试用例讲清安装方式、Proxy YAML 配置、hook 到干预点的映射、会话关联、流式处理策略与边界限制让你能在不侵入上游代码的情况下把 ACS 的 8 类干预点能力挂到 LiteLLM Proxy 上。ACS 集成模型宿主代码与无状态运行时分离理解该集成前需要先抓住 ACS 的设计原则运行时保持无状态stateless宿主在每一个干预点提交一份完整快照snapshot并收取归一化裁决verdict。LiteLLM Proxy guardrail hook 正是这类宿主集成代码——它负责把 LiteLLM 的生命周期事件翻译成 ACS 干预点调用而裁决逻辑Rego 策略、annotator 等全部由 ACS 运行时完成。从 Python SDK 结构 可以看到AgentControlLiteLLMGuardrail和LiteLLMProxyMiddleware均通过agent_control_specification._adapters导出且对 LiteLLM 采用可选依赖设计适配器模块通过try/except ImportError包裹litellm.integrations.custom_guardrail的导入未安装 LiteLLM 时 SDK 仍可正常导入_LiteLLMCustomGuardrail退化为普通object基类保证核心库零依赖可用见 _adapters/litellm.py。Proxy 进程通过AgentControl.from_path加载 manifest策略清单文件例如AgentControl.from_path(/etc/acs/manifest.yaml)。Rego 策略在opa位于PATH时使用内置的 OPA dispatcher 执行如果 manifest 声明了 annotators则通过 Python SDK 默认 dispatcher 运行或者在自定义构造路径下由应用代码传入 dispatcher 运行——这条链路与 README 中描述的构造约定 一致默认 wheel 未启用bundled-dispatchersCargo feature因此含 annotators 的 manifest 必须显式提供annotator_dispatcher否则构造会直接失败。安装与依赖LiteLLM Proxy 集成是 ACS Python SDK 的可选 extra安装命令pip install agent-control-specification[litellm-proxy]从 pyproject.toml 可以确认该 extra 的实际依赖构成litellm-proxy [ litellm[proxy]1.40, fastapi0.100, ]注意两点一是必须安装litellm[proxy]而非裸litellm——proxy extra 会引入运行时所需的代理服务依赖README 中对此有明确警告二是 SDK 本身要求 Python 3.11核心策略引擎由 Rust 实现并通过 maturin 构建为 CPython 3.11 ABI3 wheel安装官方 wheel 不需要本地 Rust 工具链。Proxy YAML零代码注册 guardrailLiteLLM Proxy 的guardrails:配置段允许通过 YAML 直接注册 ACS 护栏无需编写任何 Python 代码。以下是最小可用配置model_list: - model_name: gpt-4o-mini litellm_params: model: openai/gpt-4o-mini api_key: os.environ/OPENAI_API_KEY guardrails: - guardrail_name: acs litellm_params: guardrail: agent_control_specification.AgentControlLiteLLMGuardrail mode: [pre_call, post_call] manifest_path: /etc/acs/manifest.yaml default_on: true streaming: buffer reject_unknown_tool_results: true session_cache_size: 512 session_ttl_seconds: 1800各参数在 AgentControlLiteLLMGuardrail 构造器 中均有对应实现含义如下参数默认值说明guardrail必填指向agent_control_specification.AgentControlLiteLLMGuardrail类LiteLLM 会反射实例化mode[pre_call, post_call]注册的 LiteLLM 事件钩子集合对应GuardrailEventHooks.pre_call与post_callmanifest_path无ACS manifest 文件路径惰性加载首次求值前通过AgentControl.from_path构建 control见_control()方法default_ontrue是否默认对全部请求启用该 guardrailstreamingbuffer流式处理策略可选buffer/fail_closed/evaluate_only见下文专节reject_unknown_tool_resultstrue对无法关联到已知 tool_call 的 tool 结果失败即拒绝session_cache_size512每实例会话缓存的最大条目数LRU 上限session_ttl_seconds1800会话空闲 TTL超时条目被回收0 表示不启用 TTL 清理构造器对入参做了严格校验control与manifest_path必须二选一同时传或都不传都会抛ValueErrorstreaming仅接受三个枚举值之一。session_cache_size与session_ttl_seconds经max(1, ...)与max(0.0, ...)归一化后用于构建_LiteLLMSessionCache。Hook 到干预点的映射guardrail 的核心价值在于把 LiteLLM 的钩子事件翻译成 ACS 干预点求值。完整映射关系如下LiteLLM hookACS 干预点快照字段async_pre_call_hook末尾消息roleuserinputinput,metadata,transportasync_pre_call_hook每个转发请求pre_model_callmodel_request,metadata,transportasync_post_call_success_hookpost_model_callmodel_request,model_response,metadata,transportasync_post_call_success_hook含 assistanttool_callspre_tool_calltool_call,model_response,metadata,transport下一次async_pre_call_hook末尾消息roletoolpost_tool_calltool_call,tool_result,metadata,transportasync_post_call_success_hook无 tool callsoutputoutput,model_request,model_response,metadata,transportasync_post_call_streaming_iterator_hook缓冲后的post_model_call/pre_tool_call/output组装后的完整响应对照源码可以还原这条求值链路_adapters/litellm.pyasync_pre_call_hook先读取请求messages列表末尾消息的角色user时先求值INPUT快照携带input为用户消息内容tool时求值POST_TOOL_CALL把上一轮记录的tool_call_id → 工具名关联起来快照携带tool_call与tool_result随后每个请求都会求值PRE_MODEL_CALL快照携带完整model_request。若末尾是 assistant 消息且启用了reject_unknown_tool_results会直接以acs_litellm_terminal_assistant原因拦截——因为 ACS 无法把这种请求映射到 input 或 post_tool_call。async_post_call_success_hook先求值POST_MODEL_CALL同时携带model_request与model_response若响应含 assistanttool_calls则对每个工具调用求值PRE_TOOL_CALL快照携带tool_call其args会把 JSON 字符串参数反序列化为对象并把tool_call_id → name记入会话缓存若无 tool calls则求值OUTPUT。转换transform落地ENFORCE模式下若裁决带转换目标源码会就地改写请求/响应对象——改写用户消息内容、替换整个model_request字典、重写 tool call 参数_set_tool_call_args会用紧凑 JSON 序列化、重写 assistant 内容等然后返回修改后的对象给 LiteLLM 继续转发。测试用例 验证了这些映射test_pre_call_maps_user_input_then_pre_model_and_applies_transforms断言依次产生INPUT → PRE_MODEL_CALL两次求值且请求被改写test_post_call_maps_post_model_pre_tool_and_records_correlation断言POST_MODEL_CALL → PRE_TOOL_CALL顺序及 tool 参数被脱敏为{query:redacted}。会话关联与有界缓存LiteLLM 会把 assistant 的 tool call 与后续的 tool 结果拆分到不同的 HTTP 请求中因此 hook 需要在进程内维护一个有界缓存把模型下发的tool_call.id映射到工具名。从源码结构看_adapters/litellm.py该缓存由_LiteLLMSessionCache实现每实例一份具备三个特性LRU 驱逐缓存达session_cache_size上限时淘汰最久未使用的条目空闲 TTL 清理每次访问前按session_ttl_seconds检查并移除超时条目ttl_seconds为 0 时禁用每会话互斥锁locked(sid)上下文管理器以会话 id 为粒度串行化 hook 体执行防止并发请求交错破坏 tool call 关联状态。设计上刻意保持的边界是该缓存只是适配器状态不属于 ACS 运行时也不会在快照中伪造合成tool_call.id——快照里的 id 全部来自模型真实下发。会话 id 解析遵循固定优先级对应源码_litellm_session_id见 _adapters/litellm.pymetadata.agent_control_session_idmetadata.acs_session_idmetadata.litellm_session_id顶层litellm_session_id顶层user若以上均不存在hook 为该次调用生成ephemeral:前缀的临时 id——这意味着后续的 tool 结果无法关联到先前的 tool call在强制模式下会失败即拒绝fail closed。测试test_unknown_tool_result_fails_closed_before_pre_model正是验证了这一行为伪造tool_call_id的 tool 消息在求值 pre_model 之前即被拦截。此外请求失败时async_post_call_failure_hook会主动drop该会话条目避免脏状态残留。流式处理策略buffer / fail_closed / evaluate_only流式响应是代理场景最棘手的部分——策略必须等完整响应才能裁决但客户端期望边生成边接收。hook 通过streaming参数提供三种模式源码实现见async_post_call_streaming_iterator_hook_adapters/litellm.pybuffer默认排干 LiteLLM 流用_assemble_litellm_chunks把各分片按delta.content拼接、按tool_calls索引聚合重建完整 assistant 响应再执行 ACS 求值。裁决允许时按原始分片逐块回放响应被转换时只回放一个替换分片_stream_replacement_chunk把转换后的完整内容封装为单个chat.completion.chunk强制模式下被拒绝时在回放任何缓冲分片之前直接抛错客户端不会看到被截断的流。fail_closed强制模式下直接拒绝流式请求抛出acs_litellm_streaming_unsupported适用于不允许流式弱化的严格环境。evaluate_only照常缓冲并求值但总是回放原始分片不执行流上转换的强制落地。该模式只建议用于审计或灰度rollout因为它对流的输出不做转换强制。测试用例对这三种行为均有覆盖test_streaming_buffer_evaluates_complete_response_before_replay验证缓冲模式把Hel lo组装成完整的Hello再求值并回放两个分片test_streaming_transform_emits_replacement_chunk验证转换时只产生一个内容为clean的替换分片test_streaming_evaluate_only_uses_per_call_mode_without_disabling_enforcement则验证 evaluate_only 仅作用于流式路径非流式调用仍保持强制模式。需要区分的是hook 的流式处理与guard_litellm_proxy()中间件的流式行为是两套实现中间件在 ASGI 层对 SSE 流做assemble_sse_stream→ 求值 →synthesize_sse_stream重建且只对/chat/completions与/v1/chat/completions路径做流式护栏其他 schema 的流式请求直接抛AdapterUnsupportedErrorfail closed避免错误重建响应见 _adapters/litellm.py。拒绝与转换的落地语义无论哪种接入方式强制模式的裁决语义保持一致对应_evaluate与_has_transform[_adapters/litellm.py](https://link.gitcode.com/i/93e7a5942b4bdb781f78123f420703d0#L514-L522, L596-L599)denyAgentControlBlocked被转换为HTTPException(400)响应体为{error: {type: acs_guardrail_block, code: 原因, message: 消息}}code优先取裁决的reason否则使用acs_干预点_blocked形式。transform仅在ENFORCE模式下应用把transformed_policy_target写回请求/响应对象改写 prompt、tool 参数、tool 结果或 assistant 内容。evaluate_only所有求值仍发生可用作审计但不产生任何改写测试test_evaluate_only_observes_without_mutating_...对请求、响应、tool 参数、tool 结果逐项断言原样返回。进程内接入备选guard_litellm_proxy() ASGI 中间件除 YAML 注册外仓库还提供进程内 ASGI 中间件形态guard_litellm_proxy(control, app)返回LiteLLMProxyMiddleware可包住litellm.proxy.proxy_server.app或省略 app 让其惰性加载。它把整个请求当作一次模型调用pre_model_call在重放请求体给上游前求值post_model_call在捕获上游响应后、发给客户端前求值JSON 与 chat-completion SSE 响应均先缓冲再释放保证post_model_call的脱敏/替换生效README 中建议用post_model_call做代理响应脱敏通用output点不会被该中间件求值。中间件默认拦截POST且路径匹配/chat/completions、/v1/chat/completions、/embeddings、/messages、/responses等默认路径集合。完整的可运行示例见 examples/real_packages/litellm_proxy.py它演示了用 BLOCKME 消息验证PRE_MODEL_CALL拦截的 smoke test。已知限制与设计边界文档明确了该集成的边界规划生产落地时需逐条对照映射范围有限只映射 OpenAI chat 风格的messages、assistanttool_calls与 tool 结果消息其他 schema 不在 hook 的映射范围内。未知/伪造 tool 结果失败即拒绝启用reject_unknown_tool_results时引用未知tool_call_id的结果在强制模式下直接拦截。并行 tool call 逐个求值并行工具调用是逐个评估的若宿主需要原子批回滚应禁用并行 tool call 或改用进程内适配器如guard_litellm_proxy()中间件或run_model_call/guard_tool等通用封装见 _adapters/_generic.py。审批挂起走 SDK 异常路径escalate裁决的挂起suspend以 SDK 异常方式浮出Proxy 部署需要自行把AgentControlSuspended翻译成应用侧审批流。绕过代理的流量不可控本地工具local tools、客户端侧工具执行、以及不经过 Proxy 的非 chat 路由都不在 ACS 中介范围之内——这要求你在架构上保证相关流量统一经过受控入口。小结LiteLLM Proxy guardrail hook 是 ACS无状态运行时 宿主适配哲学的典型落地Proxy 只负责把 hook 事件翻译成快照求值策略判定完全交给 ACS。通过一段guardrails:YAML 即可获得 input / pre_model_call / post_model_call / pre_tool_call / post_tool_call / output 六类干预点覆盖配合有界会话缓存、三种流式策略与 fail-closed 默认值可在不改上游应用代码的前提下为 OpenAI 兼容流量建立策略执行层。深入源码与测试适配器实现、guardrail 测试、SDK 说明可以进一步掌握其求值顺序与边界语义为生产部署的合规与审计能力提供代码级依据。【免费下载链接】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），仅供参考