Agent开发不就是调用 LLM API?-多 Agent 系统：什么时候需要 Agent 协作？

程序学到昏

614人浏览 · 2026-03-03 16:19:28

程序学到昏 · 2026-03-03 16:19:28 发布

产品经理觉得可以做得更专业——把它拆成五个"专家 Agent"：安全审核、版权检测、质量评估、分类打标、舆情风险，各司其职。

拆完上线，延迟从 3 秒变成了 14 秒，任务成功率从 82% 掉到了 61%。

不是哪个子 Agent 出了问题。是五个 Agent 之间的协调开销、错误传播、上下文重复传输——这些东西把收益全吃掉了，还倒欠了一截。

多 Agent 是当前最热的架构方向，也是被滥用最多的。单 Agent 能做好的事，拆成多 Agent 只会更慢更难调试。这篇文章说清楚：什么时候真的需要多 Agent，三种协作模式怎么选，以及出了问题怎么定位。

01 先问五个问题

在决定拆多 Agent 之前，先诚实地回答这五个问题。如果大多数答案是"否"，单 Agent 大概率够用。

问题	如果是"是"
任务能被清晰拆分成几个独立子任务，且子任务之间依赖关系少？	有并行化空间，多 Agent 可以提速
不同子任务需要完全不同的工具集和领域知识，放在一个 Agent 里会导致上下文过载？	专业化分工有实际价值
单 Agent 跑完整任务的 token 消耗已经到了无法接受的程度？	子 Agent 只传结论可以降成本
某些子任务需要比主任务更高或更低的安全权限？	权限隔离是硬需求
任务规模会持续增长，单 Agent 存在明确的扩展瓶颈？	多 Agent 是规模化的准备

五个问题里如果只有一两个"是"——比如"感觉专业化分工更好看"——这不是拆多 Agent 的理由。

02 三种协作模式

决定用多 Agent 之后，下一个问题是选哪种协作架构。三种主流模式解决的问题不同，不可互换。

模式一：Orchestrator-Worker

中心化调度，主 Agent 拆解任务并分配

一个主 Agent 负责理解任务、制定计划、把子任务分配给各个 Worker Agent，收集结果后汇总。Worker Agent 只执行分配到的子任务，不需要理解全局。

用户 │ ▼ Orchestrator（规划 + 调度） ├── Worker A（数据收集） ├── Worker B（内容生成） └── Worker C（质量检查） │ ▼ 汇总结果 → 用户

✓ 适合：任务可以被清晰拆解、子任务相对独立、有明确的汇总逻辑

✗ 不适合：子任务之间强依赖、Orchestrator 需要频繁干预执行过程

模式二：Plan-then-Execute

规划和执行分离，两个专职 Agent

规划 Agent 专门负责把任务拆解成步骤计划，不执行任何操作。执行 Agent 拿到计划后逐步执行，遇到问题反馈给规划 Agent 重新规划。两者角色严格分离。

用户 │ ▼ Planner（只做规划，不执行） │ 计划 ▼ Executor（只做执行，不规划） │ 遇到问题 ▼ Planner（重新规划） │ ▼ 结果 → 用户

✓ 适合：任务复杂、执行中途可能需要动态调整计划、规划质量对结果影响极大

✗ 不适合：任务简单直接、规划和执行之间反复迭代次数多（每次都有通信开销）

模式三：Peer-to-Peer

无中心调度，Agent 之间平等协商

没有主 Agent，各个 Agent 地位平等，通过消息传递协商决策。每个 Agent 看到其他 Agent 的输出，可以提出异议、补充信息或认可结论。

Agent A ←──→ Agent B ↑ ↑ └───→ Agent C ←─┘ （互相传递消息，达成共识）

✓ 适合：需要多视角验证、没有明确的主导方、辩论式决策场景（如代码审查、方案评审）

✗ 不适合：任务有明确执行顺序、需要一个明确的决策点、Agent 数量超过 3 个（通信爆炸）

03 通信协议：Agent 间消息怎么设计

多 Agent 系统里，Agent 之间传递的消息格式直接影响系统的可靠性。消息格式随意，调试时就是噩梦。

一个稳定的 Agent 间消息格式应该包含：

from dataclasses import dataclassfrom typing import Any, Optionalfrom enum import Enumimport uuidfrom datetime import datetimeclass MessageType(Enum):    TASK        = "task"        # 分配子任务    RESULT      = "result"      # 返回执行结果    ERROR       = "error"       # 报告错误    CLARIFY     = "clarify"     # 请求澄清    REJECT      = "reject"      # 拒绝任务（超出能力范围）@dataclassclass AgentMessage:    id: str                      # 消息唯一 ID，用于追踪    type: MessageType    sender: str                  # 发送方 Agent 名称    receiver: str                # 接收方 Agent 名称    task_id: str                 # 归属的顶层任务 ID    parent_message_id: Optional[str]  # 回复哪条消息    content: Any                 # 消息内容    confidence: float            # 发送方对这条消息的置信度 0-1    timestamp: datetime = None    def __post_init__(self):        if not self.id:            self.id = str(uuid.uuid4())[:8]        if not self.timestamp:            self.timestamp = datetime.now()    def is_reliable(self, threshold=0.7) -> bool:        """置信度低于阈值时，接收方应该要求澄清而不是直接使用"""        return self.confidence >= threshold

几个设计细节值得注意：

每条消息带 confidence 字段。

接收方 Agent 收到低置信度的消息，应该发 CLARIFY 消息请求澄清，而不是直接使用可疑的结果继续执行。
parent_message_id 保留回复链。

调试时能完整还原哪条消息触发了哪条响应，错误溯源从 O(n) 变成 O(1)。
REJECT 类型不能省。

Worker Agent 发现任务超出自己的能力范围时，必须能明确拒绝，而不是硬撑着跑出一个错误结果。没有 REJECT，错误会被静默传播到最终输出。

04 错误传播：最难防的失败模式

单 Agent 出错，问题就在那一个地方。多 Agent 出错，错误会沿着调用链传播，最终输出是错的，但没有任何一个 Agent 自己报错。

这是多 Agent 系统最难调试的地方。

防止错误传播的核心机制是结果验证——每个 Agent 在使用上游 Agent 的结果之前，先做一次基本的合理性检查。

class ValidatedOrchestrator:    def __init__(self, workers: dict, validator_llm):        self.workers = workers        self.validator = validator_llm    async def run_with_validation(self, task: str) -> dict:        plan = await self.plan(task)        results = {}        for step in plan.steps:            worker = self.workers[step.agent]            raw_result = await worker.run(step.sub_task)            # 执行结果验证，再传给下一步            validation = await self._validate(                sub_task=step.sub_task,                result=raw_result,                expected_format=step.expected_output_format            )            if not validation["passed"]:                # 结果有问题：重试一次，还不行就标记并上报                retry_result = await worker.run(                    step.sub_task + f"\n注意：{validation['issue']}"                )                retry_validation = await self._validate(                    step.sub_task, retry_result, step.expected_output_format                )                if not retry_validation["passed"]:                    # 两次都失败，停止继续传播这个错误                    return {                        "status": "failed",                        "failed_step": step.id,                        "issue": validation["issue"]                    }                results[step.id] = retry_result            else:                results[step.id] = raw_result        return {"status": "success", "results": results}    async def _validate(self, sub_task, result, expected_format) -> dict:        verdict = await self.validator.ainvoke(f"""            子任务：{sub_task}            执行结果：{result}            期望格式：{expected_format}            检查：            1. 结果是否回答了子任务？            2. 格式是否符合要求？            3. 是否存在明显错误或自相矛盾？            输出：passed（是/否）和 issue（如果否，具体问题）        """)        return parse_verdict(verdict)

级联失败的隔离

验证之外，还需要隔离机制——一个 Worker 失败，不能让整个系统崩溃。

import asyncioasync def run_workers_with_isolation(workers, tasks, timeout=30):    """并行运行多个 Worker，单个失败不影响其他"""    results = {}    async def run_one(agent_name, task):        try:            result = await asyncio.wait_for(                workers[agent_name].run(task),                timeout=timeout            )            results[agent_name] = {"status": "ok", "data": result}        except asyncio.TimeoutError:            results[agent_name] = {"status": "timeout", "data": None}        except Exception as e:            results[agent_name] = {"status": "error", "data": str(e)}    await asyncio.gather(*[        run_one(name, task)        for name, task in tasks.items()    ])    return results

超时和异常都被捕获，返回明确的状态标记。Orchestrator 收到结果后，根据哪些 Worker 成功、哪些失败，决定是部分汇总还是整体重试。

05 多 Agent 调试：分布式追踪是标配

多 Agent 系统上线之后出问题，最难受的一件事：不知道是哪个 Agent 在哪一步出了错。

单 Agent 的日志是线性的，从头读到尾就能还原执行过程。多 Agent 的日志是并发交织的，不同 Agent 的消息混在一起，没有追踪工具几乎没法定位问题。

分布式追踪是多 Agent 系统的标配，不是可选项。

import timefrom contextlib import contextmanagerclass AgentTracer:    def __init__(self, task_id: str):        self.task_id = task_id        self.spans = []    @contextmanager    def span(self, agent_name: str, operation: str):        """记录一个 Agent 操作的完整生命周期"""        span = {            "task_id":    self.task_id,            "agent":      agent_name,            "operation":  operation,            "start_time": time.time(),            "status":     "running",            "input":      None,            "output":     None,            "error":      None,        }        self.spans.append(span)        try:            yield span            span["status"] = "success"        except Exception as e:            span["status"] = "error"            span["error"]  = str(e)            raise        finally:            span["duration_ms"] = int((time.time() - span["start_time"]) * 1000)    def get_trace(self) -> list:        return sorted(self.spans, key=lambda s: s["start_time"])    def find_first_error(self) -> dict | None:        """快速定位第一个出错的 Agent 和操作"""        for span in self.get_trace():            if span["status"] == "error":                return span        return None# 使用示例tracer = AgentTracer(task_id="task-001")async def worker_with_tracing(agent_name, task, tracer):    with tracer.span(agent_name, "execute") as span:        span["input"] = task        result = await agent.run(task)        span["output"] = result    return result

有了追踪之后，排查问题的流程变成：