苦猿的大模型日记 · Day19 · Multi-Agent 协作——一个 Agent 干不过来时怎么办-帮普通人把AI学进简历系列

前言:Multi-Agent 上线一晚的成本账

先甩一组真实数字。

某团队把一个单 GPT-4o Agent 升级成"GPT-4o 当 Supervisor + GPT-4o 当 Planner + GPT-4o 当 Coder"的三 Agent 架构,跑了一个晚上,大约 400 次用户请求——账单从 ¥38 涨到 ¥410,10 倍。

同一个任务,换成"一份 DeepSeek key 跑三个角色",账单是 ¥41。几乎打平单 GPT-4o Agent。

反差点就一句:Multi-Agent 不是"用三个模型",是"用三种角色"。模型可以共用,角色不能共用。

很多人 RAG / Agentic RAG 跑顺之后,下一步就想拆 Multi-Agent。上一篇我把检索从写死的流水线变成 Agent 可以自己调用的工具——但那还是一个 Agent 在做所有事。

当一个 Agent 同时要拆任务、调工具、写答案、自检反思,它的 prompt 会越长越乱,工具越多越选不准。

到了某个临界点,加 prompt 救不回来,加工具救不回来,换更强的模型也救不回来——只能拆。

但拆完怎么不烧钱、不死锁、不黑盒?这就是今天全部要讲的事。

先泼盆冷水:Multi-Agent 不是单 Agent 的升级版,是另一种东西。单 Agent 求全能,Multi-Agent 求分工,思维模型完全不一样。把单 Agent 那套优化思路硬套过来,会全错。

---

PART 01:单 Agent 的天花板——它会被自己撑爆

很多人觉得"我的 Agent 还能加功能,再撑撑"。先看看四个撑爆信号。

信号 1:prompt 撑爆

一个客服 Agent 的 system prompt 里同时塞了"退货政策 / 物流查询 / 技术支持 / 闲聊兜底"——四个角色的指令打架。

实测数据:prompt 超 3k token 之后,工具选择准确率从 92% 掉到 71%。模型不是不会用工具,是"不知道现在该用哪把"。

信号 2:工具选择失准

塞了 12 个工具——rag_search / calculator / order_query / weather_api / …… 模型在多工具场景下选错的概率,随工具数指数上升。

经验阈值我给到很死:工具数超过 8 个,就该拆。 拆成"客服 Agent 只看客服工具""订单 Agent 只看订单工具",每个 Agent 的工具数压到 5 个以内,准确率立刻回来。

信号 3:单点失败

一个 Agent 同时负责"查资料 + 写代码 + 自检反思"——任何一步出错,整条链路崩。没有降级,没有重试,没有 fallback。

单 Agent 的容错只能靠"模型自己注意点",这在生产环境约等于没有。

信号 4:不可并行

用户问"X3 和 X5 哪个适合养猫"——单 Agent 只能串行查 X3 再查 X5。

两个 Worker 并行各查一个,延迟能砍 40%。但单 Agent 的代码结构里没有"并行"这个出口。

一个反直觉判断

这四种信号,靠加 prompt、加工具、加示例都救不回来。

单 Agent 的天花板是结构性的,不是参数问题。唯一的解法是拆——让一个 Agent 只做一件事,多件事就拆成多个 Agent 协作。

---

PART 02:三大协作范式——Supervisor / Pipeline / Debate 怎么选

Multi-Agent 不是"把 Agent 塞一起",是有明确范式的。选错范式比单 Agent 还差。

范式 1:Supervisor 模式(主管派活)

一个 Supervisor Agent 接用户请求 → 决定派给哪个 Worker → 收 Worker 结果 → 决定继续派还是收尾。

       用户
        ↓
   Supervisor ←→ Worker A
        ↑       ←→ Worker B
        ↑       ←→ Worker C

适用场景:任务可拆但拆法不固定——客服、研究助理、复杂 RAG。

关键约束:Worker 之间不直接对话,只通过 Supervisor。这一条防消息爆炸、防死锁、便于审计。我见过新手让所有 Worker 共享一个"黑板",跑两轮就乱成一锅粥。

范式 2:Pipeline 模式(流水线串联)

Agent A 的输出直接喂给 Agent B,B 的输出喂给 C,没有中枢调度。

用户 → Agent A → Agent B → Agent C → 答案

适用场景:任务拆法固定——翻译→润色→校对、资料检索→摘要→润色。

关键优势:比 Supervisor 省 30% token,因为不需要 Supervisor 那一轮决策调用。代价是灵活性差,流程一变就得改代码。

范式 3:Debate 模式(辩论收敛)

多个 Agent 各自独立答同一问题,再互相 critique,最后收敛到一个答案。

适用场景:答案需要交叉验证的高风险场景——医疗、法律、金融判断。

关键坑:成本是单 Agent 的 N 倍(N = 参与辩论的 Agent 数)。生产慎用,大多数业务场景用不上。

决策建议

90% 的生产场景用 Supervisor。 Pipeline 适合流程固定、Debate 适合高风险高预算。今天的实战代码也聚焦 Supervisor。

反直觉点再强调一次:协作范式选错,比单 Agent 还差。比如把客服做成 Debate——一次查询烧 3 倍 token,答案反而被多个 Agent 互相"和稀泥"带偏。

---

PART 03:实战第一刀——一份 DeepSeek key 跑三个 Agent

这一节是全文最反差的代码段。

市面上 Multi-Agent 教程的默认配置是"GPT-4o 当 Supervisor + Claude 当 Worker + Gemini 当 Reviewer",账单一晚上烧几百刀。

核心反差点:Multi-Agent 不需要三个模型,需要三个角色。 一个 DeepSeek API key + 三套不同 system prompt + 三个 temperature,就能跑出三个角色分明的 Agent。

三角色的 system prompt 设计

这是全文最关键的工程投入——Supervisor 的派活准确率、Worker 的执行质量,80% 由 prompt 决定。

# Planner:拆任务,求稳,temperature 低
PLANNER_PROMPT = """你是一个任务规划员。接到用户请求后,把它拆成 1-3 个具体子任务,
每个子任务说明清楚要查什么、用什么工具、预期产出。
不要自己执行任务,只规划。输出 JSON:{"tasks": [...]}"""

# Coder:执行任务,求发散,temperature 中
CODER_PROMPT = """你是一个执行员。收到子任务后,调用可用工具(rag_search / calculator)
完成它,给出执行结果。如果工具不够用,明确说"需要 XX 工具",
不要硬编。"""

# Reviewer:审结果,求严格,temperature 极低
REVIEWER_PROMPT = """你是一个审稿员。拿到执行结果后,判断三点:
1. 是否真的回答了原始问题
2. 是否有明显遗漏
3. 是否有事实错误
输出 JSON:{"pass": true/false, "reason": "..."}"""

角色配置代码

# 一个 API key,三种角色
def make_agent(role, temperature):
    return {
        "model": "deepseek-chat",
        "system": {
            "planner": PLANNER_PROMPT,
            "coder": CODER_PROMPT,
            "reviewer": REVIEWER_PROMPT,
        }[role],
        "temperature": temperature  # planner=0.3, coder=0.7, reviewer=0.1
    }


PLANNER = make_agent("planner", 0.3)
CODER = make_agent("coder", 0.7)
REVIEWER = make_agent("reviewer", 0.1)

两个关键工程坑

坑 1:Supervisor 自己也可以是规则,不一定要 LLM。

简单路由(关键词匹配 / 工具白名单)用规则——能省一次 LLM 调用,延迟和成本砍半。复杂判断才上 LLM Supervisor。

判断标准:派活逻辑能用 if-else 写清楚,就别用 LLM。 Supervisor 用 LLM 只在"任务拆法不固定"时才值。

坑 2:temperature 不是越高越好。

Planner 求稳(0.3)、Coder 求发散(0.7)、Reviewer 求严格(0.1)——一个 API key 通过 temperature 区分角色,比换三个模型更可控。

很多新手把所有 Agent 都设成 temperature=0.7,结果 Planner 每次拆的任务都不一样,Reviewer 审的标准也飘——整个系统不可复现,debug 到崩溃。

---

PART 04:跑通最小 Multi-Agent——Supervisor 主循环代码

把 PART 03 的三个角色接到一个 Supervisor 主循环里,跑通一个会自己派活、收结果、自检的多 Agent 系统。基于 DeepSeek client 和上一篇搭好的 rag_search 工具。

Supervisor 主循环

from openai import OpenAI
import json

client = OpenAI(
    api_key="你的 DeepSeek API Key",
    base_url="https://api.deepseek.com/v1"
)


def call_agent(agent_cfg, user_msg, trace_id, tools=None):
    """统一的 Agent 调用入口,带 trace_id 落库"""
    log_event(trace_id, agent_cfg["system"][:20], "input", user_msg)
    resp = client.chat.completions.create(
        model=agent_cfg["model"],
        messages=[
            {"role": "system", "content": agent_cfg["system"]},
            {"role": "user", "content": user_msg}
        ],
        tools=tools,
        temperature=agent_cfg["temperature"]
    )
    out = resp.choices[0].message.content
    log_event(trace_id, agent_cfg["system"][:20], "output", out,
              tokens=resp.usage.total_tokens)
    return out


def supervisor_loop(user_query, max_rounds=4):
    trace_id = gen_trace_id()  # 全链路追踪 ID,PART 06 详讲
    log_event(trace_id, "user", "input", user_query)

    for round_idx in range(max_rounds):
        # Step 1: Planner 拆任务
        plan = call_agent(PLANNER, user_query, trace_id)
        tasks = json.loads(plan)["tasks"]
        log_event(trace_id, "supervisor", "plan", tasks)

        # Step 2: Coder 串行/并行执行每个子任务
        results = []
        for task in tasks:
            r = call_agent(CODER, task, trace_id,
                           tools=[rag_search_tool, calc_tool])
            results.append({"task": task, "result": r})

        # Step 3: Reviewer 审一遍
        review = call_agent(REVIEWER,
                            json.dumps(results, ensure_ascii=False),
                            trace_id)
        review = json.loads(review)
        log_event(trace_id, "reviewer", "verdict", review)

        # Step 4: 通过则答,不通过则带着反馈再循环
        if review["pass"]:
            return synthesize_answer(user_query, results, trace_id)
        else:
            user_query = f"{user_query}\n[审稿反馈]{review['reason']}"

    return "(超过最大轮次,强制返回当前最佳结果)"

跑一个真实例子

用户问"我们 X3 和 X5 哪个适合养猫家庭?给我一个推荐"——

单 Agent 版本(前一篇的 Agentic RAG):一次性召回 → 答 → 大概率漏维度,因为单次检索捞不全"养猫"背后的隐藏需求。

Multi-Agent 版本:

• Planner 拆成"查 X3 规格 + 查 X5 规格 + 查猫相关维度(防缠绕/滤网/噪音)"
• Coder 三个子任务并行查,各自召回
• Reviewer 审"是否覆盖养猫场景"——发现漏了"噪音敏感度",打回
• 第二轮 Coder 补查噪音数据,Reviewer 通过
• Supervisor 综合三份结果给推荐

质量明显更高,但延迟是单 Agent 的 2-4 倍。 这是 Multi-Agent 的固有代价。

一个必须设的工程兜底

max_rounds 一定要设上限。

不设的话,Reviewer 一直说"不够" → Planner 一直重拆 → Coder 一直重跑 → 死循环烧 token。配上 PART 05 讲的超时强制接管机制,这块才能上线。

---

PART 05:消息协议与死锁坑——Multi-Agent 上线必踩的雷

这一节是工程亮点最密集的一块。Multi-Agent 上线 80% 的故障,都出在 Agent 之间怎么传消息、怎么不互相卡死。

坑 1:消息传全量 vs 摘要

错误做法:Coder 返回 2000 token 报告,Reviewer 把这 2000 token 全塞进 context 审——三轮之后 context 撑爆,token 账单原地翻倍。

正确做法:Worker 之间只传摘要,不传完整历史。每个 Agent 的 context 是隔离的,只接收上家的结构化摘要。

# 结构化消息协议(不是自然语言全量)
def pack_message(from_agent, task_id, summary, key_data):
    return {
        "from": from_agent,        # 谁发的
        "task_id": task_id,         # 任务版本号
        "summary": summary,         # 200 字以内摘要
        "key_data": key_data,       # 结构化关键字段
        "turn_id": gen_turn_id()    # 这一轮的版本号
    }

Worker A 返回的 2000 token 报告,压成 200 字摘要再传给 Reviewer——token 成本能砍一个数量级。 这是 Multi-Agent 省钱的头号手段。

坑 2:Agent 互相甩锅死循环

场景:Planner 说"这事 Coder 干" → Coder 说"我工具不够,让 Planner 重新拆" → Planner 又拆一遍同样的事 → Coder 又说不够……

解法:Supervisor 必须有超时强制接管机制。

# 每个 Agent 的失败次数有上限
if agent_fail_count[task_id] >= 2:
    # 强制走 fallback 路径:直接给用户兜底答案
    return fallback_response(user_query)

这条不写,Multi-Agent 跑一晚能烧掉一周的预算。

坑 3:并行 Worker 结果冲突

场景:Worker A 查 X3 适合养猫(说"适合"),Worker B 查 X5 适合养猫(也说"适合")——Supervisor 不知道怎么推荐。

解法:Reviewer 必须有冲突仲裁规则。 要么再派一个 Worker 做对比查询,要么 Supervisor 按预设优先级(如"价格优先 / 功能优先")裁决。

冲突不能靠 LLM 临场判断,得在 prompt 里写死仲裁规则。

坑 4:消息黑盒,出问题无法 debug

场景:用户投诉"答错了",你打开日志只看到最终答案——不知道是 Planner 拆错了、Coder 查错了、还是 Reviewer 放过了。

解法:全链路 trace_id + 决策日志落库。 一次用户请求 → Planner/Coder/Reviewer 的每次 Think/Act/Observe 都带同一个 trace_id 落库,出问题能完整回放。

一个反直觉判断

Multi-Agent 的可观测性投入,应该跟代码投入 1:1。

不写日志的 Multi-Agent 上线等于裸奔——单 Agent 黑盒已经够难调,多 Agent 黑盒就是地狱。下一篇的最后一节给一段 30 行的可直接抄走的日志装饰器,接到 PART 04 的主循环上即可。

---

PART 06:30 行日志装饰器 + 最终决策表

把可观测性做成一段可直接复用的代码。这段装饰器接到前面 call_agent 上,一次接入,全链路可观测。

日志装饰器(可直接抄)

import time, uuid, json
from functools import wraps

# 全局 trace 上下文
_trace_store = {}  # {trace_id: [events]}


def gen_trace_id():
    return f"trc-{uuid.uuid4().hex[:8]}"


def gen_turn_id():
    return uuid.uuid4().hex[:6]


def log_event(trace_id, agent, event_type, payload, tokens=0):
    """所有 Agent 调用统一走这里落库"""
    _trace_store.setdefault(trace_id, []).append({
        "ts": time.time(),
        "agent": agent,           # planner / coder / reviewer / supervisor
        "event": event_type,      # input / output / tool_call / error
        "payload": payload,
        "tokens": tokens          # 这一调用花了多少 token
    })


def traced_agent(role):
    """装饰器:自动给 Agent 调用打 trace"""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, trace_id=None, **kwargs):
            tid = trace_id or gen_trace_id()
            log_event(tid, role, "input",
                      {"args": args, "kwargs": kwargs})
            start = time.time()
            result = fn(*args, **kwargs)
            latency = time.time() - start
            log_event(tid, role, "output",
                      {"result": str(result)[:200]},
                      tokens=kwargs.get("usage_tokens", 0))
            log_event(tid, role, "latency_ms", int(latency * 1000))
            return result
        return wrapper
    return decorator


def replay_trace(trace_id):
    """调试时一行调用回放整条决策链"""
    for e in _trace_store.get(trace_id, []):
        print(f"[{e['ts']:.2f}] {e['agent']:>10} | "
              f"{e['event']:<12} | {e['payload']}")

用法:把 PART 04 的 call_agent 内部那几次 LLM 调用,用 @traced_agent("planner") / @traced_agent("coder") 装饰——一次接入,出问题 replay_trace(trace_id) 一行回放整条链。

最终决策表——什么时候上 Multi-Agent

场景	该用	理由
客服 FAQ(答案明确单一)	单 Agent 或普通 RAG	Multi-Agent 是浪费,反而引入不确定性
复杂业务咨询(多约束条件)	Agentic RAG	单 Agent + 多轮检索够用,不必拆
任务可拆且子任务能并行(如对比查询)	Multi-Agent(Supervisor)	并行能省 40% 延迟
一个 Agent prompt 撑到 4k+ 还在加	Multi-Agent	结构性撑爆,必须拆
答案需要交叉验证(医疗/法律/金融)	Multi-Agent(Debate)	唯一能交叉验证的方案
流程固定的多步处理(翻译→润色→校对)	Multi-Agent(Pipeline)	比 Supervisor 省 30% token

三个收尾工程判断

1. 延迟爆炸:Multi-Agent 一轮 = N 次 LLM 调用。实时聊天必须加流式输出 + "思考中..."动画。
2. token 成本翻 3-10 倍:用 DeepSeek 单 key 多角色能压到接近单 Agent 成本,但 GPT-4o 三 Agent 会让你怀疑人生。
3. 可观测性是底线:没有 trace_id + 决策日志的 Multi-Agent 不要上生产,黑盒 debug 会让你通宵。

Multi-Agent 的核心价值不在"答得更准",而在"能处理单 Agent 处理不了的复杂任务"——别为了时髦而拆,按场景决策。

---

结尾:Multi-Agent 不是升级,是另一种工程

单 Agent 像一个全能选手——啥都能干一点,但啥都不精;Multi-Agent 像一支球队——每个人只干一件事,但配合起来能赢。

一个 API key 跑三个 Agent,省的不是钱,是"用角色替代模型"的认知——这个认知,价值远超 token 单价差。

Multi-Agent 不是单 Agent 的升级,是另一种工程。 单 Agent 求全能,Multi-Agent 求分工——把单 Agent 那套优化思路硬套过来,全错。

互动时间:你看过最离谱的 Multi-Agent 翻车是啥?是死循环烧 token,还是几个 Agent 互相甩锅甩到天荒地老?评论区聊聊。

---

下一篇预告:连着写了 RAG 和 Agent(Day16-19),Day20 回归微调主题。攒了一波 GRPO 之后的实战经验,准备写出来。RAG/Agent 这条线告一段落,关注苦猿看微调回归。

— END —

苦猿 · 帮普通人把 AI 学进简历