第 3 章：多 Agent 协作与转接（Handoff）

一个 Agent 什么都干，就像一个人既当前台又当技术支持还兼财务——迟早崩盘。这一章我们学会让多个 Agent 各司其职，通过 Handoff 机制互相"传球"。

本章目标

掌握 Handoff 机制，学会搭建"分诊 Agent + 专业 Agent"的多 Agent 协作系统。

为什么需要多个 Agent？

先看一个反面案例。假设你要做一个客服系统，你可能会这样写：

agent = Agent(
    name="万能客服",
    instructions="""你是一个客服，你需要：
    1. 判断用户说的是什么语言
    2. 用对应语言回复
    3. 如果是技术问题，帮用户排查
    4. 如果是账单问题，帮用户查账
    5. 如果是退款问题，帮用户申请退款
    6. 如果是投诉，安抚用户情绪
    ...（越写越长）""",
    tools=[查天气, 查订单, 查账单, 提工单, 退款, ...],  # 工具也越塞越多
)

问题很明显：

解决方案很简单——分工：

前台（Triage Agent）：判断用户要什么，转给对的人
技术支持：只处理技术问题，只带技术工具
账单客服：只处理账单问题，只带账单工具

每个 Agent 的 instructions 短而精确，工具少而专业，效果自然好得多。

Handoff 基础：最简单的转接

Handoff 翻译过来就是"交接"。最直白的理解：一个 Agent 把对话交给另一个 Agent 继续处理。

在 SDK 里用起来非常简单——给 Agent 加一个 handoffs 参数就行：

import asyncio
from openai import AsyncOpenAI
from agents import Agent, OpenAIChatCompletionsModel, Runner, set_tracing_disabled

# 关闭追踪（本地开发不需要）
set_tracing_disabled(True)

# 连接本地模型
client = AsyncOpenAI(base_url="http://localhost:8317/v1", api_key="sk-12345678")
model = OpenAIChatCompletionsModel(model="gpt-5.2", openai_client=client)

# 两个专业 Agent
chinese_agent = Agent(
    name="中文客服",
    instructions="你是中文客服，只用中文回答，语气亲切。",
    model=model,
)

english_agent = Agent(
    name="English Agent",
    instructions="You are an English agent. Reply in English only.",
    model=model,
)

# 前台 Agent，通过 handoffs 指定它可以转给谁
triage_agent = Agent(
    name="前台",
    instructions="根据用户使用的语言，转给对应的客服。中文转中文客服，英文转 English Agent。",
    model=model,
    handoffs=[chinese_agent, english_agent],
)

async def main():
    result = await Runner.run(triage_agent, input="帮我查一下订单状态")
    print(f"最终处理的 Agent: {result.last_agent.name}")
    print(f"回复: {result.final_output}")

if __name__ == "__main__":
    asyncio.run(main())

就这么几行代码，背后发生了什么：

  1. 用户消息发给 triage_agent
  2. triage_agent 看到是中文，决定转给 中文客服
  3. SDK 自动把对话上下文传给 中文客服
  4. 中文客服 接手，生成回复

result.last_agent 告诉你最终是谁在干活——这个信息在日志记录和路由统计时很有用。

分诊模式：Triage Agent

上面的例子其实就是经典的分诊模式（Triage Pattern）：

                    用户
                     |
                     v
              +-----------+
              | Triage    |
              | Agent     |
              +-----+-----+
                    |
         +----------+----------+
         |          |          |
         v          v          v
      Agent A    Agent B    Agent C
      (专业1)    (专业2)    (专业3)

Triage Agent 自己不干活（不带 tools），只做一件事：看清问题，转给对的人。

这个模式的好处：

handoff_description：让分诊更准确

当你把一个 Agent 放到 handoffs 列表里时，SDK 会自动生成一个工具描述告诉 Triage Agent："这个人能干什么"。但自动生成的描述可能不够精确，你可以用 handoff_description 显式指定：

tech_agent = Agent(
    name="技术支持",
    # 这个描述是给 Triage Agent 看的，帮它判断什么时候该转过来
    handoff_description="处理技术问题，包括系统故障、登录异常、软件报错等",
    instructions="你是技术支持专员...",
    model=model,
)

billing_agent = Agent(
    name="账单客服",
    handoff_description="处理账单和支付相关问题，包括查询账单、退款、付款失败等",
    instructions="你是账单客服...",
    model=model,
)

handoff_description 和 instructions 的区别：

handoff() 高级用法

直接把 Agent 扔进 handoffs 列表是最简单的用法。但有时候你需要更多控制——比如在转接的瞬间做一些准备工作。这时候就需要 handoff() 函数。

on_handoff 回调：转接时做准备

想象一个航空公司客服系统：当用户要改座位时，Triage Agent 把对话转给"座位预订 Agent"。在转接的那一刻，我们需要自动分配一个航班号。

import asyncio
import random
from dataclasses import dataclass

from openai import AsyncOpenAI
from agents import (
    Agent,
    OpenAIChatCompletionsModel,
    Runner,
    RunContextWrapper,
    function_tool,
    handoff,
    set_tracing_disabled,
)

set_tracing_disabled(True)

client = AsyncOpenAI(base_url="http://localhost:8317/v1", api_key="sk-12345678")
model = OpenAIChatCompletionsModel(model="gpt-5.2", openai_client=client)


# ---- 共享上下文 ----

@dataclass
class ServiceContext:
    """客服系统的共享上下文"""
    customer_name: str | None = None
    flight_number: str | None = None
    ticket_id: str | None = None


# ---- 工具 ----

@function_tool
def update_seat(ctx: RunContextWrapper[ServiceContext], confirmation: str, new_seat: str) -> str:
    """更新航班座位。"""
    # 从上下文拿到之前设置好的航班号
    flight = ctx.context.flight_number
    return f"航班 {flight}，确认号 {confirmation}，座位已更新为 {new_seat}"


@function_tool
def lookup_faq(question: str) -> str:
    """查询常见问题。"""
    faqs = {
        "行李": "每位乘客可携带一件手提行李（不超过 7kg）和一件托运行李（不超过 23kg）。",
        "wifi": "机上提供免费 WiFi，连接 Airline-WiFi 即可使用。",
        "餐食": "经济舱提供免费简餐，商务舱提供正餐。",
    }
    for keyword, answer in faqs.items():
        if keyword in question.lower():
            return answer
    return "抱歉，暂时没有找到相关信息。"


# ---- on_handoff 回调：转接时自动分配航班号 ----

async def on_seat_booking_handoff(ctx: RunContextWrapper[ServiceContext]) -> None:
    """当转给座位预订 Agent 时，自动分配一个航班号"""
    flight_number = f"CA-{random.randint(1000, 9999)}"
    ctx.context.flight_number = flight_number
    print(f"[系统] 已分配航班号: {flight_number}")


# ---- Agent 定义 ----

faq_agent = Agent[ServiceContext](
    name="FAQ助手",
    handoff_description="回答常见问题，比如行李规定、WiFi、餐食等",
    instructions="你是 FAQ 助手，使用 lookup_faq 工具回答用户的常见问题。如果回答不了，告诉用户联系人工客服。",
    model=model,
    tools=[lookup_faq],
)

seat_agent = Agent[ServiceContext](
    name="座位预订",
    handoff_description="帮用户修改航班座位",
    instructions="你是座位预订专员。先问用户的确认号，再问想换到哪个座位，然后用 update_seat 工具完成修改。",
    model=model,
    tools=[update_seat],
)

triage_agent = Agent[ServiceContext](
    name="航空客服前台",
    instructions="你是航空公司客服前台，根据用户需求转给对应的专员。座位相关的转给座位预订，其他问题转给 FAQ 助手。",
    model=model,
    handoffs=[
        faq_agent,                                                     # 普通转接
        handoff(agent=seat_agent, on_handoff=on_seat_booking_handoff),  # 带回调的转接
    ],
)


# ---- 运行 ----

async def main():
    context = ServiceContext(customer_name="张三")

    # 测试 FAQ
    result = await Runner.run(triage_agent, input="请问可以带多重的行李？", context=context)
    print(f"[{result.last_agent.name}] {result.final_output}")
    print()

    # 测试座位预订（会触发 on_handoff 回调）
    result = await Runner.run(triage_agent, input="我想换个靠窗的座位", context=context)
    print(f"[{result.last_agent.name}] {result.final_output}")
    print(f"[系统] 上下文中的航班号: {context.flight_number}")

if __name__ == "__main__":
    asyncio.run(main())

关键点：

on_handoff 带输入参数

如果你希望在转接时，LLM 还能传一些结构化数据给回调函数，可以用 input_type：

from pydantic import BaseModel

class HandoffData(BaseModel):
    reason: str        # 转接原因
    priority: str      # 优先级

async def on_escalation(ctx: RunContextWrapper[ServiceContext], data: HandoffData) -> None:
    print(f"转接原因: {data.reason}, 优先级: {data.priority}")

escalation_handoff = handoff(
    agent=senior_agent,
    on_handoff=on_escalation,
    input_type=HandoffData,  # 告诉 SDK 期望的输入类型
)

这样 LLM 在决定转接时，还会生成一段符合 HandoffData schema 的 JSON，SDK 会自动解析并传给你的回调。

input_filter：控制转接时传递的上下文

默认情况下，转接给新 Agent 时，整个对话历史都会传过去。但有时候你不希望这样——比如之前的工具调用记录对新 Agent 来说是噪音。

input_filter 让你在转接时过滤对话历史：

from agents import HandoffInputData, handoff
from agents.extensions import handoff_filters

def clean_handoff_filter(data: HandoffInputData) -> HandoffInputData:
    """转接时去掉所有工具调用记录，只保留用户和助手的对话"""
    # SDK 内置了常用的过滤器
    return handoff_filters.remove_all_tools(data)

triage_agent = Agent(
    name="前台",
    model=model,
    handoffs=[
        handoff(
            agent=specialist_agent,
            input_filter=clean_handoff_filter,  # 转接时过滤掉工具记录
        ),
    ],
)

HandoffInputData 包含三部分：

字段 说明
input_history Runner.run() 调用前的输入历史
pre_handoff_items 当前轮之前产生的消息
new_items 当前轮（触发转接的那一轮）产生的消息

你可以随意修改这些数据，返回过滤后的版本。常见场景：

双向转接：Agent 之间可以互转

默认的 handoff 是单向的：Triage 能转给专业 Agent，但专业 Agent 处理完就结束了，不会转回来。

但实际场景中，用户可能在和"技术支持"聊着聊着突然问了个账单问题。这时候你希望"技术支持"能把对话转回前台，让前台重新分诊。

做法很简单——给专业 Agent 也加上 handoffs：

# 先创建所有 Agent（暂时不设置 handoffs）
triage_agent = Agent(name="前台", model=model, instructions="...")
tech_agent = Agent(name="技术支持", model=model, instructions="...", tools=[submit_ticket])
billing_agent = Agent(name="账单客服", model=model, instructions="...", tools=[check_bill])

# 前台可以转给技术支持和账单客服
triage_agent.handoffs = [tech_agent, billing_agent]

# 技术支持和账单客服都可以转回前台
tech_agent.handoffs.append(triage_agent)
billing_agent.handoffs.append(triage_agent)

注意这里用 .handoffs.append() 是因为 Agent 之间存在循环引用（A 转给 B，B 转回 A），没法在构造函数里直接写。先创建对象，再互相添加引用就行。

形成的拓扑：

         +---------> tech_agent --------+
         |                              |
triage_agent <--------------------------+
         |                              |
         +---------> billing_agent -----+

用户可以这样对话：

用户: 我电脑登录不上  --> 前台转给技术支持
技术支持: 请问报什么错？
用户: 对了顺便帮我查一下这个月账单  --> 技术支持转回前台，前台再转给账单客服
账单客服: 您这个月消费 299 元...

完整可运行代码：中英日翻译分诊系统

把前面学到的所有知识点整合起来，做一个完整的例子：

"""
中英日翻译分诊系统
- 分诊 Agent：判断用户说的是什么语言，转给对应的翻译 Agent
- 翻译 Agent：把用户的话翻译成目标语言
- 支持双向转接：翻译 Agent 可以转回分诊 Agent
- 使用 on_handoff 回调记录转接日志
"""

import asyncio
from dataclasses import dataclass, field

from openai import AsyncOpenAI
from agents import (
    Agent,
    OpenAIChatCompletionsModel,
    Runner,
    RunContextWrapper,
    handoff,
    set_tracing_disabled,
)

set_tracing_disabled(True)

# 连接模型（换成你自己的服务地址）
client = AsyncOpenAI(base_url="http://localhost:8317/v1", api_key="sk-12345678")
model = OpenAIChatCompletionsModel(model="gpt-5.2", openai_client=client)


# ---- 共享上下文：记录转接历史 ----

@dataclass
class TranslationContext:
    """翻译系统的上下文"""
    source_language: str | None = None    # 用户使用的语言
    handoff_history: list[str] = field(default_factory=list)  # 转接记录


# ---- 转接回调 ----

async def on_handoff_to_chinese(ctx: RunContextWrapper[TranslationContext]) -> None:
    ctx.context.source_language = "中文"
    ctx.context.handoff_history.append("-> 中文翻译")

async def on_handoff_to_english(ctx: RunContextWrapper[TranslationContext]) -> None:
    ctx.context.source_language = "English"
    ctx.context.handoff_history.append("-> 英文翻译")

async def on_handoff_to_japanese(ctx: RunContextWrapper[TranslationContext]) -> None:
    ctx.context.source_language = "日本語"
    ctx.context.handoff_history.append("-> 日文翻译")


# ---- 专业翻译 Agent ----

chinese_translator = Agent[TranslationContext](
    name="中文翻译",
    handoff_description="当用户使用中文交流时，转给这个 Agent",
    instructions=(
        "你是中文翻译专员。用中文和用户交流。"
        "如果用户发了其他语言的内容，帮他翻译成中文。"
        "如果用户想翻译成其他语言，转回分诊台让它重新分配。"
    ),
    model=model,
)

english_translator = Agent[TranslationContext](
    name="英文翻译",
    handoff_description="当用户使用英文交流时，转给这个 Agent",
    instructions=(
        "You are an English translator. Communicate in English. "
        "If the user sends content in other languages, translate it to English. "
        "If the user wants to translate to another language, hand off back to the triage agent."
    ),
    model=model,
)

japanese_translator = Agent[TranslationContext](
    name="日文翻译",
    handoff_description="当用户使用日文交流时，转给这个 Agent",
    instructions=(
        "あなたは日本語翻訳者です。日本語でコミュニケーションしてください。"
        "ユーザーが他の言語でコンテンツを送信した場合、日本語に翻訳してください。"
        "ユーザーが他の言語に翻訳したい場合は、トリアージエージェントに戻してください。"
    ),
    model=model,
)


# ---- 分诊 Agent ----

triage_agent = Agent[TranslationContext](
    name="翻译分诊台",
    instructions=(
        "你是翻译分诊台。根据用户使用的语言，转给对应的翻译专员：\n"
        "- 中文 -> 中文翻译\n"
        "- English -> 英文翻译\n"
        "- 日本語 -> 日文翻译\n"
        "不要自己回答用户的问题，直接转接。"
    ),
    model=model,
    handoffs=[
        handoff(agent=chinese_translator, on_handoff=on_handoff_to_chinese),
        handoff(agent=english_translator, on_handoff=on_handoff_to_english),
        handoff(agent=japanese_translator, on_handoff=on_handoff_to_japanese),
    ],
)

# 双向转接：翻译 Agent 可以转回分诊台
chinese_translator.handoffs.append(triage_agent)
english_translator.handoffs.append(triage_agent)
japanese_translator.handoffs.append(triage_agent)


# ---- 运行测试 ----

async def main():
    context = TranslationContext()

    # 测试 1：中文输入
    print("=== 测试 1：中文 ===")
    result = await Runner.run(triage_agent, input="你好，请帮我翻译一句话", context=context)
    print(f"处理 Agent: {result.last_agent.name}")
    print(f"回复: {result.final_output}")
    print(f"转接记录: {context.handoff_history}")
    print()

    # 测试 2：英文输入
    print("=== 测试 2：English ===")
    result = await Runner.run(triage_agent, input="Hello, can you help me translate something?", context=context)
    print(f"处理 Agent: {result.last_agent.name}")
    print(f"回复: {result.final_output}")
    print(f"转接记录: {context.handoff_history}")
    print()

    # 测试 3：日文输入
    print("=== 测试 3：日本語 ===")
    result = await Runner.run(triage_agent, input="こんにちは、翻訳を手伝ってくれますか？", context=context)
    print(f"处理 Agent: {result.last_agent.name}")
    print(f"回复: {result.final_output}")
    print(f"转接记录: {context.handoff_history}")


if __name__ == "__main__":
    asyncio.run(main())

运行后你会看到类似这样的输出：

=== 测试 1：中文 ===
处理 Agent: 中文翻译
回复: 你好！当然可以，请告诉我你想翻译什么内容，以及想翻译成哪种语言？
转接记录: ['-> 中文翻译']

=== 测试 2：English ===
处理 Agent: 英文翻译
回复: Of course! What would you like me to translate?
转接记录: ['-> 中文翻译', '-> 英文翻译']

=== 测试 3：日本語 ===
处理 Agent: 日文翻译
回复: もちろん！何を翻訳しますか？
转接记录: ['-> 中文翻译', '-> 英文翻译', '-> 日文翻译']

注意 context.handoff_history 记录了每一次转接，这得益于 on_handoff 回调。

Handoff vs Agent-as-Tool：该用哪个？

你可能会想："我也可以把另一个 Agent 当工具调用啊，为啥要用 Handoff？"

没错，两种方式都可以实现多 Agent 协作，但它们的语义不同：

Handoff（转接） Agent-as-Tool（当工具用）
控制权 完全交出，新 Agent 接管对话 不交出，调用方仍在掌控
类比 转接电话 打内线问同事
对话上下文 新 Agent 看到完整历史 新 Agent 只看到传入的参数
适用场景 问题需要另一个专家全权处理 只需要从另一个 Agent 拿个结果
result.last_agent 变成新 Agent 还是原来的 Agent

简单的判断标准：

小结

这一章我们学到了：

核心思想就一句话：让专业的 Agent 干专业的事。

下一步

多 Agent 协作搞定了，但 Agent 可能会跑偏——用户问了不该问的问题，或者 Agent 回了不靠谱的答案怎么办？下一章我们学习护栏（Guardrails），给 Agent 装上安全刹车。

← 第2章 工具 第4章 护栏 →