企业官网建设流程全解析

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-toolkit

Agent Control Specification（ACS）为自主 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", "fastapi>=0.100", ]

注意两点：一是必须安装litellm[proxy]而非裸litellm——proxy extra 会引入运行时所需的代理服务依赖（README 中对此有明确警告）；二是 SDK 本身要求 Python >= 3.11，核心策略引擎由 Rust 实现并通过 maturin 构建为 CPython 3.11+ ABI3 wheel，安装官方 wheel 不需要本地 Rust 工具链。

Proxy YAML：零代码注册 guardrail

LiteLLM 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 构造器 中均有对应实现，含义如下：

构造器对入参做了严格校验：control与manifest_path必须二选一（同时传或都不传都会抛ValueError）；streaming仅接受三个枚举值之一。session_cache_size与session_ttl_seconds经max(1, ...)与max(0.0, ...)归一化后用于构建_LiteLLMSessionCache。

Hook 到干预点的映射

guardrail 的核心价值在于把 LiteLLM 的钩子事件翻译成 ACS 干预点求值。完整映射关系如下：

对照源码可以还原这条求值链路（_adapters/litellm.py）：

• async_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.py）：

1. metadata.agent_control_session_id
2. metadata.acs_session_id
3. metadata.litellm_session_id
4. 顶层litellm_session_id
5. 顶层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.py）：

• buffer（默认）：排干 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 的流式请求直接抛AdapterUnsupportedError（fail closed），避免错误重建响应（见 _adapters/litellm.py）。

拒绝与转换的落地语义

无论哪种接入方式，强制模式的裁决语义保持一致（对应_evaluate与_has_transform，[_adapters/litellm.py](https://link.gitcode.com/i/93e7a5942b4bdb781f78123f420703d0#L514-L522, L596-L599)）：

• deny：AgentControlBlocked被转换为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