Agent · 多智能体编排（Multi-Agent）

💡 一句话理解

多智能体像一个项目经理带专家团: PM(Supervisor)自己不动手, 他把"调研竞品""审查代码""写报告"分派给三位专家(subagent); 每位专家在自己办公室(独立上下文)翻资料干活, 完了交一页纸摘要, 而不是把整屋子的草稿全堆到 PM 桌上。判断要不要拆多 agent 只看一条: 主上下文是否被中间产物撑爆——单 agent 干得又快又好的活, 拆开只会更贵、更慢、更难调试。

🧠 必知必会 必考 & 必会

何时才需要多 agent

三个正当理由: 上下文隔离、专业分工(不同 system+工具)、并行加速。除此之外的拆分都是成本乘数。

需要: 主上下文要被"翻 50 个文件"撑爆 → 拆
# 不需要: 简单改写/单轮问答 → 单 agent 更快更省
# 判据: 中间产物体积 vs 主上下文预算

Supervisor 模式

编排者-工人: 主控只做拆解/分派/汇总, 不执行具体工作; 工人只管自己那摊。类似微服务的编排式 saga。

plan = [research, review, report]
for task in plan: dispatch(agent[task.kind], task)
# 关键: 主控不亲自翻文件 — 它翻一次, 上下文就脏一次

Handoff 移交

不是"分派完等结果", 而是把控制权整个交给下一个 agent(客服分诊场景), 交接载荷要精简。

handoff_payload = {"user_id", "intent", "history_summary"}
# 错: 把 40 轮闲聊原样打包移交 → 下家上下文爆炸

上下文隔离

subagent 的价值就是"主控看不见过程": 30 个网页/200k tokens 的 repo 内容全在子上下文里消化, 只交摘要。

sub.run("翻 30 个网页总结竞品定价")  # 200k tokens 在子上下文
# 主控收到: "A ¥99/月, B ¥199/月, 结论..." (180 tokens)
# 隔离被破坏 = 全文回灌 = 花两倍钱买一个脏上下文

结构化回传

subagent 返回值用 pydantic schema 约束: 字段、类型、长度上限。校验不过就打回重写, 不让复读机污染主控。

class Report(BaseModel):
    summary: str = Field(max_length=300)   # 硬上限
    verdict: Literal["ok", "failed"]
# 不过校验 → 连原报错一起打回, 让它重写

工具权限收窄

每个 agent 只绑定自己必需的工具: 研究员没有写库工具, 审查员没有执行工具。默认拒绝, 新工具不进名单就不可达。

TOOLSETS = {"researcher": [search, read_url],
            "reviewer":   [read_file, grep]}
# 错: 所有 agent 共享全局 ALL_TOOLS → 权限蔓延成事故温床

并行 fan-out

互不依赖的子任务并发执行, 总延迟≈最慢那个; 有依赖的必须排后续波次。

wave1 = gather(research_A, research_B, pull_metrics)
wave2 = write_report(wave1)          # 依赖 wave1, 只能等
# 错: 把有依赖的任务也 gather → 拿着空结果开工

子代理预算

subagent 是完整的 agent 循环, 同样能失控。每个都要带 max_steps/max_cost/timeout, 且可以比主控更紧。

Budget(max_steps=10, max_cost=0.15)  # 子任务预算更小
# 一个 research sub 失控烧 $5 = 编排层的守卫失职

A2A 协议

Agent-to-Agent: 用 agent card 声明"我是谁/在哪/会什么/怎么鉴权", 让别的团队的 agent 能发现并调用你。

card = {"name": "travel-planner", "endpoint": url,
        "skills": ["itinerary"], "auth": "oauth2"}
# 跨团队互操作的标准名片 — 类比 MCP 之于工具

黑板 vs 消息

协作两种底座: 黑板(共享任务板, 原子认领)与消息传递(点对点)。多人协作优先黑板, 链式流程优先消息。

blackboard.claim(task_id, agent)   # 原子认领, 防重复劳动
# 消息传递适合流水线; 黑板适合"一群人抢一堆活"

失败隔离

一个 subagent 超时/崩溃不应拖垮整体: 隔离错误 → 降级结果上抛 → 主控决定跳过/重试/收尾。

except Timeout: return {"status": "degraded",
                        "summary": "子任务超时已跳过"}
# 错: 让一个 sub 的异常冒泡毁掉整次编排

成本乘数

多 agent 的 token 开销 ≈ Σ(各自完整循环)。拆 N 个 agent 不是 N 倍智能, 是 N 倍账单——所以必须逐 agent 记账。

ledger = {"supervisor": 0.03, "researcher": 0.21, ...}
# researcher 占 70% → 它是优化对象, 不是"再加一个 agent"

🏭 生产实战 real world

async def supervise(goal: str) -> str:
    plan = await make_plan(goal)        # LLM 拆解成任务列表
    results = {}
    for task in plan:
        agent = pick_agent(task.kind)   # kind → 独立上下文+工具集
        results[task.id] = await agent.run(
            task, budget=Budget(max_steps=10, max_cost=0.20))
    return await synthesize(goal, results)  # 只拿摘要做总结
# 主控全程上下文 < 5k tokens, 三个 sub 各自 50k+ 也无所谓

review_agent = Agent(
    name="code-reviewer",
    system=REVIEWER_PROMPT,          # 干净上下文, 只带审查规则
    tools=[read_file, grep, diff],   # 无写权限
    max_steps=15, max_cost=0.10)

async def run_review(pr_id: str) -> ReviewReport:
    raw = await review_agent.run(f"审查 PR {pr_id}, 输出问题清单")
    return ReviewReport.model_validate_json(raw)
# 主上下文收到: 12 个问题 + 1 个 diff; 不是 200k 文件内容

class ReviewReport(BaseModel):
    issues: list[Issue]
    verdict: Literal["approve", "request_changes"]
    summary: str = Field(max_length=300)   # 硬上限, 防复读

def accept(raw: str) -> ReviewReport:
    try:
        return ReviewReport.model_validate_json(raw)
    except ValidationError as e:
        raise RetryWith(raw, hint=str(e))  # 带错误打回重写
# → 回传格式永远可用; summary 超长 = 编排层的失职

wave1 = await asyncio.gather(
    run_sub(research("竞品A定价"), budget=0.15),
    run_sub(research("竞品B定价"), budget=0.15),
    run_sub(pull_metrics("Q3 转化率"), budget=0.05))
wave2 = await run_sub(write_report(wave1), budget=0.10)
# 依赖关系写在波次里, 而不是祈祷模型自己懂顺序

async def triage(msg: str, user) -> str:
    intent = await classify(msg, ["billing", "tech", "other"])
    if intent == "other":
        return escalate_human(user)
    agent = {"billing": billing_agent, "tech": tech_agent}[intent]
    payload = {"user_id": user.id, "intent": intent,
               "history": summarize(user.recent_msgs, 300)}
    return await agent.handle(payload)   # 移交摘要, 不移交谈天记录

TOOLSETS = {
    "researcher": [web_search, read_url],     # 只读外网
    "reviewer":   [read_file, grep, diff],    # 只读仓库
    "ops":        [restart_pod, scale],       # 高危, 需审批
}
def pick_agent(kind: str) -> Agent:
    return Agent(tools=TOOLSETS[kind], system=SYS[kind])
# 新工具不进任何 TOOLSETS 就永远不可达 — 默认拒绝

async def run_with_fallback(agent, task, timeout=60):
    try:
        return await asyncio.wait_for(agent.run(task), timeout)
    except (asyncio.TimeoutError, AgentCrashed):
        return {"status": "degraded",
                "task_id": task.id,
                "summary": f"子任务超时已跳过, 报告将缺少该部分"}
# 主控看到 degraded 标记 → 报告里写明 limitation, 而非静默缺块

// https://agents.corp/.well-known/agent.json (jsonc)
{
  "name": "fx-rates",
  "description": "查询实时汇率, 输入币种对, 输出金额换算",
  "endpoint": "https://agents.corp/fx",
  "auth": "oauth2",
  "skills": ["convert", "historical"]
}
// 本方按 card 把对方当"一个工具"接入 — 协议同构 MCP 思想

class Blackboard:
    """共享任务板: 状态唯一真相, agent 只改自己名下条目"""
    def claim(self, task_id: str, agent: str) -> bool:
        r = db.execute(
            "UPDATE tasks SET owner=%s, status='claimed'"
            " WHERE id=%s AND status='open'",   # 原子抢占
            (agent, task_id))
        return r.rowcount == 1
# 抢不到 = 别人已在做 → 换下一个, 无需任何协调对话

class Ledger:
    def __init__(self, run_id: str):
        self.rows: dict[str, dict] = {}
    def add(self, agent: str, usage):
        r = self.rows.setdefault(agent, {"tokens": 0, "cost": 0.0})
        r["tokens"] += usage.total_tokens
        r["cost"]   += cost_of(usage)
# → {'supervisor': $0.03, 'researcher': $0.21, 'reviewer': $0.09}
# researcher 占 70% → 优化它的提示词, 而不是再拆一个 agent

⚠️ 编码注意与常见坑 pitfalls

# 错: messages += subagent.messages   # → 全文回灌, 隔离归零
# 对: messages.append(user(report.summary))

# 错: agent 互相委托无上限       # → 循环烧钱到熔断
# 对: if depth >= 2: return solve_locally(task)

# 错: Agent(tools=ALL_TOOLS)       # → 权限蔓延
# 对: Agent(tools=TOOLSETS[role])

# 错: gather(*[w(shared_state) ...])  # → 交错写
# 对: results = gather(...); merged = merge(results)

# 错: await sub.run(task)            # → 可能永不返回
# 对: await asyncio.wait_for(sub.run(task), 60)

# 错: 问天气也 planner→worker→synthesizer
# 对: 拆分判据 = 上下文是否会被中间产物撑爆

# 错: handoff({"intent": "billing"})   # → 下家两眼一抹黑
# 对: + "history": summarize(recent, 300)

# 错: billing 与 support 都含 "refund" 关键词  # → 互转
# 对: 路由表唯一: refund → billing; 纠纷 → 人工

# 错: 所有 agent 日志混一个文件
# 对: span(agent="reviewer", trace_id=..., task_id=...)

# 错: sub.run(task)                  # → 无步数无费用上限
# 对: sub.run(task, budget=Budget(10, 0.15))

# 错: json.loads(sub_output)        # → 不是 JSON 直接炸
# 对: Report.model_validate_json(raw) + RetryWith

# 错: 直接采用 researcher 的"竞品停产"
# 对: 要求 evidence: [{url, quote}] 字段, 缺证据降置信

# 错: if task.status=='open': task.owner=me  # 竞态
# 对: UPDATE ... WHERE status='open' 看 rowcount

# 错: summary: str                 # → 800 字"摘要"
# 对: summary: str = Field(max_length=300)

# 错: agent A 的工具里藏着"调用 agent B"
# 对: supervisor 统一分派; agent 之间禁止互调

# 错: gather(research, report_of(research))
# 对: wave1 = gather(a, b); wave2 = report(wave1)

# 错: dispatch("查一下价格")        # → 哪家? 哪国? 何时?
# 对: brief 含目标/范围/输出格式/禁做事项

# 错: 销售 agent ↔ 售前 agent ↔ 交付 agent 逐级传话
# 对: 一个 agent + 好工具, 常常胜过三个 agent 传话

# 错: except: return ""              # → 悄悄缺块
# 对: return {"status": "degraded", ...} 上抛主控

# 错: 只看总 tokens            # → 无法归因
# 对: ledger["researcher"]["cost"] 按 agent 出账