代理钩子是代理框架的一项原生能力,用于在代理执行过程中明确定义的节点上施加治理和运行时控制。 它实现与框架无关 的 AGENT-HOOKS-0.1 协定,因此策略引擎、审批网关、预算防护、内容筛选器和出口控件可以面向一个常见的控制面。
重要
Agent Hooks 是控制平面,而不是遥测平面。 每个拦截器都会返回一个结论。 在 enforce 模式中,框架对该判决执行操作;在 evaluate_only 模式下,它会记录判决而不更改执行。 对被动跟踪、指标和日志使用 可观测性 。
Agent Hooks 在 Python 中尚处于实验阶段。 工厂会在首次使用时发出 ExperimentalWarning ,其 API 可以在正式发布之前进行更改。
何时使用智能体挂钩
当独立开发的控件需要在代理输入、模型调用、工具调用和最终输出之间采用一套统一且可强制执行的约定时,请使用代理钩子。
| Capability | 将其用于 |
|---|---|
| 智能体挂钩 | 跨智能体生命周期标准化的策略决定、转换、审批、预算和流出量控制。 |
| 代理中间件 | 不需要智能体挂钩协定或其核心运行时保证的应用程序特定的交叉行为。 |
| 借助 FIDES 实现代理安全 | 不受信任的或机密内容的确定性信息流标签和策略。 |
| 工具审批 | 人工确认单个函数工具调用。 |
| 可观察性 | 不控制执行的被动式跟踪、指标和日志。 |
代理框架强制实施的内容
当你为代理添加 Agent Hooks 时,Agent Framework 会在代理运行、模型调用和工具调用中实施一套协调一致的约束边界。 运行时提供以下保证:
- 失败关闭:拒绝会阻止受保护的操作。 无效上下文、无效判定、拦截器失败和强制执行失败不会在无提示的情况下绕过控制。
- 转换回写: 转换会更改执行实际使用的原生消息、工具参数、工具结果或最终响应。 如果无法应用转换,运行将失败并关闭。
- 缓冲流: 在完整的模型响应和最终输出通过截获点之前,没有响应更新到达调用方。
-
以裁定为准的持久化:持久化会等待适用于它的裁定。 标准运行后持久化会等待
output;每次服务调用的历史记录持久化会等待每个post_model_call。 - 完成捆绑包安装: 代理、聊天和函数部件作为一个单元进行安装,因此无法意外配置不完整的强制边界。
该约定是用于协作的,而非进程隔离边界。 侦听器在主机进程中运行,并接收做出决策所需的内容。 仅注册你信任的侦听器。
安装智能体挂钩
将 Agent Hooks SDK 作为直接依赖项安装:
pip install agent-hooks-sdk
如果使用 uv:
uv add agent-hooks-sdk
依赖项 agent-hooks-sdk 是延迟导入的。 除非你创建 Agent Hooks 中间件包,否则导入 agent_framework 并不会加载 SDK。
注释
agent-framework-core 不包括额外的 agent-hooks 内容。 在创建 Agent Hooks 中间件捆绑包之前,请先单独安装 agent-hooks-sdk。
添加侦听器
拦截器接收一个 agent_hooks.AgentContext(即规范的上下文映射,而不是代理中间件使用的 agent_framework.AgentContext),并返回判断结果。 以下拦截器阻止包含单词 secret的最终输出。 该示例假定 client 已配置代理框架聊天客户端。
from agent_framework import Agent, create_agent_hooks_middleware
from agent_hooks import ALLOW, AgentContext, InterceptionBlocked, Verdict
class SecretEgressGuard:
def intercept(self, context: AgentContext) -> Verdict:
if (
context["interception_point"] == "output"
and "secret" in str(context["target"]).lower()
):
return Verdict.deny(
reason="secret_in_output",
message="The final response contains restricted content.",
)
return ALLOW
hooks = create_agent_hooks_middleware(
{"secret-egress": SecretEgressGuard()},
)
agent = Agent(
client=client,
instructions="You are a helpful assistant.",
middleware=[hooks],
)
try:
response = await agent.run("Summarize the account details.")
except InterceptionBlocked as exc:
print(f"Blocked: {exc.result.verdict.reason}")
将捆绑包作为智能体的 middleware 列表的一个元素传递。 在每个智能体上安装一个智能体挂钩捆绑包。
拦截点
代理框架会自动生成相应的拦截点:
| 拦截点 | 发出时 | 变换目标 |
|---|---|---|
agent_startup |
在智能体挂钩会话中的首次输入之前 | 不可转换 |
input |
当外部请求到达代理时 | 输入内容和角色 |
pre_model_call |
在每个模型请求之前 | 发送给模型的消息 |
post_model_call |
在每个完整的模型响应之后 | 响应内容、由框架执行的工具调用和完成原因 |
pre_tool_call |
在每个框架执行的工具调用之前 | 工具参数 |
post_tool_call |
工具成功或失败后 | 工具结果 |
output |
在最终响应到达调用方之前 | 最终响应内容 |
agent_shutdown |
当 Agent Hooks 会话完成、失败或被取消时 | 不可转换 |
agent_startup.tools_registered 是运行启动工具快照。 每个 pre_model_call 载荷都在其可选的 tools 字段中包含该模型调用的有效工具。 这包括在运行过程中由上下文提供程序、已连接的 MCP 服务器或渐进式披露添加的工具。 当调用没有工具或工具集无法投影时,将省略该字段。
调用工具的运行过程通常会输出如下内容:
agent_startup
input
pre_model_call
post_model_call →→→→ pre_tool_call → post_model_callpost_tool_callpre_model_calloutput →→→→agent_shutdown
判决
合同有三个决定: allow, deny和 transform。 Python SDK 还提供警告和可解除拒绝的帮助程序。
| 结果 | Python API | Behavior |
|---|---|---|
| 允许 |
ALLOW 或 Verdict(decision=Decision.ALLOW) |
继续执行目标不变。 |
| 警告后允许 | Verdict.warn(...) |
继续,并将该警告包含在拦截记录中。 |
| Deny | Verdict.deny(...) |
阻止受保护操作。 |
| 拒绝待审批 | Verdict.escalate(...) |
除非配置的审批解析程序返回许可裁定,否则阻止。 |
| 转换 | Verdict(decision=Decision.TRANSFORM, transform=Transform(...)) |
改写 $target 下的值,然后继续使用改写后的值。 |
运行级别和模型级别拒绝引发 InterceptionBlocked 并阻止受保护的结果到达调用方或下一阶段。 在工具衔接处,策略拒绝会阻止工具操作或丢弃其结果,并向模型返回包含策略原因的控制错误,没有被拒绝的目标有效负载。 这允许代理循环继续。 主机故障或强制执行失败会中止运行。
在函数中间件内中止运行
从 MiddlewareFailure 导入 agent_framework。 函数中间件通常将普通异常转换为工具错误结果,然后让代理循环继续。 如果函数中间件无法安全地继续执行,请通过基础异常引发 MiddlewareFailure。 运行时中止运行并将失败传播到调用方,而不是将其转换为工具结果。
不要在中间件中捕获 MiddlewareFailure 。 捕获该异常会使循环继续执行,并将 fail-closed 行为变为 fail-open 行为。 当其函数中间件强制执行层失效时,Agent Hooks 会在内部使用此信号。 按顺序传递自定义的故障关闭型中间件,例如 middleware=[policy_middleware]。
对于并发工具调用,运行时会先取消其他仍在进行中的同级调用,然后再传递该失败。 取消是合作的,因此已在工作线程中运行的同步工具可能会完成其副作用,但其结果将被丢弃。
应用转换
转换路径必须始于 $target。 例如,侦听器可以替换最终响应内容:
from agent_hooks import ALLOW, AgentContext, Decision, Transform, Verdict
class OutputRedactor:
def intercept(self, context: AgentContext) -> Verdict:
if context["interception_point"] != "output":
return ALLOW
return Verdict(
decision=Decision.TRANSFORM,
reason="redacted_output",
transform=Transform(
path="$target.content",
value="[Response removed by policy]",
),
)
转换将应用于 Agent Framework Content 值,保留支持的丰富内容,而不是将每个值都简化为纯文本。 格式错误的路径或不兼容的替换将失败并关闭,而不是使用原始值继续。
工具批准和参数转换
Agent Framework 工具批准和 Agent Hooks 批准接口是彼此独立的机制。 对于具有 approval_mode="always_require" 的函数工具,Agent Framework 会在函数中间件运行之前创建人工审批请求。 因此,转换 pre_tool_call 可以在用户批准原始值后更改参数。
Warning
对于使用 approval_mode="always_require" 的工具,不要在 pre_tool_call 处转换参数。 在 post_model_call 处转换工具调用,使框架审批请求包含转换后的值,或在 Verdict.escalate(...) 处返回 resolver,并通过 Agent Hooks pre_tool_call 处理审批。
流式处理和持久性
Agent Hooks 保留了流式 API,但采用缓冲输出机制。 代理框架组装完整的模型响应、发出 post_model_call、组装最终代理响应,并在发布任何更新之前发出 output 。 如果任一点都拒绝响应,则调用方不会收到部分更新。
此行为以逐令牌延迟换取失败关闭输出强制。 输出转换也会反映在最终向调用方发布的更新中。
持久化受涵盖持久化操作的拦截点控制:
- 默认情况下,历史记录和其他运行后提供程序工作都会等待
output裁定。 被拒绝的输出不会被持久保存,而输出转换会在转换完成后被持久保存。 - 当您在
Agent构造函数或client.as_agent(...)上设置require_per_service_call_history_persistence=True时,每次模型交换都会在post_model_call裁定允许后持久化。 之后的output拒绝不会回滚已许可的历史记录。 - 对于默认的运行后持久化,重试尝试仍会在最终
output决定之后。 按服务调用模式则会持久保留每个通过post_model_call的模型响应。
重要
如果模型内容不能持久,在 require_per_service_call_history_persistence=True 时在 post_model_call 强制执行该策略。 仅输出流出量策略保护到达调用方的内容,但不会追溯性地删除已经在 post_model_call 许可并持久化的模型交换。
会话和审核记录
默认情况下,每个代理运行都会创建一个代理挂钩会话。
agent_startup 和 agent_shutdown 标示运行的开始和结束,记录接收一个单调递增序列的会话 ID。
使用 record_sink 接收每个 InterceptionRecord:
records = []
hooks = create_agent_hooks_middleware(
{"secret-egress": SecretEgressGuard()},
record_sink=records.append,
)
截获记录会记录决策、原因、拦截器摘要、模式、身份标识和序列,而不会将截获的有效负载复制到审计记录中。 拦截器本身仍会接收到完整的上下文。
使用一个会话跨越多个运行
当应用程序有更长期的智能体挂钩会话时,如与一个审批分类帐的对话,使用 create_agent_hooks_middleware_from_emitter():
from agent_framework import Agent, create_agent_hooks_middleware_from_emitter
from agent_hooks import AgentContextBuilder, InterceptionEmitter
emitter = InterceptionEmitter().register(SecretEgressGuard())
builder = AgentContextBuilder(
agent_id="support-agent",
framework="agent-framework",
session_id="conversation-42",
)
hooks = create_agent_hooks_middleware_from_emitter(emitter, builder)
agent = Agent(client=client, middleware=[hooks])
await emitter.emit(builder.agent_startup(tools_registered=[]))
await agent.run("First turn")
await agent.run("Second turn")
await emitter.emit(builder.agent_shutdown(reason="completed"))
在这种形式下,应用程序负责配置发射器,并负责启动、关闭以及错误清理。 中间件输出从 input 到 output 的每次运行点数。
配置强制执行
create_agent_hooks_middleware() 接受以下控件:
| Parameter | Purpose |
|---|---|
interceptors |
拦截器序列或名称到拦截器的映射。 至少需要一个。 |
resolver |
通过审批渠道解决可解除的拒绝。 如果没有解析程序,拒绝仍然有效。 |
mode |
"enforce" 应用裁定。
"evaluate_only" 记录将发生的情况,但允许执行每个操作。 |
composition |
选择多个拦截器判定结果的合并方式。 |
identity_provider |
生成与内容绑定的上下文标识符。 默认值为 "jcs-sha256"。 |
timeout |
可等待调用的每个拦截器和解析程序超时。 默认值为 5 秒。 阻止事件循环的同步拦截器或解析程序不能被此超时抢占。 |
record_sink |
接收每条无有效负载的拦截记录。 |
默认组合是顺序的 first_deny,配置了停止折叠的审批。 因此,拦截器的顺序很重要:将必须始终运行的控制项放在可请求审批的控制项之前。 在选择另一个组合配置文件之前,请先参阅智能体挂钩生产检查清单。
以仅评估模式推出
使用 evaluate_only 在正式强制执行前衡量策略行为:
hooks = create_agent_hooks_middleware(
{"secret-egress": SecretEgressGuard()},
mode="evaluate_only",
record_sink=records.append,
)
在此模式下,拦截器会运行,记录中也会包含其判定结果,但任何操作都不会被阻止或转换。 不要将 evaluate_only 部署描述为强制治理。
组合规则
将该捆绑包置于代理的中间件列表首位,使其形成最外层的执行边界:
agent = Agent(
client=client,
middleware=[
create_agent_hooks_middleware([SecretEgressGuard()]),
application_middleware,
],
)
遵循以下规则:
- 为每个代理恰好安装一个 Agent Hooks 软件包。 堆叠捆包不予接受。
- 保持捆绑包完整。 无法单独安装其代理、聊天和函数中间件。
- 在
Agent上安装捆绑包,而不是直接在聊天客户端或通过上下文提供程序安装。 - 位于捆绑包之前的中间件不在强制执行边界之内。 将外部位置视为外部信任。
- 当每个嵌套代理的内部模型和工具活动也需要拦截时,为每个嵌套代理提供自己的捆绑包。
当前限制
- 仅Python:代理挂钩尚未在 .NET 或 Go SDK 中实现。
- 实验性 API: 在正式发布之前,工厂签名和行为可能会更改。
- 缓冲式流式处理:更新不会逐令牌发布,因为在作出失败关闭判定之前,必须先完成全部输出。
-
托管工具:由模型提供方执行的工具不会经过 Agent Framework 的函数调用接口层。 它们的调用和输出结果会显示在
post_model_call中,但pre_tool_call和post_tool_call无法阻止提供方的服务器端执行。 - 协作边界: Agent Hooks 不会对拦截器进行沙箱隔离,也不会防御恶意主机。 未涵盖绕过受保护的代理管道的代码路径。
- 侦听器可用性会影响代理可用性: 在强制模式下,拦截器故障或超时会按设计阻止受保护的操作。
有关生产推出、失败原因和警报指南,请参阅智能体挂钩操作 runbook。