跳到主要内容

交接(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:42HandoffInputData 携带 input_historypre_handoff_itemsnew_items)。

还有 nest_handoff_history——可把交接前的长历史折叠成一段摘要再传(handoffs/history.py)。

一句话对比:handoff vs as_tool

handoffAgent.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:1195asyncio.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:962run.py:1099)。它拿到 agent、最终输出、context 三样东西做校验。

运行机制

护栏对象的 run 方法很薄——调用你的函数、包成 *GuardrailResult(guardrail.py:111guardrail.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__.pyHandoffHandoffInputData
handoff 工厂src/agents/handoffs/__init__.pyhandoff_invoke_handoffHandoff.default_tool_name
交接历史映射src/agents/handoffs/history.pydefault_handoff_history_mappernest_handoff_history
交接执行src/agents/run_internal/turn_resolution.pyexecute_handoffs
输入护栏src/agents/guardrail.pyInputGuardrailinput_guardrail
输出护栏src/agents/guardrail.pyOutputGuardrailoutput_guardrail
护栏结果/输出src/agents/guardrail.pyGuardrailFunctionOutputInputGuardrailResult
护栏调度src/agents/run_internal/guardrails.py(run_input_guardrails / run_output_guardrails)