多智能体协作架构实战：从单体智能体到智能体网络

上周一个朋友问我，说他用大模型做了个小工具，开始还挺好用，后来任务一复杂就频频出错，一会儿“忘记”前面查过的资料，一会儿把上一个任务的信息混进来。我听完第一反应是：你这不是个例，这是单体智能体撞上天花板了。单靠一个智能体把所有事都扛下来，上下文窗口扛不住，工具调用精度也扛不住，任务一多就像一个人既当前台又当财务还要写方案，不出问题才怪。这种时候就得换一种玩法——不是训练一个全能选手，而是组织一支智能体队伍，各有分工、互相协作。这就是“agency-agents”这套思路的核心。

这名字直译过来就是“代理机构”，我更喜欢把它理解成“智能体团队”。核心做法很简单：把一个大任务拆成多个子任务，交给不同的智能体分别处理，智能体之间通过消息或任务队列进行协作，最后由协调者汇总结果。它是当前AI工程化落地中最实用的组织方式之一，也是很多所谓“Agentic Workflow”真正跑起来的基础。这篇文章我打算从最朴素的工程视角出发，把agency-agents是什么、为什么需要、主流实现路径有哪些、以及手写一个最小可用系统要怎么做，完整梳理一遍。内容面向正在做AI应用开发、被单体智能体可靠性问题折磨过的工程师，也适合刚接触智能体编排、想理解底层原理的初学者。

1. agency-agents本质：从单体智能到智能体协作网络

1.1 单体智能体的三个天花板

先说说我观察到的一个现象：很多团队做AI应用的第一版，都喜欢把所有功能塞给同一个智能体。系统提示词里写了一大堆角色设定，后面挂了几十个工具，表面上看起来“什么都能干”，一旦接入真实业务，问题就一个个冒出来。

第一个瓶颈是上下文窗口。你让一个智能体完成“先调研行业动态、再分析竞品、最后生成一份报告”这种多阶段任务，它运行到后半段时，早把前半段的关键信息忘得差不多了。不是模型变笨了，而是上下文太长之后，注意力会被后面的内容主导，早期信息要么被截断，要么被淹没。我做测试时最直观的感受是：任务链条超过三个环节，输出质量就开始跳水。

第二个瓶颈是工具选择的可靠性。给一个智能体挂30个工具，它的系统提示词光工具说明就要占去大量token，更麻烦的是调用精度会下降——模型在几十个函数里选错工具、填错参数的概率，比只有5个工具时高得多。这就像让一个员工既管财务系统又管客服系统，切换上下文的过程中，必然出现操作错乱。

第三个瓶颈是难以并行。单体智能体处理一个长任务，通常是串行的：先查资料，再分析，再生成，哪怕某些环节完全可以并行，也没办法拆开。任务一多，整体耗时线性增长，根本扛不住生产环境的吞吐要求。这三个瓶颈叠加起来，结论就很清晰：复杂任务不该由一个智能体包揽，而该由一个智能体网络来承载。

1.2 代理网络的核心设计要素

agency-agents不是新框架，更准确地说是一种架构模式。它把智能体当成网络中的节点，每个节点只做自己最擅长的一件事，节点之间通过消息通信，由明确的编排逻辑来组织和协调。

我把这种模式里真正起作用的要素拆成五个。

角色定义是地基。每个智能体必须有清晰的职责边界，一个智能体只干一类事，系统提示词里明确“你是谁、你能做什么、你不能做什么”。不要把两个职责塞进同一个智能体，哪怕它们看起来很像。比如“资料调研”和“内容审核”最好完全分开，否则角色边界一模糊，行为就开始漂移。

消息通信是血管。智能体之间不直接共享内存，而是通过消息传递数据。消息里必须包含发送方、接收方、消息类型、业务负载和链路追踪ID。这个设计我后面会详细讲，它是整个网络能够被调试、被维护的关键。

编排逻辑是大脑。要明确整个流程是中心化的——由一个协调者统一派发任务；还是去中心化的——智能体之间自由通信。我自己的经验是：生产环境优先用中心化编排，因为每个任务的状态、结果、异常都集中在协调者手里，出了问题好定位。去中心化适合探索性场景，但运维成本高得多。

记忆与状态管理是肌肉。每个智能体需要知道自己当前处理的是哪个任务、前面已经拿到什么结果、下一步该往哪里走。这些状态应该被显式管理，要么放在协调者的内存字典里，要么放到Redis这类外部存储里，而不是依赖模型自己在上下文中“记住”。把状态交给模型记忆，是最不靠谱的做法。

反馈闭环是安全保障。任务不是发出去就结束了，要能返回结果、要能被校验、要能在异常时重试或终止。尤其是涉及写库、发邮件这类危险操作，必须在链路里加入人工审批钩子。

这五个要素合起来，本质上就是把一个团队的分工协作机制搬到了智能体系统里。一个智能体网络能否稳定运行，主要看的是这些工程要素有没有落地，而不是看单个模型的能力有多强。

1.3 什么场景适合上代理网络，什么场景别碰

我见到不少团队一听到“多智能体”就兴奋，恨不得把所有的功能都改成智能体协作。但说实话，这玩意儿有它明确的适用边界。

适合用agency-agents的场景有三类。第一类是内容生产流水线，比如“调研-写作-审核-发布”这种流程，天然可以拆成多个环节由不同智能体负责。第二类是信息密集型任务，比如行业研究、竞品分析、报告生成，需要大量检索和归纳，而且中间产物需要多轮加工。第三类是模拟与推演类场景，比如让多个智能体扮演不同角色进行辩论或用户模拟，这种场景本身就是多角色结构。

不太适合的场景也很明确。单个确定性任务，比如“把这个文件转换成PDF”“调用API查询天气”，直接写代码调用更可靠，没必要引入智能体网络。强交互实时场景，比如用户和智能体对话、每一步都要即时反馈，这种场景下一旦引入多智能体编排，延迟会明显上升，体验反而变差。还有对成本极其敏感的批量任务，多个智能体串联意味着多次大模型调用，token成本会成倍增加，需要先算清楚账再决定。

我曾经在一个电商客服项目里试过用多智能体处理“售前咨询-订单查询-售后处理”，结果发现订单查询这种强规则任务，智能体绕来绕去不如十几个if-else干脆。最后只保留了“售前推荐”和“售后安抚”这两个真正需要生成式能力的环节。这个教训我一直记得：架构模式是为问题服务的，不是反过来。

2. 实现路径选型：三种主流方案怎么选

2.1 代码级编排：最灵活但工程量大

第一种实现路径是纯代码级编排。你自己写消息队列、智能体基类、任务调度逻辑，LLM只作为其中的一个计算单元被调用。这种方式没有任何框架依赖，所有控制流都由代码显式表达。

为什么说它灵活？因为你能精确控制每一步。任务怎么拆、消息怎么路由、重试几次、超时怎么办，全部写在代码里。模型的行为永远是不可预测的，但代码的边界可以把不确定性关在笼子里。

代价也很直接：工程量大。你需要自己处理消息的序列化、并发安全、异常处理、日志追踪、状态持久化。做得好，这套系统会非常稳定可控；做得糙，还不如直接用单体智能体。我的看法是，如果你所在团队本来就有不错的工程能力，或者当前项目流程足够复杂、需要深度定制，代码级编排其实是更可靠的选择。

2.2 框架级方案：LangGraph、AutoGen、CrewAI怎么选

如果不想从头造轮子，主流的选择是LangGraph、AutoGen、CrewAI这几个框架，它们解决的问题基本类似，但设计哲学差异很大。

LangGraph的思路是把智能体运行看成一张“状态图”。节点是处理逻辑，边是转移关系，状态由框架显式维护。这种设计非常贴合工作流明确的业务场景，比如审批流、处理流，它最大的杀手锏是支持状态持久化和断点恢复——系统中途挂了，可以从断点处接着跑，而不是从头开始。代价是你必须把流程想得足够清楚，边和状态都要提前定义。

AutoGen则走了另一条路线，它是“多智能体对话驱动”的设计。多个智能体之间通过对话互动来推进任务，特别适合需要多轮讨论、互相辩论、逐步收敛的场景，比如团队头脑风暴、方案评审。但也因为它的驱动方式是对话，流程天然不可控，生产环境里需要额外加护栏。

CrewAI的核心理念是模拟一个团队，所以它特别强调角色和任务的定义。你创建角色、分派任务、按顺序或并行动态执行，代码风格很直观，非常适合内容生产流水线，比如“研究员+写手+审核员”这种结构。上手成本比LangGraph低不少。

我的选型建议是：流程复杂但相对固定，选LangGraph；需要多角色多轮讨论，选AutoGen；核心诉求是“内容流水线”这种角色分明的工作流，选CrewAI会事半功倍。

2.3 协议层探索：A2A与MCP

除了自己写和用框架，还有一种视角值得了解：协议层。MCP（Model Context Protocol）解决的是智能体与工具之间的连接问题，它把数据库、文件系统、外部API统一封装成标准接口，智能体通过统一协议调用这些资源。A2A（Agent-to-Agent）则往前推了一步，它关注的是智能体与智能体之间如何通信、如何发现彼此，更像智能体网络里的“TCP/IP”。

我个人判断是，MCP现在已经进入实用期，如果你在搭建智能体工具层，直接对接MCP标准是划算的，生态丰富且工具侧适配越来越多。A2A还处在早期探索阶段，跨团队、跨平台的智能体协作如果真能标准化，未来的想象空间很大，但短期内自己实现这个协议的意义有限，知道有这个东西、留出扩展位就够了。

2.4 我的选型建议

很多朋友问我到底用哪个，我通常给三层决策逻辑。

第一层看任务是否单一。任务单一只需要一个智能体，不上框架，别折腾。第二层看流程结构是否明确。如果业务的处理步骤相对固定，优先选择轻量级代码编排，或者用CrewAI把角色和任务模型建起来。第三层看是否需要复杂状态流转与容错恢复。这层才需要考虑LangGraph这类状态图框架。

我自己的习惯是：先用代码级方案快速做一个最小可行的POC，把任务拆解、消息协议、链路追踪这些基础设施跑通，然后评估流程的稳定性。如果流程稳定，再决定是否用框架重构；如果流程本身还在摸索，就保留代码级方案。框架是工具，工程问题才是根本。

3. 手写最小可用系统：消息总线+三类智能体

3.1 角色设计与职责划分

我直接上实际方案。先做一个最小可用的agency-agents系统，它在功能上覆盖一个典型的“调研-写作”工作流，包含三个智能体。

协调者（Coordinator）负责接收用户原始任务，拆解子任务，分发给下游智能体，并汇总最终结果。它的系统提示词很简单：“你是任务协调者，负责拆解用户任务，并将结果汇总返回。”研究智能体（Researcher）负责执行资料搜索或数据获取，只对协调者发来的查询任务响应。写作智能体（Writer）负责基于调研结果生成最终内容，不做研究也不做审核。

三个角色各干一件事，边界清楚，不交叉。这就是角色定义的核心，职责的边界越清楚，系统的行为就越可预期。

3.2 消息协议：字段、路由与链路追踪

智能体之间通信的格式我建议直接用轻量JSON结构，一个标准Messsage包含这样几个字段：msg_id消息唯一ID、msg_type消息类型（task/result/error）、target目标智能体名、source来源智能体名、payload业务负载、trace_id链路追踪ID、created_at时间戳。

协议里最容易被忽略的就是trace_id，它必须在生成任务的源头创建，后续这条链路上的所有消息都带着同一个trace_id。没有它，排障时你面对的就是一堆孤立的日志，根本拼不出一次任务的全貌。有它之后，你可以随时用一条命令把这个任务经过所有智能体的处理过程抽出来看。

路由规则也很简单：target必填，消息要么投递给具体智能体，要么广播给全部。生产环境中要特别注意，禁止智能体之间跨级通信，研究智能体只能和协调者通信，写作智能体也只能和协调者通信。跨级通信一旦放开，消息流转就乱了。

3.3 核心代码实现骨架

下面给出一份可以直接运行的Python代码骨架，演示消息总线和三类智能体的配合。代码里同时用到了threading处理并发和queue做消息缓冲，这个思路同样可以移植到Redis Stream或Kafka上做分布式版本。

import json import queue import threading import uuid import time from dataclasses import dataclass, field @dataclass class Message: msg_id: str = field(default_factory=lambda: str(uuid.uuid4())) msg_type: str = "task" target: str = "" source: str = "" payload: dict = field(default_factory=dict) trace_id: str = "" created_at: float = field(default_factory=time.time) class MessageBus: def __init__(self): self._queues = {} self._lock = threading.Lock() def register(self, agent_name: str): with self._lock: if agent_name not in self._queues: self._queues[agent_name] = queue.Queue() def send(self, message: Message): with self._lock: if message.target not in self._queues: raise ValueError(f"agent not found: {message.target}") self._queues[message.target].put(message) def receive(self, agent_name: str, timeout: float = 10.0) -> Message: with self._lock: if agent_name not in self._queues: raise ValueError(f"agent not found: {agent_name}") q = self._queues[agent_name] return q.get(timeout=timeout) class Agent: def __init__(self, name: str, bus: MessageBus, system_prompt: str = ""): self.name = name self.bus = bus self.system_prompt = system_prompt bus.register(name) def send(self, target: str, msg_type: str, payload: dict, trace_id: str): self.bus.send(Message( msg_type=msg_type, target=target, source=self.name, payload=payload, trace_id=trace_id )) def run(self): while True: msg = self.bus.receive(self.name) try: self.handle(msg) except Exception as e: self.send("coordinator", "error", {"error": str(e), "origin_msg_id": msg.msg_id}, msg.trace_id) def handle(self, msg: Message): raise NotImplementedError class Coordinator(Agent): def __init__(self, name: str, bus: MessageBus): super().__init__(name, bus, "你是任务协调者，负责拆解用户任务并汇总结果。") self.results = {} def handle(self, msg: Message): if msg.msg_type == "task": user_task = msg.payload["user_task"] trace_id = str(uuid.uuid4()) self.results[trace_id] = {"user_task": user_task} self.send("researcher", "task", {"query": user_task}, trace_id) elif msg.msg_type == "result": trace_id = msg.trace_id if "research_result" in msg.payload: self.results[trace_id].update(msg.payload) self.send("writer", "task", self.results[trace_id], trace_id) elif "final_output" in msg.payload: self.results[trace_id].update(msg.payload) fin = self.results[trace_id]["final_output"] print(f"[complete {trace_id[:8]}] {fin}") elif msg.msg_type == "error": print(f"[error] {msg.payload}") class Researcher(Agent): def __init__(self, name: str, bus: MessageBus): super().__init__(name, bus, "你是研究智能体，负责搜索资料并输出结构化结果。") self.tools = { "search": lambda q: f"[模拟搜索结果] 关于'{q}'的调研摘要" } def handle(self, msg: Message): if msg.msg_type == "task": query = msg.payload["query"] result = self.tools["search"](query) self.send("coordinator", "result", {"research_result": result}, msg.trace_id) class Writer(Agent): def __init__(self, name: str, bus: MessageBus): super().__init__(name, bus, "你是写作智能体，基于调研结果撰写最终内容。") def handle(self, msg: Message): if msg.msg_type == "task": research = msg.payload.get("research_result", "") user_task = msg.payload.get("user_task", "") final_output = f"[成稿] 针对'{user_task}'的完整内容：{research}" self.send("coordinator", "result", {"final_output": final_output}, msg.trace_id) if __name__ == "__main__": bus = MessageBus() coordinator = Coordinator("coordinator", bus) researcher = Researcher("researcher", bus) writer = Writer("writer", bus) threads = [ threading.Thread(target=coordinator.run, daemon=True), threading.Thread(target=researcher.run, daemon=True), threading.Thread(target=writer.run, daemon=True), ] for t in threads: t.start() bus.send(Message( msg_type="task", target="coordinator", source="user", payload={"user_task": "写一篇关于智能体协作的行业观察"}, trace_id="" )) time.sleep(3)

这份骨架跑起来后，会在控制台输出一条带trace_id前缀的完成日志。真实项目中，你只需把研究智能体里的self.tools["search"]实现替换成大模型调用或真实搜索API，把写作智能体里的“成稿”逻辑替换成LLM生成调用，整个代码骨架就直接变成可用的生产级智能体协作系统。

3.4 与外部工具对接的细节

智能体网络真正发挥威力，是从它能调用外部工具开始的。这里有几个细节值得注意。

先把工具封装统一成函数接口。比如都写成def call_tool(name: str, params: dict) -> str，内部再去分发不同执行逻辑。这样智能体不需要关心工具用什么API、走什么协议，面对的就是一个统一的工具入口。

工具权限要最小化。研究智能体可以调用搜索引擎，写作智能体只能调写作相关工具，所有智能体默认没有写数据库、发邮件的权限。设计原则是每个智能体只获得它完成职责所必需的工具权限，权限一旦扩大，行为不可控的风险就成倍上升。

危险操作必须加人的审批环节。这不能省，而且我不建议用模型来判断某个操作是否危险，应该用代码硬拦。比如发邮件、转账、删除记录，这类工具封装时就加上一个状态标记：requires_approval=true，让消息流到“人工审批”这个节点，等人在控制台上确认后，消息再继续流转。这块做扎实，系统出不了大事故。

3.5 这个方案能扩展到什么程度

这套消息总线模型可以平滑扩展到更大规模。它的优秀之处在于，智能体彼此解耦，运行时互不感知对方存在，所有通信都通过总线中转。想加一个Reviewer审核智能体，新写一个Agent子类，注册到总线，再在协调者逻辑里加一条分发规则就行，现有代码几乎不需要改动。

扩展到分布式也只是把内存队列换成Redis Stream或者RabbitMQ，消息格式完全不变。同一个智能体如果想要横向扩容，比如两个Researcher并行处理不同查询请求，把同一个name注册到多个worker实例上，总线把消息轮询分发就行。这套从单机到分布式、从三个智能体到几十个智能体的路径，是我目前实践下来最平滑的方案。

4. 真实踩坑记录：代理网络跑不起来的那些坑

4.1 上下文串扰：最隐蔽的bug

第一个坑就是上下文串扰，它隐蔽在系统的“看起来正常”之下。之前调试一个多智能体系统时，我发现明明查询的是A行业的数据，到最后生成报告时，内容却混进了B行业下午的调研信息，链路完全对不上。

排查下来，问题出在实现时图省事，让多个智能体共享了一个全局上下文对象，后面的任务把前面任务的聊天记录全带上了。模型区不区分得清？偶尔能，但多数情况下分不清，它只会把上下文里所有信息混合使用。

解决方案我后来总结成两条硬规则。第一，每个任务构建独立的上下文快照，任务的输入和输出通过消息传递，绝不共享一个大内存对象。第二，智能体之间的中间结果尽量用结构化数据传递，比如JSON字段，而不是传递完整聊天文本。

4.2 任务风暴与死循环

第二个坑是任务风暴。表现在系统日志里出现同一个trace_id反复产生新任务，队列积压越来越深，token消耗飙升，直到把预算烧穿。根本原因是协调者的终止条件没写好，比如收到结果后无脑继续下发，或某个智能体出错后不断重试同一个动作。

我的解决办法是在系统层面硬编码护栏。字段里加上max_depth和max_total_tasks，协调者每次拆分任务前检查当前任务深度，超过阈值直接返回错误；单条链路的总任务量超过限制直接熔断；每个智能体重试次数最多三次，超过就把异常抛回协调者，由协调者决定终止或转人工。记住，模型不会自己“停下来”，你必须用代码帮它停下来。

4.3 幻觉在链路上的放大

第三个坑是幻觉的链路级放大。单个智能体产生幻觉，影响是局部的；但在agency-agents架构里，一个智能体的错误结论会被当成“可信输入”传给下一个智能体，下个智能体在错误基础上继续加工，产出看起来好像逻辑自洽，其实整个链条的前提都是虚构的。

我之前做行业研报项目就栽过。研究智能体编了一个不存在的市场数据，写作智能体不仅没校验，还把它当成核心论据展开。最终报告结构完整、语言流畅、结论漂亮，但数据来源是零。从那以后我形成了两条铁律。第一，每个智能体的中间结果必须携带证据来源，比如具体URL或文件ID。第二，智能体的系统提示词里明确写：证据不足时直接输出unknown，绝不猜测。

4.4 调试与可观测性建设

排障体验直接决定这个系统能不能在生产环境活下来。我见过太多团队出了事故才知道加日志，加的时候又忘了trace_id，结果根本无从定位。

我的建议是每位智能体打印的日志都使用结构化JSON，必须包含trace_id、agent_name、msg_id、msg_type、事件名这几个字段。本地开发阶段就把链路追踪脚本写好，拿到一个trace_id，就能按时间线把这个任务经过两个智能体的消息流转完整抽出来回放。这脚本很简单，grep加排序就够用，但价值巨大。

遇到消息丢失或超时问题，优先看消息总线的积压情况，再看智能体是否抛了异常。遇到结果质量不对，先看是哪一步的中间结果开始偏的，找到第一个产错结果的智能体，单独喂给它的原始输入做复现。这套排查思路在改造后帮我把排障时间缩短了至少一半。

4.5 避坑速查表

这张表来自实打实的排障记录，而不是理论推演，很多坑都是反反复复踩过以后总结出来的。

我个人的体会是，agency-agents本质上是一个软件工程问题，而不只是提示词工程问题。多智能体系统里的角色设计、系统提示词固然重要，但真正决定系统稳定性的，是消息协议、状态管理和链路追踪这些硬骨头。我在实践中最满意的一个决策，就是给每条消息烙上trace_id，这个动作在早期看起来“不过就是加个字段”，但后来每一次排障都在感谢当时的自己。架构模式终会过时，但这套“用代码兜底不确定性”的思路，放在任何AI项目里都是通用的。