多 Agent 协作 · Agent 调用 Agent

一、三种协作模式

二、Programmatic:最简单的"多 Agent"

说白了就是你用 Python 的循环/条件,按序调用 Agent。大多数需求这个就够——不要一上来就搞多 Agent 复杂编排。

async def research_and_write(topic: str) -> str:
    # Step 1: 研究员 Agent 收集要点
    research_result = await researcher.run(f"收集关于 {topic} 的 5 个要点")

    # Step 2: 基于要点让写作 Agent 写稿
    draft = await writer.run(
        f"基于以下要点写 500 字:\n{research_result.output}"
    )

    # Step 3: 让审校 Agent 改稿
    final = await editor.run(
        f"审校并改写成更精炼的版本:\n{draft.output}"
    )
    return final.output

没有任何框架魔法——就是三个普通的 await。好处:

• 流程清晰,像普通函数一样读
• 每一步的输入输出都是 Python 类型
• 测试时三个 Agent 可以各自 mock
• 错误定位容易——哪个 Agent 出问题,stacktrace 直接指

三、Delegate:让 Agent 把 Agent 当工具调

有时候决策逻辑太复杂,不适合你手写——比如"简单问题用 mini 模型答,复杂问题调专家 Agent"这个判断本身,LLM 比你的代码更会做。

这时候 Delegate 模式登场:把子 Agent 包装成一个工具,注册给父 Agent。

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

# ── 专家 Agent(子) ──
expert = Agent(
    "openai:gpt-4o",   # 贵的模型
    system_prompt="你是资深 Python 专家,回答需要极深入细节。",
)

# ── 前台 Agent(父) ──
frontdesk = Agent(
    "openai:gpt-4o-mini",   # 便宜模型
    system_prompt=(
        "你是技术前台。简单问题自己回答。"
        "复杂/深度问题调用 ask_expert 工具。"
    ),
)

@frontdesk.tool
async def ask_expert(ctx: RunContext[None], question: str) -> str:
    """问一个资深 Python 专家——适用于深入细节问题。"""
    r = await expert.run(
        question,
        usage=ctx.usage,     # ★ 把父 Agent 的 usage 传下去
    )
    return r.output

async def main():
    r = await frontdesk.run("GIL 在 Python 3.13 的 free-threaded 模式下有什么区别?")
    print(r.output)
    print("总用量(父+子):", r.usage())

关键点是 usage=ctx.usage——把父 Agent 的 RunUsage 对象传给子 Agent,这样子 Agent 的 token 会累加到父 Agent 的统计里。成本追踪才能正确。

四、Handoff:交接后不回来

Delegate 模式是父 Agent 调用子 Agent,子 Agent 返回后父继续——父始终在控制。

Handoff 模式是父 Agent 说"这事不归我,交给客服二线处理",交接后父就退出了——控制权彻底转移。

Pydantic AI 没有专门的"Handoff"原语——它靠结构化输出 + 业务层路由实现:

from typing import Literal
from pydantic import BaseModel

class RouteDecision(BaseModel):
    """路由器的决策结果:要么自己答,要么交接。"""
    kind: Literal["answer", "handoff"]
    answer: str | None = None
    handoff_to: Literal["billing", "tech", "legal"] | None = None
    handoff_reason: str | None = None

router = Agent(
    "openai:gpt-4o-mini",
    output_type=RouteDecision,
    system_prompt=(
        "你是通用客服。能答就自己答(kind=answer),"
        "涉及账单/技术/法律专业问题则交接(kind=handoff,选对应 handoff_to)。"
    ),
)

billing = Agent("openai:gpt-4o", system_prompt="你是账单专员,权限可以查流水、做退款。")
tech = Agent("openai:gpt-4o", system_prompt="你是技术支持,精通产品内部架构和排障。")
legal = Agent("openai:gpt-4o", system_prompt="你是法务,负责合规、协议、投诉处理。")

async def handle(q: str) -> str:
    r = await router.run(q)
    match r.output:
        case RouteDecision(kind="answer", answer=a):
            return a
        case RouteDecision(kind="handoff", handoff_to=dest):
            specialist = {"billing": billing, "tech": tech, "legal": legal}[dest]
            r2 = await specialist.run(q)
            return r2.output

这种"决策用结构化输出,业务代码做路由"的模式,比把 Agent 绑死在一起更灵活——随时能调整 handoff 的条件、加新专家、换模型。

五、共享 Usage:跨 Agent 成本追踪

多 Agent 协作时,成本追踪不能丢。RunUsage 是一个可累加的对象:

from pydantic_ai.usage import RunUsage

shared_usage = RunUsage()

r1 = await agent_a.run("...", usage=shared_usage)
r2 = await agent_b.run("...", usage=shared_usage)
r3 = await agent_c.run("...", usage=shared_usage)

print(shared_usage)
# RunUsage(requests=3, input_tokens=..., output_tokens=..., total_tokens=...)

三次 run 都累加到同一个 RunUsage。Delegate 模式里你用 usage=ctx.usage,就是把当前 run 的 usage 对象传给子 Agent,自然累加。

设预算上限

RunUsage 还能带预算:

from pydantic_ai import UsageLimits

# 限制本次多 Agent 任务总输出 token 不超过 2000
limits = UsageLimits(output_tokens_limit=2000, request_limit=10)

try:
    r1 = await agent_a.run("...", usage=usage, usage_limits=limits)
    r2 = await agent_b.run("...", usage=usage, usage_limits=limits)
except UsageLimitExceeded as e:
    print("已达到预算上限:", e)

超过限额会抛 UsageLimitExceeded——生产环境强烈建议开,防止失控循环烧钱。

六、消息历史在多 Agent 间传递

第 6 章讲过 message_history——跨 Agent 也能用。比如一个聊天机器人前 3 轮用便宜模型,第 4 轮升级到贵模型,历史不丢:

cheap = Agent("openai:gpt-4o-mini", system_prompt="闲聊助手")
pro = Agent("openai:gpt-4o", system_prompt="深度思考助手")

history = []
for i in range(3):
    r = await cheap.run(input("> "), message_history=history)
    print(r.output)
    history = r.all_messages()

# 用户说"给我深入分析一下",升级
r = await pro.run(input("> "), message_history=history)
print(r.output)
history = r.all_messages()

但要注意跨 Agent 的 system_prompt 会叠加——前面 cheap 的 system 消息还在历史里,pro 自己的 system 又加一条。纯粹替换建议用 new_messages() 截掉 system,或者做一次手工 trim。

七、实战:多 Agent 写作流水线

把今天学的串起来做一个完整的"研究→写作→审校"流水线,带成本控制、delegate、限额:

import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, UsageLimits
from pydantic_ai.usage import RunUsage

# ── 结构化结果 ──
class Outline(BaseModel):
    title: str
    bullets: list[str]

class Article(BaseModel):
    title: str
    body: str
    word_count: int

# ── 专门 Agent ──
researcher = Agent("openai:gpt-4o-mini", output_type=Outline,
    system_prompt="根据主题列出 5 个深度要点。")
writer = Agent("openai:gpt-4o", output_type=Article,
    system_prompt="根据要点写 800 字文章,文风清晰老练。")
editor = Agent("openai:gpt-4o-mini", output_type=Article,
    system_prompt="改写为更凝练版本,词数减半。")

# ── 协调 Agent(父)——用 delegate 模式 ──
chief = Agent(
    "openai:gpt-4o-mini",
    output_type=Article,
    system_prompt=(
        "你是主编。收到主题后,依次调用 research→write→edit 三个工具,返回最终稿。"
    ),
)

@chief.tool
async def research(ctx: RunContext[None], topic: str) -> Outline:
    return (await researcher.run(topic, usage=ctx.usage)).output

@chief.tool
async def write(ctx: RunContext[None], outline: Outline) -> Article:
    return (await writer.run(outline.model_dump_json(), usage=ctx.usage)).output

@chief.tool
async def edit(ctx: RunContext[None], draft: Article) -> Article:
    return (await editor.run(draft.model_dump_json(), usage=ctx.usage)).output

# ── 跑 ──
async def main():
    usage = RunUsage()
    limits = UsageLimits(output_tokens_limit=3000, request_limit=8)
    r = await chief.run(
        "帮我写一篇关于 Python GIL 2025 进展的文章",
        usage=usage,
        usage_limits=limits,
    )
    print(r.output.body)
    print("用量:", usage)

asyncio.run(main())

这一段就能看出"多 Agent delegate"的价值:主编 Agent 自己决定工具顺序,调用成本汇总到 usage,出现预算超限自动中止。

八、什么时候该用 Graph(第 7 章)而不是多 Agent?

实际项目大部分场景多 Agent delegate 就够——Graph 是备选方案,别过度工程。

九、八个常见坑

1. 忘记传 usage=ctx.usage:子 Agent 的 token 没算进总统计,成本报表少算。
2. 子 Agent 没设 usage_limits:万一子 Agent 循环跑工具,token 爆表。统一在顶层限额。
3. 多 Agent 各自用不同 provider:追踪成本时注意不同 provider 价格不一样,混合用会让简单的 token 数折合成本出错。
4. Handoff 模式下共享 message_history 产生多层 system prompt:模型混乱。用 new_messages 或者干脆每个 Agent 开新会话。
5. 一股脑什么都用多 Agent:"写个天气助手用 5 个 Agent 协作"——过度工程。先用单 Agent + 工具,不够再拆。
6. 子 Agent 是重型模型做小事:调个 gpt-4o 只问一句话——浪费。专家 Agent 用贵模型,通用用便宜的。
7. 多 Agent 同时并发调同一 provider 撞限额:rate limit。生产环境强烈建议用 LiteLLM Proxy 或自建限流中间件。
8. 循环 delegate 没有终止条件:父调子,子又调父——陷入来回甩锅。Agent 描述要写清楚"什么时候停"。

十、本章小结