第 4 章 Google ADKMulti-agentSub-agent

第 4 章 多智能体协作:从单兵到团队

第 4 章 多智能体协作:从单兵到团队

前面两章,我们的云销客服是一个"全能单兵"——一个 Agent 会查订单、会查政策、会聊天。但现实中的客服中心不是这样的:有人专门接电话、有人专门处理退款、有人专门盯物流。专业化分工,才能处理复杂业务。这一章,我们把云销客服从"单兵"改造成"团队"。

4.1 为什么需要多个 Agent

4.1.1 单 Agent 的极限

先看看我们的云销客服 v0.2 有什么问题。

它现在能做的:查订单、查退款政策、记住会话。但当你真正运营一个客服系统,会发现"一个 Agent 包打天下"有三个致命问题:

第一,指令越来越长,越来越互相冲突。

要给一个 Agent 加"查订单 + 查退款 + 查物流 + 处理投诉 + 推荐商品"的指令,这个 prompt 会膨胀到失控。不同业务的规则开始打架:"退款要礼貌"和"处理投诉要强硬"放一起,模型会困惑。

第二,一个模型做所有事,既贵又慢。

客服场景里,查物流这种简单任务用便宜的小模型就够了,而处理复杂投诉需要强模型。一个 Agent 只能用一种模型——你要么所有请求都用贵的模型(浪费),要么都用便宜的(复杂任务质量差)。

第三,职责不清,难维护、难评估。

所有逻辑揉在一个 Agent 里,出了问题你分不清是"查订单的逻辑错了"还是"回答话术错了"。评估和迭代都无从下手。

4.1.2 团队化:专业化分工

解决办法是团队化——把一个"全能 Agent"拆成多个"专业 Agent",各管一摊,再由一个"协调者"来调度。

这就是多智能体(Multi-agent)架构。它的好处:

  • 指令精简:每个子 Agent 只管一件事,prompt 短而清晰
  • 模型按需:简单的子 Agent 用便宜模型,复杂的用贵模型,成本最优
  • 职责清晰:每个 Agent 可以独立开发、测试、评估、迭代
  • 可扩展:加一个新业务,就加一个子 Agent,不动其他 Agent

4.2 Coordinator 协调者模式:ADK 的多智能体骨架

4.2.1 核心思想

ADK 的多智能体架构,核心就是协调者模式(Coordinator Pattern)

一个父 Agent(协调者)负责任务的理解和分配;多个子 Agent(Sub-agent)各自负责一个专业领域。父 Agent 把任务委派给合适的子 Agent,子 Agent 完成后把结果交回来。

这个模式最关键的一点:谁来决定"这个任务该给谁"? 答案是——父 Agent 的 LLM。ADK 会自动给父 Agent 注入"委托工具"(delegation tools),每个子 Agent 对应一个委托工具。父 Agent 的模型通过调用这些工具来决定把任务交给谁。

4.2.2 最小多智能体示例

一个父 Agent 带两个子 Agent,代码极其直观:

from google.adk import Agent


def get_weather(city: str) -> str:
    """Returns the current weather for a city."""
    return f"Weather in {city}: sunny, 22C"


def search_flights(origin: str, destination: str) -> str:
    """Searches for available flights."""
    return f"3 flights from {origin} to {destination}"


weather_agent = Agent(
    name="weather_checker",
    mode="single_turn",   # 子 Agent:不需要和用户交互
    tools=[get_weather],
)

flight_agent = Agent(
    name="flight_booker",
    mode="task",          # 子 Agent:可以问用户问题
    tools=[search_flights],
)

root_agent = Agent(
    name="travel_planner",   # 父 Agent(协调者)
    model="gemini-flash-latest",
    sub_agents=[weather_agent, flight_agent],
    # ADK 自动注入委托工具:weather_checker, flight_booker
)

注意看,父 Agent 通过 sub_agents=[...] 声明子 Agent,然后什么都不用干了。ADK 自动做了这些事:

  1. 为每个子 Agent 生成一个同名的委托工具(weather_checkerflight_booker
  2. 把委托工具加进父 Agent 的工具列表
  3. 当父 Agent 的 LLM 决定"这个任务该天气 Agent 处理"时,它调用 weather_checker 工具
  4. 控制权转到子 Agent,子 Agent 执行完毕后自动返回结果

这个"自动注入委托工具"的设计非常聪明——父 Agent 不需要写任何调度代码,调度是模型基于对子 Agent 的 description 的理解自然完成的。

4.3 三种协作模式:子 Agent 的性格

子 Agent 有一个关键参数 mode,它决定了子 Agent 和用户交互的方式。三种模式,三种性格:

4.3.1 mode="chat":需要深度对话的子 Agent

这是默认模式。子 Agent 可以跟用户完整对话,像一个小型 Agent 一样。它有两个特点:

  • 可以主动和用户多轮交流
  • 任务完成后,必须显式调用 transfer_to_agent 才能把控制权还给父 Agent

适用场景:需要跟用户充分沟通的业务,比如"处理投诉"这种需要来回对话的。

complaint_agent = Agent(
    name="complaint_handler",
    mode="chat",   # 默认,可省
    instruction="你是投诉处理专员,耐心倾听用户投诉,必要时转回主客服。",
)

4.3.2 mode="task":干活为主、偶尔澄清的子 Agent

task 模式的子 Agent 以"完成任务"为目标,只在必要时问用户澄清问题。它的特点:

  • 任务完成后自动返回父 Agent(通过内部工具 finish_task
  • 可以通过 input_schema / output_schema 定义输入输出的结构契约
  • 上下文隔离:task 子 Agent 在独立的会话分支里运行

适用场景:有明确任务边界的业务,比如"查订单详情"。

from pydantic import BaseModel

class OrderQuery(BaseModel):
    order_id: str

class OrderResult(BaseModel):
    order_id: str
    status: str
    logistics: str

order_agent = Agent(
    name="order_query_agent",
    mode="task",
    input_schema=OrderQuery,
    output_schema=OrderResult,
    tools=[get_order_status],
)

注意 task 模式的限制:task 模式的 Agent 必须是叶子节点,不能再有子 Agent(不能嵌套)。另外在 ADK Python 2.0.0 中,task 模式在基于图的工作流里暂时被禁用(后面第 5 章会提到)。

4.3.3 mode="single_turn":纯执行、不交互的子 Agent

single_turn 模式是最轻量的——子 Agent 只执行一轮,不跟用户交互,完成后自动返回。它的特点:

  • 无用户交互,执行快
  • 可以并行执行(多个 single_turn 子 Agent 同时跑)
  • 天然适合"查天气、算价格"这种纯函数式子任务
price_agent = Agent(
    name="price_calculator",
    mode="single_turn",
    tools=[calc_total_price],
)

4.3.4 三种模式怎么选

模式 用户交互 返回方式 适用场景
chat 完整多轮对话 手动 transfer_to_agent 投诉、咨询等深度交流
task 必要时澄清 自动 finish_task 有明确任务边界的业务
single_turn 自动返回 纯执行子任务、可并行

一句话记忆:chat 是"聊",task 是"干",single_turn 是"秒干完"

4.4 主线项目:云销客服 v1.0 —— 三 Agent 团队

4.4.1 团队设计

现在我们把云销客服升级到 v1.0,设计成三 Agent 团队:

                    ┌─────────────────┐
                    │  yunxiao_cs      │  协调者(父 Agent)
                    │  (Coordinator)   │  理解用户意图,分派任务
                    └────────┬────────┘
          ┌──────────────────┼──────────────────┐
          ▼                  ▼                  ▼
  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
  │ order_agent  │  │ refund_agent │  │ logistics    │
  │ 订单查询专员  │  │ 退款处理专员  │  │ 物流查询专员  │
  └──────────────┘  └──────────────┘  └──────────────┘
  • 协调者yunxiao_cs):听懂用户要什么,分派给对应专员
  • 订单专员order_agent):查订单状态,task 模式
  • 退款专员refund_agent):查退款政策、处理退款咨询,task 模式
  • 物流专员logistics_agent):查物流详情,single_turn 模式(纯查询,可并行)

4.4.2 完整代码

from google.adk import Agent
from pydantic import BaseModel


# ---------- 共享工具 ----------
def get_order_status(order_id: str) -> dict:
    """查询订单当前状态。"""
    mock_orders = {
        "ORD-20260901-001": {"status": "已发货", "logistics": "顺丰 SF1234567890"},
        "ORD-20260901-002": {"status": "待发货", "logistics": "仓库备货中"},
        "ORD-20260903-003": {"status": "已签收", "logistics": "已于 9 月 6 日签收"},
    }
    order = mock_orders.get(order_id)
    if order:
        return {"status": "success", "order_id": order_id, **order}
    return {"status": "error", "order_id": order_id, "message": f"未找到订单 {order_id}"}


def get_refund_policy(product_category: str) -> dict:
    """查询某类商品的退款政策。"""
    policies = {
        "electronics": {"window": "7 天", "condition": "商品未拆封"},
        "home": {"window": "15 天", "condition": "商品及包装完好"},
    }
    policy = policies.get(product_category)
    if policy:
        return {"status": "success", "product_category": product_category, **policy}
    return {"status": "error", "message": f"未知商品类别 {product_category}"}


def get_logistics(tracking_no: str) -> dict:
    """查询物流轨迹。"""
    mock_logistics = {
        "SF1234567890": {"current": "广州转运中心", "history": ["深圳已揽收", "广州转运中心"]},
        "SF1234567891": {"current": "派送中", "history": ["已到达北京", "派送中"]},
    }
    info = mock_logistics.get(tracking_no)
    if info:
        return {"status": "success", "tracking_no": tracking_no, **info}
    return {"status": "error", "message": f"物流单号 {tracking_no} 未找到"}


# ---------- 输入/输出契约 ----------
class OrderQuery(BaseModel):
    order_id: str


class OrderResult(BaseModel):
    order_id: str
    status: str
    logistics: str


# ---------- 子 Agent ----------
order_agent = Agent(
    name="order_agent",
    mode="task",
    description="订单查询专员,负责查询订单状态。当用户询问订单状态、订单到哪了时,转给我。",
    input_schema=OrderQuery,
    output_schema=OrderResult,
    tools=[get_order_status],
)

refund_agent = Agent(
    name="refund_agent",
    mode="task",
    description="退款处理专员,负责退款政策咨询。当用户询问能否退款、退款政策、退货流程时,转给我。",
    tools=[get_refund_policy],
)

logistics_agent = Agent(
    name="logistics_agent",
    mode="single_turn",
    description="物流查询专员,负责查询快递物流轨迹。当用户给出快递单号询问物流时,转给我。",
    tools=[get_logistics],
)


# ---------- 协调者(父 Agent) ----------
root_agent = Agent(
    name="yunxiao_cs",
    model="gemini-flash-latest",
    description="云销电商客服团队协调者。",
    instruction=(
        "你是云销电商客服团队的主管。你的工作是听懂用户的诉求,"
        "然后把任务分配给最合适的专员。\n"
        "规则:\n"
        "1. 询问订单状态 → 转给 order_agent\n"
        "2. 询问退款/退货政策 → 转给 refund_agent\n"
        "3. 询问物流轨迹(有快递单号)→ 转给 logistics_agent\n"
        "4. 如果用户诉求不明确,先向用户澄清,不要随意分派\n"
    ),
    sub_agents=[order_agent, refund_agent, logistics_agent],
)

4.4.3 关键设计点

第一,每个子 Agent 都写了自己的 description 这不是装饰——父 Agent 的 LLM 靠 description 来理解"这个子 Agent 是干什么的、什么时候该转给它"。description 写得越清楚,分派越准。 这是多智能体系统里最重要的"元数据"。

第二,任务明确的子 Agent 用 task 模式,纯查询用 single_turn。 order_agent 需要输入订单号(可能缺,需要澄清),所以 task 模式;logistics_agent 拿到单号就查,single_turn 模式。

第三,父 Agent 的指令里明确写了分派规则。 虽然模型能自己判断,但显式的规则("询问退款 → 转 refund_agent")能大幅提升分派准确率。指令 + description 双保险。

4.4.4 跑起来

adk runadk web 跑,试试对话:

You: 帮我看看 ORD-20260901-001 到哪了
Agent: (协调者判断是订单查询 → 转 order_agent)
       订单 ORD-20260901-001 已发货,物流:顺丰 SF1234567890。

You: 我想退这个电子产品,能退吗?
Agent: (协调者判断是退款政策 → 转 refund_agent)
       电子产品支持 7 天无理由退货,条件是商品未拆封、不影响二次销售。

You: 快递单号 SF1234567890 现在到哪了?
Agent: (协调者判断是物流查询 → 转 logistics_agent)
       您的包裹当前在「广州转运中心」,轨迹:深圳已揽收 → 广州转运中心。

每个问题都被协调者精准分派给了对应专员。这就是团队化。

4.5 顺序、并行、路由:三种协作语义

多智能体协作不止"协调者分派"这一种形态。ADK 还支持更细粒度的协作语义。我们逐个看。

4.5.1 顺序协作(Sequential)

任务必须一步一步来,前一个 Agent 的输出是后一个 Agent 的输入。比如:订单专员查完订单 → 退款专员根据订单判断可退金额。

# 概念示意:A 完成 → B 用 A 的结果继续
sequential_workflow = Workflow(
    name="sequential_flow",
    edges=[
        ("START", order_agent, refund_calc_agent, completed)
    ],
)

顺序协作适合有严格依赖关系的流水线任务。(完整的工作流 API 见第 5 章。)

4.5.2 并行协作(Parallel)

多个独立子任务同时执行,互不等待。比如同时查订单状态、查物流、查退款政策,最后汇总。

single_turn 模式的子 Agent 天然支持并行——它们不需要用户交互,可以同时跑。这能显著降低总延迟。

4.5.3 路由协作(Routing)

根据条件把任务分发到不同分支,这是"if/else"在 Agent 世界的表达。ADK 的路由通过 router 函数返回 Event(route=...) 实现(第 5 章详述)。

4.5.4 三种语义怎么选

语义 依赖关系 典型场景
顺序 强依赖 流水线:查 → 算 → 办
并行 无依赖 同时查多项信息再汇总
路由 条件分支 按类型分发到不同处理分支

4.6 什么时候该用多智能体,什么时候不该

多智能体很强大,但它不是银弹。用错了,比单 Agent 更糟。这一节我们诚实讨论边界。

4.6.1 该用多智能体的信号

  • 业务天然分领域:订单、退款、物流就是三个领域,拆分后每个指令清晰
  • 需要不同模型:简单子任务想用便宜模型,复杂任务用强模型
  • 需要独立迭代:退款流程经常改,订单查询很少改,拆开后互不影响
  • 指令已膨胀到冲突:一个 Agent 的 prompt 超过一定长度且规则打架时,该拆了

4.6.2 不该用多智能体的信号

  • 任务简单:一个函数就能搞定的事,拆成三个 Agent 是杀鸡用牛刀
  • 过度拆分:把一个连贯的对话拆成无数碎片 Agent,模型在"该转给谁"上反复横跳,延迟和成本暴增,准确率反而下降
  • 无清晰边界:几个子 Agent 的职责互相重叠,分派靠猜
  • 调试成本超过收益:多 Agent 系统的事件流更复杂,如果团队还小、没有可观测性工具,先别急着拆

4.6.3 反模式:为拆而拆

一个真实的教训:有人把一个"帮用户写邮件"的 Agent 拆成"理解意图 Agent"+"起草 Agent"+"校对 Agent"三个。结果每个环节都要过一遍 LLM,一封邮件多花 3 倍 token,而质量没提升——因为"起草"和"校对"本来就是同一个模型几分钟内能完成的事。

多智能体的价值在于"专业化分工带来的边界清晰",不在于"把 Agent 变多"。当拆分让每个 Agent 的职责更简单、更聚焦时,拆;当拆分只是增加调用链时,别拆。

「为什么 ADK 这样设计」:Agent 通信 vs 直接调函数

一个很自然的问题:子 Agent 之间通信,为什么不直接用 Python 函数调用?父 Agent 直接 result = order_agent.run(...) 不就行了?

这就是 ADK 设计上很值得玩味的一个点。我们来对比两条路:

路线一:Agent 直接调函数(过程式)

  • 父代码里写死:if 是订单: order_agent.run(query)
  • 优点:确定性强、没有 LLM 参与调度
  • 致命问题:调度逻辑硬编码了。用户说"我想知道我的东西到哪了",这句话到底是订单还是物流?"东西到哪了"既可能是订单状态也可能是物流轨迹。硬编码的 if/else 无法处理这种语义模糊——只有 LLM 能理解自然语言背后的意图。

路线二:LLM 驱动的委托(ADK 的做法)

  • 父 Agent 的 LLM 通过委托工具决定分派
  • 优点:意图理解交给 LLM,执行交给子 Agent。模型理解"东西到哪了"→ 判断是物流 → 调 logistics_agent 工具
  • 代价:调度这一步要花 LLM 的 token,且可能分派错

ADK 的选择是"意图理解智能,执行确定"——用 LLM 处理语义模糊的"分派决策",用子 Agent 的专业工具处理"确定执行"。这正是我们第 1 章反复强调的哲学:该让 LLM 判断的地方给 LLM,该确定执行的地方用代码。多智能体的"协调者 + 子 Agent"就是这个哲学在架构层面的体现。

而如果某条路径完全确定(比如 A 做完必须 B 做),你不需要 LLM 调度——用图工作流的确定性边(第 5 章)反而更好。多智能体处理"模糊分发",图工作流处理"确定流程",两者互补,这正是 ADK 2.0 的核心设计。

本章小结

  • 多智能体的价值:专业化分工让指令精简、模型按需、职责清晰、易于迭代
  • Coordinator 模式:父 Agent 通过 sub_agents=[...] 声明子 Agent,ADK 自动注入委托工具
  • 三种协作模式chat(深度对话,手动返回)、task(干活为主,自动返回)、single_turn(纯执行,可并行)
  • 子 Agent 的 description 是分派的关键:父 Agent 的 LLM 靠它理解"该把任务给谁"
  • 三种协作语义:顺序(强依赖)、并行(无依赖)、路由(条件分支)
  • 不是所有场景都该用多智能体:简单任务、职责重叠、为拆而拆,都是反模式
  • ADK 的哲学:意图理解智能(LLM 调度),执行确定(子 Agent 专业工具)

练习

  1. 给团队加第四个专员:给云销客服加一个"投诉处理专员"(用 chat 模式,因为它需要和用户深度对话),并在协调者指令里加一条分派规则。
  2. 观察分派质量:故意用模糊的说法提问("我的东西呢?""我要退东西"),看协调者能不能正确判断该转给谁。如果分派错了,是 description 的问题还是指令的问题?
  3. 对比单 Agent vs 多 Agent:用一个复杂的多步请求("查订单 001 的状态,再查它能不能退")分别发给 v0.2 的单 Agent 和 v1.0 的团队,对比回答质量和延迟。

下一章预告:第 5 章,我们进入 ADK 2.0 最核心的部分——图工作流。当业务流程有明确顺序和分支时(比如退货审批),依赖 LLM 自由分派是不够的,我们需要"确定性"的力量。