交接(handoff)与护栏(guardrail)
本章讲两个让 agent 系统「能协作」和「能管控」的机制:handoff 让多个 agent 接力,guardrail 给输入输出装上可熔断的安全检查。
3.1 交接(Handoff):agent 之间传话筒
它要解决的小问题
复杂任务往往该拆给不同专长的 agent:一个「分诊 agent」判断用户意图,然后把对话整个交给「退款 agent」或「技术支持 agent」。问题是:LLM 怎么表达「我要交给退款 agent」?又如何把控制权真的转过去?
思路:handoff 就是一个特殊工具
关键洞察:复用 function calling。 每个 handoff 对 LLM 而言就是一个名叫 transfer_to_xxx 的工具。模型「调用」这个工具 = 表达「我要交接」。SDK 收到这个调用后,不返回工具结果,而是切换当前 agent。
这点在 01 章的决策优先级里能看到:execute_tools_and_side_effects 发现 processed_response.handoffs 非空就走交接分支(turn_resolution.py:732),返回 NextStepHandoff,外层循环据此把 current_agent 换掉(run.py:1011)。
Handoff 对象与 handoff() 工厂
Handoff dataclass(handoffs/__init__.py:93)的核心字段:tool_name(交接工具名)、input_json_schema(交接时可带的结构化参数)、on_invoke_handoff(被调用时执行、返回目标 agent)。
handoff() 工厂(handoffs/__init__.py:225)是你常用的入口。它的妙处在 _invoke_handoff(handoffs/__init__.py:278):无论模型传不传参数,这个闭包最终都 return agent——也就是说交接目标是创建 handoff 时就固定的,on_handoff 回调只用于副作用(记录、预处理),不用来动态选目标:
# handoffs/__init__.py:278 _invoke_handoff —— 跑完 on_handoff 回调后,固定返回该 agent
async def _invoke_handoff(ctx, input_json=None):
if input_type is not None and type_adapter is not None:
validated_input = _json.validate_json(json_str=input_json, ...) # 校验交接参数
result = input_func(ctx, validated_input) # 跑你的 on_handoff(ctx, input)
if inspect.isawaitable(result): await result
elif on_handoff is not None:
... # on_handoff(ctx) 只接 context
return agent # ← 目标 agent 是闭包捕获的常量
直接把一个 Agent 放进 agent.handoffs 列表也行——AgentBase 会用 Handoff.default_tool_name(handoffs/__init__.py:172)自动给它生成交接工具名。
交接时历史怎么传(input_filter / nest_handoff_history)
默认情况下,新 agent 看到的是到目前为止的完整对话历史。但有时你想裁剪(比如不让退款 agent 看到无关的闲聊)。handoff(input_filter=...)(handoffs/__init__.py:231)接一个 HandoffInputData -> HandoffInputData 的过滤器,让你改写传给下一个 agent 的输入(handoffs/__init__.py:42 的 HandoffInputData 携带 input_history、pre_handoff_items、new_items)。
还有 nest_handoff_history——可把交接前的长历史折叠成一段摘要再传(handoffs/history.py)。
一句话对比:handoff vs as_tool
| handoff | Agent.as_tool | |
|---|---|---|
| 控制权 | 交出去,新 agent 接管 | 不交,父 agent 继续主导 |
| 新 agent 看到什么 | 完整(或过滤后的)对话历史 | 父 agent 现生成的输入 |
| 适合 | 「转接到专员」 | 「调一个子能力当工具」 |
3.2 护栏(Guardrail):可熔断的安全校验
它要解决的小问题
你想在 agent 真正干活前拦住明显有害/跑题的输入(prompt 注入、辱骂、问了不该问的),或在输出发给用户前拦住不合规的回答。而且最好不增加用户感知延迟。
思路:tripwire(绊线)+ 并行执行
护栏是一个返回 GuardrailFunctionOutput(guardrail.py:19)的函数。输出里有个 tripwire_triggered 布尔——一旦为 True,立刻抛异常中止整个 run(InputGuardrailTripwireTriggered / OutputGuardrailTripwireTriggered)。这就是「绊线」:平时无感,踩到就熔断。
输入护栏:两种时序
InputGuardrail(guardrail.py:71)有个关键字段 run_in_parallel(guardrail.py:100),默认 True:
run_in_parallel=True(默认):护栏与模型调用同时跑(run.py:1195用asyncio.create_task并发起护栏和run_single_turn)。好处是不加延迟;代价是模型可能已经开始生成,但只要护栏先触发就会取消模型任务。run_in_parallel=False(前置/阻塞):护栏先跑完才允许往下。SDK 特意用它来保证「一个会触发的护栏能在 sandbox 准备/会话创建之前就拦住」(run.py:780-802的注释明确这点)。
两类输入护栏的来源:starting_agent.input_guardrails + run_config.input_guardrails(run.py:771)。只在第 0 回合、只对起始 agent 跑。
输出护栏
OutputGuardrail(guardrail.py:134)在产出最终输出之后、返 回之前跑(01 章里 NextStepFinalOutput 分支会先调 run_output_guardrails,run.py:962、run.py:1099)。它拿到 agent、最终输出、context 三样东西做校验。
运行机制
护栏对象的 run 方法很薄——调用你的函数、包成 *GuardrailResult(guardrail.py:111、guardrail.py:165)。真正的「触发就抛」逻辑在循环侧:input_guardrails_triggered(run.py import 自 helpers)检测结果里有没有 tripwire,有就抛。
装饰器糖
@input_guardrail / @output_guardrail(guardrail.py:202 / guardrail.py:284)把一个普通函数包成护栏对象,和 @function_tool 一个套路。
# 示意,基于 guardrail.py 的真实 API
from agents import input_guardrail, GuardrailFunctionOutput
@input_guardrail
async def block_secrets(ctx, agent, user_input) -> GuardrailFunctionOutput:
triggered = "密码" in str(user_input)
return GuardrailFunctionOutput(
output_info={"checked": True},
tripwire_triggered=triggered, # True → 立刻熔断整个 run
)
3.3 巧妙之处
- handoff 复用 function calling,不引入新的模型协议— —对任何支持工具调用的模型都通用。目标 agent 在闭包里固定,语义清晰,避免「动态选目标」带来的不确定性。
- 护栏默认并行,把安全检查的延迟藏在模型调用背后;同时保留
run_in_parallel=False给「必须先拦住」的强校验(如阻止 sandbox 启动)。 - tripwire 用异常传播,让「中止」这件事在调用栈里无法被忽略——调用方必须显式 catch
*TripwireTriggered。
3.4 代码地图
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| Handoff 对象 | src/agents/handoffs/__init__.py | Handoff、HandoffInputData |
| handoff 工厂 | src/agents/handoffs/__init__.py | handoff、_invoke_handoff、Handoff.default_tool_name |
| 交接历史映射 | src/agents/handoffs/history.py | default_handoff_history_mapper、nest_handoff_history |
| 交接执行 | src/agents/run_internal/turn_resolution.py | execute_handoffs |
| 输入护栏 | src/agents/guardrail.py | InputGuardrail、input_guardrail |
| 输出护栏 | src/agents/guardrail.py | OutputGuardrail、output_guardrail |
| 护栏结果/输出 | src/agents/guardrail.py | GuardrailFunctionOutput、InputGuardrailResult |
| 护栏调度 | src/agents/run_internal/guardrails.py | (run_input_guardrails / run_output_guardrails) |