
精读 `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。instructions="Handoff to the appropriate agent based on the language of the request."if,没有规则引擎。triage_agent 读到用户输入后自己判断语言,生成一个 tool call,Runner 检测到后把对话转给对应专家。剩下的事,专家来做。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],
)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。2async 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_agentconversation_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"Handoffs are represented as tools to the LLM." 2
transfer_to_french_agentcurrent_agent = french_agentfinal_output,循环结束"If the LLM does a handoff, we update the current agent and input, and re-run the loop." 3
Runner.run_streamed() 调用,但内部 agent loop 完整跑了两轮:triage + french_agent。"A triage agent routes the conversation to a specialist, and that specialist becomes the active agent for the rest of the turn." 4
| 参数 | 作用 |
|---|---|
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(当前运行上下文)。2agent_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 个目标) | 否 | 否 | 否 |
"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
"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
is_enabled + input_filter 组合拳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(),
)on_handoff 记录 + 兜底 Agenttriage_agent 的 handoffs 列表末尾加一个通用 Agent,instructions 写「对于无法识别意图的请求,礼貌告知用户并引导重新描述需求」。on_handoff 回调用于记录路由决策过程2——把「用户输入」「被路由到哪个 Agent」「路由原因(如果用 input_type 传了 reason)」写入日志,是后续优化 triage_agent instructions 和追踪误路由的最基础数据。当前版本:本文基于 OpenAI Agents SDK v0.17.0。
下期预告 #24:Input Guardrails——Routing 把对话发给专家之前,有一道拦截层可以先跑。guardrail 本身也是并行执行的,原理和asyncio.gather有关。
openai.github.io
openai.github.io
openai.github.io
github.com
mp.weixin.qq.com
mp.weixin.qq.com
This story was produced automatically by a channel. One sentence is all it takes for Neodrop to keep producing for you.