跳转至

第5章 Agent 间交互

学习目标

  • 理解 Handoff 任务转移机制的原理与实现
  • 掌握 Supervisor 协调模式的设计方法
  • 了解 Agent as Tool 模式及其适用场景
  • 学会多 Agent 间的消息传递与协作协议
  • 能够为复杂业务设计多 Agent 协作系统

问题导入

某公司的客服系统让单个 Agent 包揽一切：售前咨询、售后退换、技术支持、投诉处理。结果 Agent 的 system prompt 越写越长（超过 2000 字），工具堆到 30 个，模型经常选错工具、答非所问，且任何调整都牵一发动全身。

问题根因：单 Agent 承载了过多职责。就像让一个人同时当销售、客服、工程师、法务，必然做不好。解决方案是多 Agent 协作——把职责拆分到专门 Agent，通过 Handoff、Supervisor、Agent as Tool 等模式让它们协同。本章讲解 Agent 间交互的主流模式与实现。

核心概念

1. Handoff 任务转移

Handoff 是 OpenAI Agents SDK 的原生能力：一个 Agent 在执行中判断"这个任务不归我管"，把整个会话连同上下文转移给另一个更合适的 Agent。

agent = Agent(name="前台", handoffs=[售后Agent, 技术Agent])

特点：①控制权完全移交，接收 Agent 接管后续全部对话；②上下文传递，新 Agent 能看到历史；③用户无感，体验上是"一个助手"。适合"路由分发"场景。

2. Supervisor 协调模式

Supervisor（主管）是一个协调者 Agent，负责任务分配与结果汇总：

  • 接收用户请求
  • 决定分派给哪个子 Agent
  • 收集子 Agent 结果
  • 必要时多轮调度
  • 汇总输出最终答案

与 Handoff 的区别：Supervisor 不移交控制权，而是"指挥"子 Agent 干活后回来汇报。适合需要综合多个专业 Agent 结果的场景。

3. Agent as Tool 模式

把一个 Agent 包装成工具，供另一个 Agent 调用：

@function_tool
def ask_analyst(question: str) -> str:
    # 内部运行分析 Agent 并返回结果
    return Runner.run_sync(analyst_agent, question).final_output

特点：①调用方主动控制何时调用、传什么参数；②被调用 Agent 的上下文不外泄，只返回结果；③可并行调用多个 Agent 工具。适合"按需咨询专家"场景。

4. 消息传递

多 Agent 间共享信息的方式：

  • 上下文共享：Handoff 时传递完整历史
  • 结构化消息：Agent 间传递 JSON 消息（如 {task, context, deadline}）
  • 共享黑板：多个 Agent 读写同一块共享状态（如 Redis 中的任务看板）

5. 协作协议

为避免多 Agent 协作混乱，需约定协议：

  • 任务格式：统一的任务描述结构（目标、输入、约束）
  • 结果格式：统一的结果结构（状态、数据、置信度）
  • 错误传播：子 Agent 失败时如何上报、Supervisor 如何降级
  • 终止条件：何时认为协作完成

原理图

flowchart TB
    User[用户请求] --> Triage[分诊 Agent]

    subgraph Handoff模式
        Triage -->|售前问题| Sales[售前 Agent]
        Triage -->|售后问题| After[售后 Agent]
        Triage -->|技术问题| Tech[技术 Agent]
        Sales --> Resp1[直接回复用户]
        After --> Resp2[直接回复用户]
        Tech --> Resp3[直接回复用户]
    end

    subgraph Supervisor模式
        Sup[Supervisor] -->|派发| A1[分析 Agent]
        Sup -->|派发| A2[写作 Agent]
        A1 -->|结果| Sup
        A2 -->|结果| Sup
        Sup --> Final[汇总输出]
    end

    subgraph AgentAsTool模式
        Main[主 Agent] -->|调用工具| Tool1[专家Agent1]
        Main -->|调用工具| Tool2[专家Agent2]
        Tool1 --> Main
        Tool2 --> Main
        Main --> Out[输出]
    end

    style Triage fill:#fff9c4
    style Sup fill:#bbdefb
    style Main fill:#c8e6c9

图：三种多 Agent 交互模式——Handoff 移交控制权、Supervisor 指挥汇报、Agent as Tool 按需调用

示例代码

1. Handoff 任务转移

"""
第5章示例：Handoff 任务转移
分诊 Agent 根据问题类型把会话转移给专门 Agent
依赖：pip install openai-agents
"""
import asyncio
from agents import Agent, Runner


# ========== 专门 Agent ==========
sales_agent = Agent(
    name="售前顾问",
    instructions=(
        "你是售前顾问，擅长产品介绍、价格咨询、方案推荐。"
        "只回答售前问题，若超出范围请说明。"
    ),
    model="gpt-4o-mini",
)

after_sales_agent = Agent(
    name="售后专员",
    instructions=(
        "你是售后专员，处理退换货、物流查询、发票问题。"
        "需要订单号才能查询，若用户未提供请主动询问。"
    ),
    model="gpt-4o-mini",
)

tech_agent = Agent(
    name="技术支持",
    instructions=(
        "你是技术支持工程师，解决产品使用问题、故障排查。"
        "给出清晰的排查步骤，必要时引导联系人工。"
    ),
    model="gpt-4o-mini",
)


# ========== 分诊 Agent（Handoff 入口）==========
triage_agent = Agent(
    name="客服分诊",
    instructions=(
        "你是客服分诊助手。判断用户问题类型并移交给对应 Agent：\n"
        "- 售前咨询（产品、价格、方案）→ 售前顾问\n"
        "- 售后问题（退换货、物流、发票）→ 售后专员\n"
        "- 技术问题（使用故障、排查）→ 技术支持\n"
        "移交时不要回答问题，直接转交。"
    ),
    handoffs=[sales_agent, after_sales_agent, tech_agent],
    model="gpt-4o-mini",
)


async def main():
    questions = [
        "你们的旗舰版多少钱？支持多少人？",       # → 售前
        "订单 A1001 还没收到，要退款",            # → 售后
        "设备开机后屏幕闪烁怎么办？",             # → 技术
    ]
    for q in questions:
        print(f"\n用户：{q}")
        result = await Runner.run(triage_agent, q)
        print(f"回复：{result.final_output}")
        # 查看实际移交给了谁
        for item in result.raw_responses:
            if hasattr(item, "output") and item.output:
                break


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

运行结果示例：

用户：你们的旗舰版多少钱？支持多少人？
回复：旗舰版售价 9999 元/年，支持最多 50 人使用...

用户：订单 A1001 还没收到，要退款
回复：请提供您的订单号，我为您查询物流并处理退款...

用户：设备开机后屏幕闪烁怎么办？
回复：建议按以下步骤排查：1.检查电源连接；2.更新显卡驱动...

2. Supervisor 协调模式

"""
Supervisor 模式：协调者派发任务给子 Agent 并汇总
"""
import asyncio
from agents import Agent, Runner, function_tool


# 子 Agent：数据分析专家
analyst_agent = Agent(
    name="数据分析师",
    instructions="你是数据分析师，给出简洁的数据洞察结论。",
    model="gpt-4o-mini",
)

# 子 Agent：文案专家
writer_agent = Agent(
    name="文案",
    instructions="你是营销文案，根据给定的洞察写出 100 字以内的推广文案。",
    model="gpt-4o-mini",
)


# 把子 Agent 包装成工具
@function_tool
async def analyze_data(data_desc: str) -> str:
    """让数据分析师分析数据并返回洞察。

    Args:
        data_desc: 数据描述
    """
    result = await Runner.run(analyst_agent, f"分析以下数据：{data_desc}")
    return result.final_output


@function_tool
async def write_copy(insight: str) -> str:
    """让文案根据洞察写出推广文案。

    Args:
        insight: 数据洞察
    """
    result = await Runner.run(writer_agent, f"基于此洞察写文案：{insight}")
    return result.final_output


# Supervisor Agent
supervisor = Agent(
    name="营销总监",
    instructions=(
        "你是营销总监。收到用户的营销需求后："
        "1) 调用 analyze_data 分析数据得到洞察；"
        "2) 调用 write_copy 基于洞察写文案；"
        "3) 汇总分析结论和文案，给用户最终方案。"
    ),
    tools=[analyze_data, write_copy],
    model="gpt-4o-mini",
)


async def main():
    result = await Runner.run(
        supervisor,
        "我们的能量饮料上月销量涨了 30%，主要在 18-25 岁人群，帮我做一波推广。",
        max_turns=8,
    )
    print("总监方案：", result.final_output)


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

3. Agent as Tool 并行咨询

"""
Agent as Tool：主 Agent 并行调用多个专家 Agent
"""
import asyncio
from agents import Agent, Runner, function_tool


legal_agent = Agent(name="法务", instructions="你是法务专家，给出合规建议。", model="gpt-4o-mini")
finance_agent = Agent(name="财务", instructions="你是财务专家，给出成本与税务建议。", model="gpt-4o-mini")
ops_agent = Agent(name="运营", instructions="你是运营专家，给出执行流程建议。", model="gpt-4o-mini")


@function_tool
async def consult_legal(question: str) -> str:
    """咨询法务专家。Args: question: 法务问题"""
    r = await Runner.run(legal_agent, question)
    return r.final_output


@function_tool
async def consult_finance(question: str) -> str:
    """咨询财务专家。Args: question: 财务问题"""
    r = await Runner.run(finance_agent, question)
    return r.final_output


@function_tool
async def consult_ops(question: str) -> str:
    """咨询运营专家。Args: question: 运营问题"""
    r = await Runner.run(ops_agent, question)
    return r.final_output


pm_agent = Agent(
    name="项目经理",
    instructions=(
        "你是项目经理。对复杂问题，并行咨询法务、财务、运营三位专家，"
        "汇总三方意见给出综合方案。一次发起所有咨询以节省时间。"
    ),
    tools=[consult_legal, consult_finance, consult_ops],
    model="gpt-4o-mini",
)


async def main():
    result = await Runner.run(
        pm_agent,
        "我们要在上海新开一家线下门店，需要评估合规、成本和运营流程。",
        max_turns=6,
    )
    print(result.final_output)


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

企业案例

案例 1：多 Agent 协作客服系统

某 SaaS 公司用 Handoff 重构客服：分诊 Agent 做意图识别，移交给售前/售后/技术三个专门 Agent。每个专门 Agent 只管自己领域的 5-8 个工具，prompt 缩短到 300 字内。上线后工具误调率从 18% 降到 4%，平均解决时长下降 40%。关键经验：①分诊 Agent 要轻量，只做路由不答题；②Handoff 时传递用户已提供信息避免重复询问；③专门 Agent 之间不直接通信，都通过分诊中转，避免耦合。

案例 2：Supervisor 模式做内容生产

某内容平台用 Supervisor 协调"选题 Agent + 资料检索 Agent + 写作 Agent + 校对 Agent"四个子 Agent 生产长文。Supervisor 按依赖调度：选题→检索→写作→校对，每步结果回传校验，不达标则重做。日均产出从人工 5 篇提升到 50 篇，人工只需终审。

常见问题

Q1：Handoff 和 Supervisor 怎么选？

A：用户问题归属单一领域用 Handoff（售前 or 售后）；需综合多方信息用 Supervisor（分析+写作+校对）。Handoff 体验更顺滑，Supervisor 控制更精细。

Q2：Agent as Tool 和直接调用函数有什么区别？

A：Agent as Tool 被调用的是另一个有推理能力的 Agent，能处理复杂子任务；普通工具是确定性函数。Agent as Tool 适合"需要 LLM 思考的子任务"，但成本更高、延迟更长。

Q3：多 Agent 会不会导致响应很慢？

A：会，尤其串行调度时。优化手段：①无依赖的子 Agent 并行调用（asyncio.gather）；②子 Agent 用小模型降低延迟；③合理设置 max_turns 避免空转；④缓存子 Agent 结果。

Q4：多 Agent 协作如何保证一致性？

A：①统一协作协议（任务/结果格式）；②Supervisor 做结果校验，不达标重做；③共享上下文（如 Session）让各 Agent 看到相同前提；④关键决策点人工审核。

Q5：多 Agent 系统怎么调试？

A：①全链路追踪，记录每个 Agent 的输入输出耗时；②先单 Agent 测试通过再组合；③用固定输入做回归测试；④Supervisor 模式下逐步排查是哪个子 Agent 出错。

面试题

  1. 简答题：Handoff 模式的控制权转移与 Supervisor 模式的指挥调度有何区别？

参考要点：Handoff 完全移交控制权，接收 Agent 接管后续；Supervisor 不移交，派发任务并收集结果，控制权始终在 Supervisor。

  1. 设计题：设计一个"旅行规划 Agent 系统"，包含分诊、行程、酒店、交通多个 Agent，说明用哪种交互模式。

参考要点：分诊用 Handoff 路由；行程规划用 Supervisor 协调酒店/交通子 Agent 汇总；或 Agent as Tool 让行程 Agent 调用各专家。

  1. 实现题：如何让主 Agent 并行调用 3 个专家 Agent 工具？写出关键思路。

参考要点：三个专家都包装为 async function_tool，主 Agent prompt 鼓励一次性发起，SDK 自动并行 await。

  1. 原理题：为什么专门 Agent 比全能 Agent 效果更好？

参考要点：prompt 短聚焦、工具少误调率低、可独立优化、职责清晰易维护。

  1. 对比题：Agent as Tool 与 Handoff 在使用场景上的差异。

参考要点：Agent as Tool 主控方主动按需调用、只取结果；Handoff 移交全部控制权、接收方接管对话。

练习题

  1. Handoff 实现：实现一个"IT 支持"分诊 Agent，能 Handoff 到网络、系统、账号三个专门 Agent。

  2. Supervisor 系统：用 Supervisor 模式实现"周报生成 Agent"，协调数据收集、分析、撰写三个子 Agent。

  3. 并行优化：在练习 2 基础上，让数据收集与分析无依赖部分并行执行，对比串行与并行耗时。

小结

多 Agent 交互让系统从"一人全能"进化为"专业分工协作"。本章核心：Handoff 适合路由分发，移交控制权让专门 Agent 接管；Supervisor 适合综合调度，指挥子 Agent 干活并汇总；Agent as Tool 适合按需咨询专家，可并行调用；协作需统一协议保证一致性。多 Agent 系统的难点在于调度与调试，全链路追踪是关键。本卷至此完成 Agent 开发基础，后续卷将深入 LangGraph、MCP 等进阶框架。