多 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这个架构的关键特性是完全解耦Orchestrator 是消息生产者和消费者的统一入口。它把用户的复杂需求拆解成子任务发布到对应的 Topic。它同时订阅所有结果 Topic当所有子任务的结果都到达后汇总返回给用户。Worker Agent 是纯粹的消息消费者。它们不知道上游是谁只关心从自己的 Topic 里拿到任务、处理、把结果发到结果 Topic。这种无知正是系统解耦的根基——你可以随时增减 Search Agent 的实例数量来应对流量变化而不影响任何其他 Agent。消息队列天然支持背压。当 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_factorytime.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_iddata.get(task_id, ), senderdata.get(sender, ), task_typedata.get(task_type, ), payloadjson.loads(data.get(payload, {})), statusTaskStatus(data.get(status, pending)), created_atfloat(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, id0, mkstreamTrue ) 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}, count50, block1000 ) 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( groupnameself.CONSUMER_GROUP, consumernameconsumer_id, streams{task_stream: }, countbatch_size, blockblock_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_idagent_msg.task_id, senderconsumer_name, task_typetask_type, payload{result: result, status: success}, statusTaskStatus.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_idagent_msg.task_id, senderconsumer_name, task_typetask_type, payload{error: str(e), status: failed}, statusTaskStatus.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_typesearch, handlerself.handle_task, consumer_nameself.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_idf{request_id}-search, senderorchestrator, task_typesearch, payload{query: user_query}, ) await self.bus.publish_task(search, search_msg) # 2. 发布分析任务依赖搜索完成后的二次处理此处简化 analyze_msg AgentMessage( task_idf{request_id}-analyze, senderorchestrator, task_typeanalyze, 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), timeout15.0, ) analyze_results await asyncio.wait_for( self.bus.wait_for_result(analyze, f{request_id}-analyze), timeout15.0, ) # 4. 汇总 return { request_id: request_id, search: search_results.payload, analyze: analyze_results.payload, } # 启动示例 async def main(): r redis.Redis(decode_responsesTrue) 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_asciiFalse, indent2)}) consumer_task.cancel() await run_system() if __name__ __main__: asyncio.run(main())四、边界分析与架构权衡1. 消息队列选型Redis Streams vs RabbitMQ vs Kafka维度Redis StreamsRabbitMQKafkaAgent 规模20适用100适用100适用消息可靠性ACK PEL丰富的确认模式持久化 分区复制运维复杂度低复用 Redis中高延迟极低微秒级低中对于大多数多 Agent 系统Agent 数量 20Redis Streams 是最务实的选择。当系统需要跨服务部署、消息量达到每秒数千条时才需要考虑 Kafka。2. 消息的顺序性保证Redis Streams 在单 Consumer Group 内保证消息按发送顺序消费。但如果多个 Worker 竞争消费同一个 Topic不同 Worker 的处理速度不同消息处理完成顺序和发送顺序可能不一致。对于先搜索后分析这种有先后依赖的任务应该用不同的 Topictasks.search 和 tasks.analyze由 Orchestrator 显式控制顺序而非依赖队列的内部顺序。3. 死信队列的必要性如果某个任务处理后始终失败、消费者不断重试它会一直占用消费者资源。生产环境必须引入死信队列——将失败超过 N 次的消息转移到专门的 DLQ由运维人员人工排查。4. 同步等待 vs 异步回调wait_for_result是同步等待模式——Orchestrator 发布任务后阻塞等待结果。对于响应式场景如在线对话这很自然。但对于长时间运行的批处理任务如大型数据分析应该改用回调模式——Orchestrator 注册一个结果回调不阻塞等待当结果就绪时由消息总线主动通知。五、总结Producer-Consumer 模式解决的是多 Agent 协作中**谁和谁怎么说话**的问题。三个核心设计决策按任务类型分 Topic而非按 Agent 分 Topic——让 Topic 数量和命名反映业务语义而不是组织架构Consumer Group 实现水平扩展——同一 Topic 多个 Worker 竞争消费无缝扩展处理能力Orchestrator 作为唯一的知道全局的组件——Worker Agent 保持无状态、无依赖降低系统复杂度这套架构最妙的地方在于当你需要增加新的 Agent 类型时你只需要定义一个新的 Topic实现一个新的 Worker而不需要修改任何已有 Agent 的代码。耦合度几乎是零。下一篇聊聊 LlamaIndex 的自定义 Ingestion Pipeline——怎么把文档解析器做成可插拔的插件。