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)。

リサーチノート

大多数开发者第一次要做「意图识别分发」,直觉反应是训练一个分类器,或者写一大堆 if/else
routing.py 告诉你另一种做法:把路由规则用自然语言写进 instructions,然后让 LLM 自己做决定。1
整个分诊 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_agenthandoffs 参数直接接收 Agent 实例列表,无需额外包裹。1 SDK 内部会把这三个 Agent 转成 LLM 可见的 tool,名字分别叫 transfer_to_french_agenttransfer_to_spanish_agenttransfer_to_english_agent2

main() 执行流程

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(一段内容完成,输出换行)。1
result.current_agent 是多轮对话的关键。第一轮用户说「Je parle français」,triage_agent 触发 handoff,result.current_agent 变成 french_agent。第二轮输入直接发给 french_agent,不再经过 triage——专家接管了对话。3
result.to_input_list() 保留完整对话历史,包含 triage_agent 的思考过程和 handoff call 记录,传给下一轮保证上下文连续。1

二、handoff 是怎么工作的

从用户输入到专家回复,中间发生了什么?
官方文档把 handoff 定性为一类特殊的工具调用:
"Handoffs are represented as tools to the LLM." 2
执行路径是这样的:
  1. 用户说「Bonjour, comment ça va?」
  2. Runner 调用 triage_agent,LLM 分析语言 → 判定为法语
  3. LLM 生成 tool call:transfer_to_french_agent
  4. Runner 的 agent loop 检测到这是 handoff call,更新 current_agent = french_agent
  5. 重新进入循环,把完整对话历史发给 french_agent
  6. 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() 函数的进阶参数

routing.py 里的 handoff 是「最极简用法」,直接把 Agent 实例传给 handoffs 列表。实际生产场景里,handoff() 函数提供了更多控制维度2
参数作用
tool_name_override覆盖默认的 transfer_to_&lt;name&gt; tool 名,用于更语义化的命名
tool_description_override覆盖默认描述,让 LLM 更准确地理解何时该触发这个 handoff
on_handoffhandoff 触发时的回调(RunContextWrapperNone),可用于日志记录、数据预取
input_typePydantic schema,让 LLM 在 handoff 时生成结构化数据(如 reasonpriority)传给 on_handoff
input_filter过滤传给新 Agent 的对话历史,接收 HandoffInputData,返回修改后的 HandoffInputData
is_enabledbool 或返回 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/ 下共有四个示例,各自代表一种编排哲学。先看一张对比:
维度RoutingDeterministicParallelizationLLM-as-Judge
编排范式LLM 编排代码编排代码编排代码编排
决策者LLM(triage agent)开发者代码代码(并发原语)LLM(evaluator)
核心 APIAgent.handoffs + Runner.run_streamedRunner.run(顺序)asyncio.gather + Runner.runRunner.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 的推理过程」污染专家的判断。2
from 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_agenthandoffs 列表末尾加一个通用 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 有关。

このコンテンツについて、さらに観点や背景を補足しましょう。

  • ログインするとコメントできます。