多Agent协作与分布式消息总线：群体智能AI Agent落地实践

今天聊一个我自己前后折腾了快三周的项目：群体智能 AI Agent。简单说，它不是靠某个超强单体大模型硬扛所有任务，而是让多个 LLM 协作，每个模型扮演一个角色，通过消息总线收发信息，把一次复杂请求拆成一条分布式处理流水线。整个系统跑在普通笔记本加一台小服务器上，实测能把单模型反复失败的长链路任务稳定完成。这篇主要记录设计思路、核心实现和踩过的坑，适合正在研究多 Agent 协作、分布式系统落地的朋友参考，全程可复现。

1. 设计思路：为什么非要“多个 LLM 协作”

1.1 单模型的瓶颈不是“不够聪明”，而是被上下文卡死

一开始我并没有想做群体智能，因为单体大模型已经能处理不少复杂任务。真正让我改变看法的，是一次失败的实测：让单个模型完成“从需求描述到生成一份完整数据分析报告”的长链路任务，需要它自行拆步骤、写代码、执行、看报错、再修。结果前几步还算正常，到第五轮交互时，上下文里的历史指令、中间结果、报错信息混在一起，模型开始“失忆”——它忘了一开始的需求约束，不断重复同一个错误修复方案，甚至自己编造不存在的库接口。

问题本质不是模型笨，而是上下文失控。长任务天然要求模型同时记住目标、约束、若干中间结果和工具状态，这远超普通模型的有效注意力窗口。群体智能的核心思路恰恰是用“组织分工”对冲这个物理限制：把一段超长上下文拆成若干个短上下文，每人只记自己负责的那一小段，再通过明确的消息协议让信息在模型之间流动。每个模型面对的始终是简短、聚焦的对话，准确率反而比一个模型“死撑到底”高得多。

1.2 群体智能的三板斧：分工、辩论、投票

群体智能听起来玄，落到工程上其实就是三种模式，我整理成了表格。绝大多数多 Agent 产品都能归到这三类里的某一类，或者它们的组合：

我的项目不是只选一种模式，而是把三种都用上了：规划 Agent 拆任务，研究 Agent 和编码 Agent 分别产出方案，评审 Agent 对结果做对抗式质疑，最后由聚合模块投票决定是否采纳。这已经不只是“多个模型一起干活”，而是真正在构建一个“微型决策组织”。

1.3 为什么选择分布式架构：隔离、扩展、故障恢复

确定了多 Agent 协作，紧接着的问题是怎么组织这些 Agent。一个最省事的方案是在单进程里开多线程，每个线程跑一个 Agent，共享一个全局上下文变量。我第一版就这么干的，结果很惨：线程之间互相污染历史消息，A Agent 的 prompt 碎片串到 B Agent 的消息里，调试了两个晚上才定位到一个共享列表被并发修改。

换了分布式架构之后，三个问题同时解决。第一是隔离性，每个 Agent 独立进程或容器，上下文、临时文件、工具调用状态互不可见，物理上杜绝了串扰。第二是可扩展性，需求上来时加节点就可以，瓶颈模型单独横向扩容，不用整个系统重写。第三是故障恢复，某个 Agent 崩了，只需要重启它这一个节点，其他节点照常运行，配合消息队列的确认机制还可以重新投递消息。代价是系统复杂度上升，但对“可靠完成长任务”这个目标来说，这笔投资很值。

2. 技术选型与架构设计

2.1 编排框架选型：自研协议 vs 现成框架

市面上已经有一批多 Agent 编排框架，名气大的那几个我也跑过 demo。它们封装修得挺漂亮，但一旦要定制协作流程，就得顺着框架的思路写代码。框架替你解决的问题和你自己要解决的问题，经常不是一个问题。我自己写这套系统时，选择了一个“半自研”的路线：Agent 通信协议、注册中心、生命周期管理自己写，底层只依赖两个成熟组件——消息总线用 Redis Streams，运行时节点用轻量级进程管理。

这个选择的好处是能把原理吃透。消息怎么路由、消息丢失怎么补、Agent 状态怎么保存，每一层都是自己可控的，出问题可以直接查。代价是很多东西要重复造轮子，比如超时重试、日志追踪、分布式锁。如果你不是想深入学架构，而是要在三天内交付一个功能，直接选现成框架更划算；如果你想真正理解群体智能运行时的每一处细节，我强烈建议自研一次。

2.2 消息总线选了 Redis Streams，而不是普通队列

Agent 之间需要可靠地传递消息，很多人第一个想到的是 RabbitMQ 或 Kafka。但对于这种规模的项目，Redis Streams 是更合适的选择。为什么不用普通 Redis List 左进右出当队列？因为缺三个关键能力：消费组（一组 Agent 协作消费同一批消息）、消息确认（处理失败不丢消息）、历史读取（新启动的 Agent 可以重放已消费消息）。这三个能力 Streams 原生就支持，配置非常简单，不需要额外引入重型消息中间件。

我当时画系统草图，消息流大致是这样：入口任务先发给规划 Agent，规划 Agent 产出任务分解后，把子任务消息分别发到研究 Agent 和编码 Agent 的订阅通道，它们处理完再发结果给评审 Agent，评审通过后由聚合器输出最终结果。整个过程是事件驱动的，每个环节只消费自己关心的消息类型，做到完全解耦。Redis Streams 的消费者组还天然支持同一个 Agent 起多个实例，任务可以横向平分。

2.3 模型接入层：把各家模型抽象成同一个函数

多 Agent 系统里最容易被低估的组件是模型接入层。项目里同时用了云端模型和本地部署的开源模型，不同模型的服务地址、鉴权方式、参数命名都不一样。如果每个 Agent 自己直接调 SDK，后期换模型就是灾难。我在底层做了一层统一路由，屏蔽所有差异。

下面是简化版的模型路由核心代码，我实际项目里就是长这样，只是加了一些监控逻辑：

# llm_router.py import os from openai import OpenAI _clients = {} def get_client(provider: str) -> OpenAI: if provider not in _clients: if provider == "cloud": _clients[provider] = OpenAI(api_key=os.getenv("CLOUD_API_KEY")) else: _clients[provider] = OpenAI( base_url="http://localhost:8000/v1", api_key="local-key" ) return _clients[provider] def chat(provider: str, messages: list, temperature: float = 0.7) -> str: model_map = { "cloud": os.getenv("CLOUD_MODEL", "default-model-v1"), "local": os.getenv("LOCAL_MODEL", "local-model-name"), } resp = get_client(provider).chat.completions.create( model=model_map[provider], messages=messages, temperature=temperature, max_tokens=2048 ) return resp.choices[0].message.content

所有 Agent 只依赖 chat 这个函数，不关心底层是哪个厂商的模型。本地模型和云端模型可以按任务难度分流，简单任务走便宜的本地模型，复杂推理走云端模型。这个抽象带来的收益是后续给系统换新模型时，只需要加一行配置，完全不需要改动任何 Agent 代码。

2.4 状态存储与任务追踪：每一步都要能被回溯

分布式系统最怕黑盒。多个 Agent 各自跑完，最后拼结果时发现答案不对，你根本不知道是哪一步出了问题。所以我在设计阶段就强制要求：每一个 Agent 的输入、输出、调用模型的消息列表、耗时、重试次数，全部写入 Redis 里的结构化记录。任务跑完可以一键生成完整的事故复盘报告。

很多人会觉得这是过度设计，但实测下来这恰恰是整个系统最重要的部分。有一次评审 Agent 给出错误结论，我通过状态记录发现，它收到编码 Agent 的代码片段时，JSON 在传输过程中被截断了。如果没有全量状态记录，这个 bug 可能要排查半天；有了记录，三十秒定位到是序列化问题。

3. 从零到一：实现一个可复用的多 Agent 协作系统

3.1 项目结构与运行流程

我先说明项目的最终目录结构，方便你建立一个整体认知。我不是追求代码量最少，而是追求“每个文件职责单一”，调试时能快速找到对应环节。

agent-swarm/ ├── agent.py # Agent 抽象基类与注册表 ├── bus.py # Redis Streams 消息总线封装 ├── llm_router.py # 统一模型接入层 ├── state_store.py # 状态读写模块 ├── roles/ │ ├── planner.py # 规划者 Agent │ ├── researcher.py # 研究员 Agent │ ├── coder.py # 编码员 Agent │ └── reviewer.py # 评审员 Agent └── main.py # 启动入口与拓扑编排

运行流程分三层：入口层接收用户任务，写入 Redis Stream；调度层根据任务类型，决定创建哪些 Agent 实例并下发启动消息；执行层的各 Agent 消费各自通道里的任务消息，完成后把结果写回总线。我用了三种角色命名的流：plan_stream、work_stream、review_stream。规划 Agent 消费 plan_stream，研究员和编码员消费 work_stream，评审员消费 review_stream。

3.2 Agent 抽象基类与注册中心

不管哪个角色，Agent 都有几个共同行为：接收消息、调用模型、产出结果。所以抽象基类里先把这些写死，子类只实现自己的业务逻辑。这一步做得好，后面加新角色非常轻松。

# agent.py import json from abc import ABC, abstractmethod from llm_router import chat class LLMAgent(ABC): def __init__(self, name: str, system_prompt: str, provider: str = "cloud"): self.name = name self.system_prompt = system_prompt self.provider = provider self.history = [] def think(self, user_message: str, temperature: float = 0.7) -> str: self.history.append({"role": "user", "content": user_message}) messages = [{"role": "system", "content": self.system_prompt}] + self.history reply = chat(self.provider, messages, temperature=temperature) self.history.append({"role": "assistant", "content": reply}) return reply @abstractmethod def handle(self, message: dict) -> dict: pass class AgentRegistry: def __init__(self): self._agents = {} def register(self, agent: LLMAgent): self._agents[agent.name] = agent def get(self, name: str) -> LLMAgent: return self._agents[name]

有个细节值得注意：每个 Agent 的 history 是自己私有的，不会共享给其他 Agent。这也呼应了前面说的上下文隔离原则。如果你想做“记忆共享”或“全局记忆”，需要显式设计一个记忆存储，而不是让两个 Agent 直接读同一个数组，否则一定会出现上下文污染。

3.3 消息总线的生产与消费封装

Redis Streams 的消息格式我是这样约定的：每条消息是一个 JSON，包含消息 ID、目标 Agent、消息类型、载荷数据。bus.py 负责三件事：向指定流发布消息、从指定流消费消息、确认消息已处理完成。

# bus.py import json import redis import uuid class MessageBus: def __init__(self, stream: str, group: str = "agent_group"): self.redis = redis.Redis(host="localhost", port=6379, decode_responses=True) self.stream = stream self.group = group try: self.redis.xgroup_create(stream, group, id="0", mkstream=True) except Exception: pass # 消费组已存在 def publish(self, target: str, msg: dict): payload = { "target": target, "msg_id": str(uuid.uuid4()), "data": json.dumps(msg) } self.redis.xadd(self.stream, payload) def consume(self, consumer: str, batch_size: int = 5, block_ms: int = 5000): while True: entries = self.redis.xreadgroup( self.group, consumer, {self.stream: ">"}, count=batch_size, block=block_ms ) if not entries: break for _, items in entries: for msg_id, fields in items: yield msg_id, json.loads(fields["data"]) self.redis.xack(self.stream, self.group, msg_id)

这里有一个我踩过的坑：如果消费端代码在消息处理中途抛了异常，消息不会被确认，会一直留在 pending 消息列表里。重跑时 Redis 会把同一批消息再次投递给消费者，可能造成重复处理。所以消费者的处理逻辑必须做“幂等设计”——收到重复消息时，结合消息 ID 判断是否已经处理过。我的做法是在状态存储里记录消息 ID 的消费指纹，重复消息直接跳过。

3.4 一个能跑通的五角色协作 Demo

纸上谈兵结束，上实战。我搭了一个“撰写数据分析报告”的完整任务，五个 Agent 分工合作：规划者拆解步骤，研究员找方法和数据源，编码员生成代码，评审员做对抗检查，最后由入口程序聚合结果。

规划 Agent 的 handle 方法核心逻辑是让模型输出 JSON 格式的任务分解：

# roles/planner.py import json from agent import LLMAgent class PlannerAgent(LLMAgent): def __init__(self): super().__init__( name="planner", system_prompt=( "你是项目规划者。把用户需求拆成不超过4个可并行子任务。" "必须输出JSON，格式：" '[{"agent": "researcher|coder", "task": "..."}]' ), provider="cloud" ) def handle(self, message: dict) -> dict: raw = self.think(message["task"], temperature=0.3) tasks = json.loads(raw) return {"type": "plan_result", "tasks": tasks}

注意温度设了 0.3，规划任务要求确定性，不让模型自由发挥。编码任务同样是低温度，而评审任务我故意把温度调到 0.8，希望评审员能从不同角度找毛病。

研究员 Agent 和编码员 Agent 的代码结构类似，区别只在 system prompt 和处理的输入输出。我把它们的核心代码合并展示：

# roles/researcher.py 和 roles/coder.py 的简化版 from agent import LLMAgent class ResearcherAgent(LLMAgent): def __init__(self): super().__init__( name="researcher", system_prompt="你是研究员，负责提供背景资料和可行的技术方案。请给出具体来源。", provider="local" ) def handle(self, message: dict) -> dict: research = self.think(f"针对任务：{message['task']} 输出简介方案") return {"type": "research_result", "content": research} class CoderAgent(LLMAgent): def __init__(self): super().__init__( name="coder", system_prompt="你是资深编码员，只输出可执行代码，并附带简短解释。", provider="cloud" ) def handle(self, message: dict) -> dict: code = self.think(f"请实现方案中的编码部分：{message['task']}") return {"type": "code_result", "content": code}

评审 Agent 负责找出前面环节的漏洞，这是对抗式设计的关键：

# roles/reviewer.py from agent import LLMAgent class ReviewerAgent(LLMAgent): def __init__(self): super().__init__( name="reviewer", system_prompt=( "你是严格的评审员，不负责写答案，只负责挑问题。" "逐条列出研究方案或代码中的缺陷、风险、遗漏点。" "如果没有问题，请明确回复：APPROVED" ), provider="cloud" ) def handle(self, message: dict) -> dict: review = self.think(f"评审以下内容：\n{message['content']}", temperature=0.8) return {"type": "review_result", "content": review}

main.py 的启动逻辑是这样的：先注册所有 Agent，然后把任务发布到 plan_stream，规划 Agent 消费后把子任务分别发布到 work_stream 的对应消费组。研究员和编码员各自消费，结果进入 review_stream，评审员产出结论。最终聚合器检查评审结果里是否包含 APPROVED，如果包含，直接汇总；如果不包含，把评审意见作为新需求重新投递给规划者，形成一次迭代循环。

实际跑下来，一个原本单模型按 20 轮交互都可能失败的“报告生成”任务，用这套协作系统通常 2 到 3 轮内部迭代就能完成。单次处理时间从原来的 8 分钟下降到 4 分钟左右，因为多个环节是并行推进的。

3.5 让 Agent 真正能调用外部工具：轻量 MCP 接入

只有语言能力没有工具能力的 Agent 价值少一半。编码员 Agent 不能真的运行代码，评审意见就只能是猜。我在系统里接入了工具调用层，没有引入太重的外部协议，而是做了一个简化实现：每个 Agent 可以声明自己能用哪些工具，消息总线增加一类 tool_call 消息。Agent 收到工具调用指令后，执行工具代码，把结果写回上下文，再返回给模型。

工具注册的代码大概是这样：

# tools.py _TOOLS = {} def register_tool(name: str, func): _TOOLS[name] = func def run_tool(name: str, args: dict): if name in _TOOLS: return _TOOLS[name](**args) raise ValueError(f"unknown tool: {name}")

我给编码员注册了一个 execute_python 工具，本质上是沙箱里跑 Python 代码并把 stdout 和 stderr 返回。实践中有两点强烈建议：工具返回值必须截断长度，否则模型上下文会爆炸；工具执行必须有超时和资源限制，否则一段死循环代码能把整台机器拖死。如果你的生产环境需要更规范的工具协议，可以直接上 MCP 标准，本质思路一样，只是多一层规范封装。

4. 真实环境下的坑与排查实录

4.1 Agent 上下文污染：多 Agent 系统最大的隐藏坑

单进程里开多线程那个版本就不提了。就算换到分布式架构，上下文污染仍然可能出现，而且更隐蔽。有一次研究员和评审员共用一个模型供应商的 API 并发请求，我用的 SDK 内部有一个线程池，结果把两条请求的对话记录交叉了。研究员明明在讨论数据分析，模型的回复里却冒出评审员上一轮的“代码规范”内容。

排查了半天，最终定位是模型接入层缺少请求级别的隔离。修复方式是每个请求创建独立的消息列表，禁止共用同一个变量；同时给每个请求加 request_id 透传，日志里能按 ID 串联一个 Agent 的完整调用链。想排查这类问题，日志里必须同时记录 Agent 名称、请求 ID、模型名称、消息条数，缺一样都可能让你怀疑人生。

4.2 分布式多 Agent 也会死锁，而且更隐蔽

单机死锁是线程互相等锁，分布式死锁则发生在“消息互相等待”。典型场景：研究员 A 给评审员 B 发了评审请求，同时评审员 C 正等待研究员 D 补充材料，而研究员 D 又在等研究员 A 的中间结果。三方互相等待，整个协作流程卡死，表现就是所有 Agent 都处于空闲，没有新消息产生，也没有报错。

这种死锁最难排查，因为没有任何异常堆栈。我的解决思路两条腿走路：第一，所有跨 Agent 消息消费必须设置超时时间，我的线下环境统一设 60 秒，超过就抛出超时信号；第二，系统里内置一个心跳监控模块，每 10 秒检查一次所有 Agent 最近一次活跃时间，超过 120 秒没有动静就触发唤醒流程，向阻塞 Agent 发送一条降级指令，让它跳过当前消息给个临时结果。虽然临时结果质量打折，但系统不会整体卡死。

4.3 错误隔离：一个 Agent 崩了不能拖垮整条链路

进程崩溃还好处理，重启就行。真正麻烦的是 Agent 没崩但状态异常了，比如某个模型供应商临时限流，所有请求都超时重试。如果不做隔离，重试逻辑会不断叠加请求量，反而把供应商接口彻底打挂，然后连锁反应所有 Agent 都开始超时。

我在每个 Agent 节点上加了熔断器，连续失败超过 3 次就停止调用该模型 30 秒，直接放回一个“服务降级”消息给上游，由上游决定是换个模型还是换个 Agent。实际跑下来，这套熔断机制至少救了我四五次，全是联调测试时赶上外部服务不稳定。不要以为只有大流量才会熔断，内部系统的偶发抖动一样需要熔断，否则一次小故障会被重试放大成全局故障。

4.4 调试多 Agent 系统唯一好用的办法：全量日志与确定性重放

我调试单 Agent 的时候很依赖“断点 + 交互式观察”，但多 Agent 分布式环境根本没有这个条件。消息是异步的，你打断点只能停住一个节点，其他节点还在继续跑。最后我总结出一个可行的笨办法：出问题时，从状态存储里导出完整消息链路，然后写一个 replay 脚本，按时间顺序把每条消息重新输入给对应 Agent。因为所有 Agent 都是确定性函数加随机温度，温度参数固定到 0 做 replay，就能复现当时的问题。

这个 playback 能力帮我定位过至少三个匪夷所思的 bug：一个是消息在 JSON 序列化时浮点数精度丢失，一个是时间字段在不同 Agent 里的时区解析不一致，一个是某条超长消息在 Streams 持久化时被自动截断。如果没有 replay，这些 bug 只看日志根本发现不了。所以状态存储不仅仅是为了“监控”，它还是分布式调试的地基。

4.5 常见问题速查表

这个表是我真实调试过程中提炼出来的，并不是通用排查手册。多 Agent 系统的特征就是“面多加水，水多加面”，上下文、消息、工具调用、模型选型互相耦合，调试时要有“从数据链路找问题”的意识，而不是单纯堆日志。

收尾：聊点系统之外的真实体会

折腾完这个系统，我最大感受是：多 Agent 协作的瓶颈不在模型能力，而在工程基建。模型再聪明，消息总线不稳定、状态存储不可靠、上下文互相污染，系统照样跑不起来。反过来说，只要把这套消息和状态的底座打磨好，哪怕用的都是开源小模型，协作效果也能明显超过一个闭源大模型单打独斗。

如果你也想自己搭一套类似系统，我的建议是从小处开始：先在一台机器上用内存队列跑通两个 Agent 的协作，感受一下“消息流”和“函数调用”的思维差异，再逐步引入 Redis Streams、独立进程、状态存储和熔断机制。千万不要第一天就上来画五张 Agent 拓扑图、上 K8s，分布式复杂度会把你淹没在无穷无尽的细节里。

另外一个私人心得：给模型下指令时，输出格式的约束一定要写进 system prompt，并配合 JSON 解析和校验。Agent 之间的消息不是给人看的散文，而是给另一个模型吃的结构化数据。第一次运行三个 Agent 时，我吃过很多次“模型输出一大段废话导致 JSON 解析失败”的亏，后来一律要求模型只输出 JSON，并且解析失败时自动重试一次，成功率会大幅提升。

做群体智能系统，本质上是在驯服“一群不完美但各有所长的个体”。过程中踩坑是常态，但只要爬过一次坑，你会发现自己对模型协作、分布式消息、故障恢复的理解，比读十篇论文都扎实。这套系统后续我还在迭代，下一步计划给每个 Agent 加上长期记忆模块，让它们在跨任务协作时能引用历史交互经验。期待这个方向有更多实践出现。