news 2026/7/8 14:49:46

多 Agent 消息路由：Producer-Consumer 模式在 Agent 协作中的工程实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
多 Agent 消息路由：Producer-Consumer 模式在 Agent 协作中的工程实践

多 Agent 消息路由：Producer-Consumer 模式在 Agent 协作中的工程实践

一、深度引言与场景痛点

单 Agent 的能力天花板很明显——一个 Agent 既要做信息检索，又要做代码分析，还要做报告生成，系统提示词越写越长，工具列表越来越多，结果就是"样样通、样样松"。于是多 Agent 协作成了必然趋势：检索 Agent 负责搜资料，分析 Agent 负责处理数据，写作 Agent 负责出报告，每个 Agent 只做自己最擅长的事。

但多 Agent 引入了一个新问题：它们之间怎么通信？

最 naive 的方案是"点对点直连"——Agent A 调用 Agent B 的函数，Agent B 把结果返回给 A。这在 2-3 个 Agent 时还算直观，但当 Agent 数量涨到 10 个甚至更多时，点对点的依赖关系会变成一张密密麻麻的蜘蛛网。更麻烦的是，Agent 的执行速度差异很大——检索 Agent 可能需要 3 秒返回结果，而写作 Agent 可能在 0.1 秒内就完成了。如果检索 Agent 被阻塞等待写作 Agent 的某个中间结果，整个系统的吞吐量就会被拖累。

Producer-Consumer（生产者-消费者）模式提供了优雅的解法。它的核心思想是：Agent 之间不直接通信，而是通过一个"消息中间人"（Message Broker）解耦——每个 Agent 只需要关心"我要往哪个 Topic 发消息"和"我要从哪个 Topic 收消息"，不需要知道对面是谁、有几个、状态如何。

二、底层机制与原理深度剖析

下面这张图展示了基于 Producer-Consumer 模式的多 Agent 协作架构：

flowchart TD U[用户请求] --> O[Orchestrator Agent] O -->|发布任务| MQ1[(Topic: tasks.search)] O -->|发布任务| MQ2[(Topic: tasks.analyze)] O -->|发布任务| MQ3[(Topic: tasks.report)] subgraph Workers [Agent Worker 集群] SA1[Search Agent 1] SA2[Search Agent 2] AA1[Analyze Agent 1] AA2[Analyze Agent 2] RA1[Report Agent 1] end MQ1 -->|消费| SA1 MQ1 -->|消费| SA2 SA1 -->|结果| MQ4[(Topic: results.search)] SA2 -->|结果| MQ4 MQ2 -->|消费| AA1 MQ2 -->|消费| AA2 AA1 -->|结果| MQ5[(Topic: results.analyze)] AA2 -->|结果| MQ5 MQ3 -->|消费| RA1 RA1 -->|结果| MQ6[(Topic: results.report)] MQ4 -->|订阅| O MQ5 -->|订阅| O MQ6 -->|订阅| O O -->|汇总结果| U style Workers fill:#e3f2fd style MQ1 fill:#fff3e0 style MQ2 fill:#fff3e0 style MQ3 fill:#fff3e0

这个架构的关键特性是完全解耦：

  1. Orchestrator 是消息生产者和消费者的统一入口。它把用户的复杂需求拆解成子任务，发布到对应的 Topic。它同时订阅所有结果 Topic，当所有子任务的结果都到达后，汇总返回给用户。

  2. Worker Agent 是纯粹的消息消费者。它们不知道上游是谁，只关心从自己的 Topic 里拿到任务、处理、把结果发到结果 Topic。这种"无知"正是系统解耦的根基——你可以随时增减 Search Agent 的实例数量来应对流量变化，而不影响任何其他 Agent。

  3. 消息队列天然支持背压。当 Search Agent 处理不过来时，任务积压在队列中，Analyze Agent 不会受影响——它有自己的队列和消费节奏。

三、生产级代码实现

下面基于 Redis Streams 实现一套完整的多 Agent 消息路由系统。选择 Redis Streams 而非 RabbitMQ/Kafka 的原因是：对于 10 个以内 Agent 的中等规模系统，Redis 足够用且运维成本极低。

import asyncio import json import uuid import time import logging from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional from enum import Enum import redis.asyncio as redis logger = logging.getLogger(__name__) class TaskStatus(Enum): PENDING = "pending" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" @dataclass class AgentMessage: """Agent 间消息""" task_id: str sender: str # 发送方 Agent 名称 task_type: str # 任务类型 payload: Dict[str, Any] status: TaskStatus = TaskStatus.PENDING created_at: float = field(default_factory=time.time) def to_dict(self) -> dict: return { "task_id": self.task_id, "sender": self.sender, "task_type": self.task_type, "payload": json.dumps(self.payload), "status": self.status.value, "created_at": self.created_at, } @classmethod def from_dict(cls, data: dict) -> "AgentMessage": return cls( task_id=data.get("task_id", ""), sender=data.get("sender", ""), task_type=data.get("task_type", ""), payload=json.loads(data.get("payload", "{}")), status=TaskStatus(data.get("status", "pending")), created_at=float(data.get("created_at", 0)), ) class AgentMessageBus: """基于 Redis Streams 的 Agent 消息总线 实现 Producer-Consumer 模式： - 每个任务类型对应一个任务 Stream 和一个结果 Stream - Consumer Group 支持多个 Worker 竞争消费 - Stream 的 ack 机制保证消息不丢失 """ STREAM_TASK_PREFIX = "agent:task:" STREAM_RESULT_PREFIX = "agent:result:" CONSUMER_GROUP = "agent-workers" def __init__(self, redis_client: redis.Redis): self.redis = redis_client # ===== 基础设施 ===== async def ensure_streams(self, task_types: List[str]): """确保所有 Stream 和 Consumer Group 已创建""" for task_type in task_types: task_stream = f"{self.STREAM_TASK_PREFIX}{task_type}" try: await self.redis.xgroup_create( task_stream, self.CONSUMER_GROUP, id="0", mkstream=True ) except redis.ResponseError as e: if "BUSYGROUP" not in str(e): raise # ===== Producer 端（Orchestrator） ===== async def publish_task(self, task_type: str, msg: AgentMessage) -> str: """发布任务到指定 Topic""" task_stream = f"{self.STREAM_TASK_PREFIX}{task_type}" msg_id = await self.redis.xadd(task_stream, msg.to_dict()) logger.debug(f"任务发布: {msg.task_id} → {task_stream} (id={msg_id})") return msg_id async def wait_for_result( self, task_type: str, task_id: str, timeout: float = 30.0 ) -> Optional[AgentMessage]: """阻塞等待指定任务的结果""" result_stream = f"{self.STREAM_RESULT_PREFIX}{task_type}" deadline = time.monotonic() + timeout while time.monotonic() < deadline: # 读取结果 Stream（从尾部开始） results = await self.redis.xread( {result_stream: "0-0"}, count=50, block=1000 ) for stream_name, messages in results: for msg_id, msg_data in messages: agent_msg = AgentMessage.from_dict(msg_data) if agent_msg.task_id == task_id: await self.redis.xdel(result_stream, msg_id) return agent_msg await asyncio.sleep(0.1) raise asyncio.TimeoutError(f"等待任务结果超时: {task_id}") # ===== Consumer 端（Worker Agent） ===== async def consume_tasks( self, task_type: str, handler: Callable[[AgentMessage], Any], consumer_name: str, block_ms: int = 5000, batch_size: int = 10, ): """消费者循环：持续拉取并处理任务 Args: task_type: 消费的任务类型 handler: 任务处理函数，接收 AgentMessage，返回处理结果 consumer_name: 消费者名称（同一 Consumer Group 下的唯一 ID） """ task_stream = f"{self.STREAM_TASK_PREFIX}{task_type}" result_stream = f"{self.STREAM_RESULT_PREFIX}{task_type}" consumer_id = f"{self.CONSUMER_GROUP}:{consumer_name}" logger.info(f"消费者启动: {consumer_name} 监听 {task_stream}") while True: try: # 从 Consumer Group 拉取新任务 # ">" 表示只取未分发给任何消费者的新消息 messages = await self.redis.xreadgroup( groupname=self.CONSUMER_GROUP, consumername=consumer_id, streams={task_stream: ">"}, count=batch_size, block=block_ms, ) for stream_name, msgs in messages: for msg_id, msg_data in msgs: agent_msg = AgentMessage.from_dict(msg_data) try: # 标记为处理中 agent_msg.status = TaskStatus.PROCESSING # 调用业务处理函数 result = await handler(agent_msg) if asyncio.iscoroutine(result): result = await result # 将处理结果发布到结果 Stream result_msg = AgentMessage( task_id=agent_msg.task_id, sender=consumer_name, task_type=task_type, payload={"result": result, "status": "success"}, status=TaskStatus.COMPLETED, ) await self.redis.xadd(result_stream, result_msg.to_dict()) # ACK 任务 await self.redis.xack(task_stream, self.CONSUMER_GROUP, msg_id) except Exception as e: logger.error(f"任务处理失败 [{agent_msg.task_id}]: {e}") # 发布失败结果 error_msg = AgentMessage( task_id=agent_msg.task_id, sender=consumer_name, task_type=task_type, payload={"error": str(e), "status": "failed"}, status=TaskStatus.FAILED, ) await self.redis.xadd(result_stream, error_msg.to_dict()) await self.redis.xack(task_stream, self.CONSUMER_GROUP, msg_id) except asyncio.CancelledError: logger.info(f"消费者 {consumer_name} 已停止") break except Exception as e: logger.error(f"消费者异常 [{consumer_name}]: {e}") await asyncio.sleep(1) # ===== 管理接口 ===== async def get_pending_count(self, task_type: str) -> int: """获取待处理任务数""" task_stream = f"{self.STREAM_TASK_PREFIX}{task_type}" info = await self.redis.xinfo_stream(task_stream) return info.get("length", 0) async def get_consumer_info(self, task_type: str) -> Dict: """获取消费者组信息""" task_stream = f"{self.STREAM_TASK_PREFIX}{task_type}" try: info = await self.redis.xinfo_groups(task_stream) return {g["name"]: g for g in info} except Exception: return {} # ================== Worker Agent 示例 ================== class SearchAgent: """搜索 Worker Agent""" def __init__(self, name: str, bus: AgentMessageBus): self.name = name self.bus = bus async def handle_task(self, msg: AgentMessage) -> Dict: """处理搜索任务""" query = msg.payload.get("query", "") logger.info(f"[{self.name}] 开始搜索: {query[:50]}") # 模拟搜索逻辑 await asyncio.sleep(1) return { "results": [ {"title": "搜索结果 1", "score": 0.95}, {"title": "搜索结果 2", "score": 0.88}, ], "query": query, } async def run(self): """启动消费者循环""" await self.bus.consume_tasks( task_type="search", handler=self.handle_task, consumer_name=self.name, ) # ================== Orchestrator 示例 ================== class Orchestrator: """编排 Agent：拆解任务、分发、汇总""" def __init__(self, bus: AgentMessageBus): self.bus = bus async def process_request(self, user_query: str) -> Dict: """处理用户请求""" request_id = str(uuid.uuid4())[:8] # 1. 发布搜索任务 search_msg = AgentMessage( task_id=f"{request_id}-search", sender="orchestrator", task_type="search", payload={"query": user_query}, ) await self.bus.publish_task("search", search_msg) # 2. 发布分析任务（依赖搜索完成后的二次处理，此处简化） analyze_msg = AgentMessage( task_id=f"{request_id}-analyze", sender="orchestrator", task_type="analyze", payload={"query": user_query}, ) await self.bus.publish_task("analyze", analyze_msg) # 3. 等待所有结果 search_results = await asyncio.wait_for( self.bus.wait_for_result("search", f"{request_id}-search"), timeout=15.0, ) analyze_results = await asyncio.wait_for( self.bus.wait_for_result("analyze", f"{request_id}-analyze"), timeout=15.0, ) # 4. 汇总 return { "request_id": request_id, "search": search_results.payload, "analyze": analyze_results.payload, } # ================== 启动示例 ================== async def main(): r = redis.Redis(decode_responses=True) bus = AgentMessageBus(r) # 初始化 Streams await bus.ensure_streams(["search", "analyze", "report"]) # 启动 Worker Agent search_agent = SearchAgent("search-worker-1", bus) # 启动 Orchestrator orchestrator = Orchestrator(bus) # 并行运行消费者和编排器 async def run_system(): consumer_task = asyncio.create_task(search_agent.run()) await asyncio.sleep(0.5) # 等待消费者就绪 result = await orchestrator.process_request("Python asyncio 性能优化") print(f"处理结果: {json.dumps(result, ensure_ascii=False, indent=2)}") consumer_task.cancel() await run_system() if __name__ == "__main__": asyncio.run(main())

四、边界分析与架构权衡

1. 消息队列选型：Redis Streams vs RabbitMQ vs Kafka

维度Redis StreamsRabbitMQKafka
Agent 规模<20（适用）<100（适用）100+（适用）
消息可靠性ACK + PEL丰富的确认模式持久化 + 分区复制
运维复杂度低（复用 Redis）中高
延迟极低（微秒级）低中

对于大多数多 Agent 系统（Agent 数量 < 20），Redis Streams 是最务实的选择。当系统需要跨服务部署、消息量达到每秒数千条时，才需要考虑 Kafka。

2. 消息的顺序性保证

Redis Streams 在单 Consumer Group 内保证消息按发送顺序消费。但如果多个 Worker 竞争消费同一个 Topic，不同 Worker 的处理速度不同，消息处理完成顺序和发送顺序可能不一致。对于"先搜索后分析"这种有先后依赖的任务，应该用不同的 Topic（tasks.search 和 tasks.analyze），由 Orchestrator 显式控制顺序，而非依赖队列的内部顺序。

3. 死信队列的必要性

如果某个任务处理后始终失败、消费者不断重试，它会一直占用消费者资源。生产环境必须引入死信队列——将失败超过 N 次的消息转移到专门的 DLQ，由运维人员人工排查。

4. 同步等待 vs 异步回调

wait_for_result是同步等待模式——Orchestrator 发布任务后阻塞等待结果。对于响应式场景（如在线对话），这很自然。但对于长时间运行的批处理任务（如大型数据分析），应该改用回调模式——Orchestrator 注册一个结果回调，不阻塞等待，当结果就绪时由消息总线主动通知。

五、总结

Producer-Consumer 模式解决的是多 Agent 协作中**"谁和谁怎么说话"**的问题。三个核心设计决策：

  1. 按任务类型分 Topic（而非按 Agent 分 Topic）——让 Topic 数量和命名反映业务语义，而不是组织架构
  2. Consumer Group 实现水平扩展——同一 Topic 多个 Worker 竞争消费，无缝扩展处理能力
  3. Orchestrator 作为唯一的"知道全局"的组件——Worker Agent 保持无状态、无依赖，降低系统复杂度

这套架构最妙的地方在于：当你需要增加新的 Agent 类型时，你只需要定义一个新的 Topic，实现一个新的 Worker，而不需要修改任何已有 Agent 的代码。耦合度几乎是零。

下一篇聊聊 LlamaIndex 的自定义 Ingestion Pipeline——怎么把文档解析器做成可插拔的插件。

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/7/8 14:48:03

基于ISOM8710与STM32的高压隔离通信系统设计

1. 项目背景与核心需求 在工业自动化、电力系统和医疗设备等领域&#xff0c;高压电路与低压控制系统的安全隔离是确保设备可靠运行和人员安全的关键技术。传统的光耦隔离方案存在速度慢、寿命有限等缺陷&#xff0c;而数字隔离器芯片配合微控制器的方案正在成为新一代解决方案…

作者头像 李华
网站建设 2026/7/8 14:46:23

暗黑破坏神2存档编辑器完整指南：5步快速修改游戏数据

暗黑破坏神2存档编辑器完整指南&#xff1a;5步快速修改游戏数据 【免费下载链接】d2s-editor 项目地址: https://gitcode.com/gh_mirrors/d2/d2s-editor 你是否曾经在暗黑破坏神2中因为角色属性分配不当而卡关&#xff1f;是否梦想过拥有完美的装备组合却不愿花费数百…

作者头像 李华
网站建设 2026/7/8 14:44:25

Windows系统文件capiprovider.dll丢失找不到问题解决

在使用电脑系统时经常会出现丢失找不到某些文件的情况&#xff0c;由于很多常用软件都是采用 Microsoft Visual Studio 编写的&#xff0c;所以这类软件的运行需要依赖微软Visual C运行库&#xff0c;比如像 QQ、迅雷、Adobe 软件等等&#xff0c;如果没有安装VC运行库或者安装…

作者头像 李华
网站建设 2026/7/8 14:41:32

从复位向量到内核入口：U-Boot SPL 到 Kernel 启动全链路逐阶段剖析

从复位向量到内核入口&#xff1a;U-Boot SPL 到 Kernel 启动全链路逐阶段剖析 一、嵌入式 Linux 启动的"黑箱时刻"&#xff1a;当板卡上电后迟迟看不到控制台输出 大多数 BSP 工程师都经历过这样的场景&#xff1a;新板卡贴片回来&#xff0c;上电后串口静悄悄&…

作者头像 李华