OpenAI Agents SDK #23:把路由逻辑写进 instructions,然后什么都不用写了
精读 `routing.py`(77 行)的完整实现:4 个 Agent 的定义逻辑(triage_agent + 3 语言专家)、`Runner.run_streamed` + `stream_events` 流式对话循环、`result.current_agent` 的多轮保持机制。深入 handoff 内部原理——LLM 如何通过 `transfer_to_<agent_name>` tool call 实现路由、agent loop 如何在同一 run 内切换 Agent。附 `handoff()` 函数 6 个关键进阶参数详解,以及 Routing / Deterministic / Parallelization / LLM-as-Judge 四模式 7 维度横向对比矩阵,结尾三条生产落地建议覆盖路由粒度设计、误路由防护(`is_enabled` + `input_filter`)和 fallback 策略(`on_handoff` 日志 + 兜底 Agent)。
Research Brief
大多数开发者第一次要做「意图识别分发」,直觉反应是训练一个分类器,或者写一大堆
if/else。整个分诊 Agent 的核心代码是这一行:
instructions="Handoff to the appropriate agent based on the language of the request."没有分类器,没有
if,没有规则引擎。triage_agent 读到用户输入后自己判断语言,生成一个 tool call,Runner 检测到后把对话转给对应专家。剩下的事,专家来做。一、routing.py 完整源码精读
文件共 77 行,有效逻辑约 50 行1。结构极简,分两层:分诊 Agent + 专家 Agent 群。
Agent 定义层
french_agent = Agent(name="french_agent", instructions="You only speak French.")
spanish_agent = Agent(name="spanish_agent", instructions="You only speak Spanish.")
english_agent = Agent(name="english_agent", instructions="You only speak English.")
triage_agent = Agent(
name="triage_agent",
instructions="Handoff to the appropriate agent based on the language of the request.",
handoffs=[french_agent, spanish_agent, english_agent],
)4 个 Agent,3 个语言专家 + 1 个分诊器。专家 Agent 的设计刻意极简:无
tools,无 handoffs,无 output_type——它们只做一件事,用指定语言回答。triage_agent 的 handoffs 参数直接接收 Agent 实例列表,无需额外包裹。1 SDK 内部会把这三个 Agent 转成 LLM 可见的 tool,名字分别叫 transfer_to_french_agent、transfer_to_spanish_agent、transfer_to_english_agent。2main() 执行流程
async def main():
current_agent = triage_agent
input_items: list[TResponseInputItem] = []
conversation_id = uuid.uuid4().hex[:16]
while True:
user_input = input("Enter a message: ")
input_items.append({"content": user_input, "role": "user"})
with trace("Routing example", group_id=conversation_id):
result = Runner.run_streamed(current_agent, input=input_items)
async for event in result.stream_events():
if not isinstance(event, RawResponsesStreamEvent):
continue
data = event.data
if isinstance(data, ResponseTextDeltaEvent):
print(data.delta, end="", flush=True)
elif isinstance(data, ResponseContentPartDoneEvent):
print()
input_items = result.to_input_list()
current_agent = result.current_agent几个细节值得注意:
conversation_id 作为 trace 的 group_id,把整个多轮会话的所有 trace 串在同一条追踪记录下,便于事后调试。流式事件过滤:
stream_events() 返回的事件有多种类型,代码只处理 RawResponsesStreamEvent,从中提取两种:ResponseTextDeltaEvent(逐字符流式输出)和 ResponseContentPartDoneEvent(一段内容完成,输出换行)。1result.current_agent 是多轮对话的关键。第一轮用户说「Je parle français」,triage_agent 触发 handoff,result.current_agent 变成 french_agent。第二轮输入直接发给 french_agent,不再经过 triage——专家接管了对话。3result.to_input_list() 保留完整对话历史,包含 triage_agent 的思考过程和 handoff call 记录,传给下一轮保证上下文连续。1二、handoff 是怎么工作的
从用户输入到专家回复,中间发生了什么?
官方文档把 handoff 定性为一类特殊的工具调用:
"Handoffs are represented as tools to the LLM." 2
执行路径是这样的:
- 用户说「Bonjour, comment ça va?」
- Runner 调用 triage_agent,LLM 分析语言 → 判定为法语
- LLM 生成 tool call:
transfer_to_french_agent - Runner 的 agent loop 检测到这是 handoff call,更新
current_agent = french_agent - 重新进入循环,把完整对话历史发给 french_agent
- french_agent 用法语生成回复,产出
final_output,循环结束
官方对第 4 步有一句精准的描述:
"If the LLM does a handoff, we update the current agent and input, and re-run the loop." 3
handoff 不终止 run,而是在同一个 run 内切换 Agent 继续执行。从 Runner 外部看只发生了一次
Runner.run_streamed() 调用,但内部 agent loop 完整跑了两轮:triage + french_agent。另外,官方文档对 triage 模式的建议场景说得很明确:
"A triage agent routes the conversation to a specialist, and that specialist becomes the active agent for the rest of the turn." 4
handoff() 函数的进阶参数
| 参数 | 作用 |
|---|---|
tool_name_override | 覆盖默认的 transfer_to_<name> tool 名,用于更语义化的命名 |
tool_description_override | 覆盖默认描述,让 LLM 更准确地理解何时该触发这个 handoff |
on_handoff | handoff 触发时的回调(RunContextWrapper → None),可用于日志记录、数据预取 |
input_type | Pydantic schema,让 LLM 在 handoff 时生成结构化数据(如 reason、priority)传给 on_handoff |
input_filter | 过滤传给新 Agent 的对话历史,接收 HandoffInputData,返回修改后的 HandoffInputData |
is_enabled | bool 或返回 bool 的函数,运行时动态启用/禁用某条 handoff 路由 |
HandoffInputData 包含 5 个字段:input_history(run 开始前的历史)、pre_handoff_items(handoff 前当前 turn 生成的内容)、new_items(当前 turn 的新内容)、input_items(可选,替换 new_items 转发给新 Agent)、run_context(当前运行上下文)。2三、Routing 在四种编排模式里处于什么位置
agent_patterns/ 下共有四个示例,各自代表一种编排哲学。先看一张对比:| 维度 | Routing | Deterministic | Parallelization | LLM-as-Judge |
|---|---|---|---|---|
| 编排范式 | LLM 编排 | 代码编排 | 代码编排 | 代码编排 |
| 决策者 | LLM(triage agent) | 开发者代码 | 代码(并发原语) | LLM(evaluator) |
| 核心 API | Agent.handoffs + Runner.run_streamed | Runner.run(顺序) | asyncio.gather + Runner.run | Runner.run + output_type |
| 延迟特性 | 约 2 轮 LLM 调用 | 串行 = 各步之和 | 并行 = 最慢 Agent | 循环 = N × 单轮 |
| 对话控制权 | 专家 Agent 接管 | 开发者全程 | 开发者全程 | 开发者全程 |
| 多轮对话 | ✅ result.current_agent 保持 | ❌ | ❌ | ❌ |
| 是否用 handoff | 是(≥2 个目标) | 否 | 否 | 否 |
官方 README 对 Routing 模式的定位是:
"In many situations, you have specialized sub-agents that handle specific tasks. You can use handoffs to route the task to the right agent." 5
四种模式的选择判断逻辑:
- 需要根据用户输入动态选择专家,且专家直接回复用户 → Routing
- 任务步骤确定、需要代码级 gate 门控 → Deterministic
- 多个子任务互不依赖、或同一任务多跑择优 → Parallelization
- 输出质量要求高、可接受多轮迭代延迟 → LLM-as-Judge
中文开发者社区把这个设计哲学概括得很直接——「OpenAI 说 Agent 是一条 handoff 链」。6 与 Anthropic(有状态运行时)、LangChain(有向图)的设计哲学并列,handoff 链是 OpenAI SDK 的核心隐喻。
官方文档对 Routing(handoffs)和「Agents as tools」的选择边界也说得很清楚:
"Use handoffs when routing itself is part of the workflow and you want the chosen specialist to own the next part of the interaction." 4
"Use agents as tools when a specialist should help with a bounded subtask but should not take over the user-facing conversation." 4
一句话区分:专家要接管对话 → handoffs(Routing);专家只完成有界子任务 → Agents as tools。
四、生产落地建议
① 路由粒度:专家 Agent 要足够专
routing.py 里的语言路由是最简单的形式。实际业务场景下,专家 Agent 的 instructions 里要把「只处理什么」写得足够清楚——不只是「你是退款专家」,而是「你处理金额在 500 元以内、下单 7 天内的退款申请;超出范围直接告知用户无法处理」。
官方文档建议投资于 prompt 质量:「明确可用工具、使用方式和参数约束」。4 专家 Agent 的边界越清晰,triage_agent 的路由决策就越精准,误路由率越低。
② 误路由防护:is_enabled + input_filter 组合拳
不要期望 LLM 100% 路由正确。两个防护层可以叠加:
is_enabled 接受一个函数,运行时动态判断某条 handoff 是否可用2——比如业务时间外自动禁用专家 Agent,强制流量走兜底通道。input_filter 控制新 Agent 接收到的对话历史。如果 triage_agent 产生了大量内部推理内容,用 input_filter 把这部分裁剪掉,让专家 Agent 只看到干净的用户输入,避免「上一轮 triage 的推理过程」污染专家的判断。2from agents.extensions.handoff_filters import remove_all_tools
handoff(
agent=refund_agent,
input_filter=remove_all_tools, # 内置过滤器,移除 tool call 记录
is_enabled=lambda ctx: is_business_hour(),
)③ fallback 设计:on_handoff 记录 + 兜底 Agent
生产级 Routing 系统需要一个兜底 Agent——当所有专家都无法匹配时,保证用户有响应。可以在
triage_agent 的 handoffs 列表末尾加一个通用 Agent,instructions 写「对于无法识别意图的请求,礼貌告知用户并引导重新描述需求」。on_handoff 回调用于记录路由决策过程2——把「用户输入」「被路由到哪个 Agent」「路由原因(如果用 input_type 传了 reason)」写入日志,是后续优化 triage_agent instructions 和追踪误路由的最基础数据。国内已有团队基于这套架构搭出了完整的客服演示系统:分诊智能体统一入口,自动识别意图,精准派单给专业智能体,「无需人工转接,Agent 之间自动完成上下文传递与任务切换」。7 路由粒度覆盖:航班信息、预订改签、座位服务、FAQ、退款理赔——5 类专家 Agent,一个 triage 门口。
当前版本:本文基于 OpenAI Agents SDK v0.17.0。
下期预告 #24:Input Guardrails——Routing 把对话发给专家之前,有一道拦截层可以先跑。guardrail 本身也是并行执行的,原理和asyncio.gather有关。
Add more perspectives or context around this content.