Agent 架构从单体到分布式多 Agent 系统怎么拆分和编排部署一、深度引言与场景痛点大家好我是赵咕咕一名每天在 Agent 堆里摸爬滚打的工程师。我有一个项目最开始只有一个 Agent —— 负责回答用户的产品咨询问题。那时候它住在一台 4C8G 的小服务器上吃得好睡得香QPS 不过百日子过得相当滋润。后来业务加码了老板说我们要支持实时数据分析于是加了一个数据分析 Agent。市场部说用户画像很重要又加了一个推荐 Agent。运营说售后也需要智能化于是售后 Agent 也来了。到我反应过来的时候这台小服务器上已经跑着七个 Agent内存占用 90%启动一个任务能把 CPU 拉满响应延迟从 200ms 飙到了 5 秒。单体 Agent 架构的甜蜜期结束了。拆还是死这是一个问题。今天我们就聊透多 Agent 系统怎么从单体拆到分布式以及拆分后的编排、部署、监控怎么搞。这不是学院派的理论堆砌而是我从一次次凌晨故障中总结的实战经验。二、底层机制与原理深度剖析2.1 单体的天花板在哪里单体 Agent 架构就是所有 Agent 逻辑跑在同一个进程中同一个 Python 解释器、共享同一块内存、共用同一条消息总线。它的问题不是功能上的而是物理上的内存瓶颈每个 Agent 的上下文、工具配置、会话状态都吃内存。七八个 Agent 加上各自的长上下文轻松突破 16GCPU 争抢LLM 推理是 CPU 密集型本地模型或 I/O 密集型API 调用多个 Agent 同时推理GIL 直接教你做人故障爆炸半径一个 Agent 的 bug比如死循环调用工具能把整个进程拖垮所有 Agent 一起挂部署粒度太粗想升级客服 Agent 的逻辑得把推荐、数据、售后 Agent 一起重新部署2.2 拆分策略按什么维度切拆分 Agent 系统的核心问题是边界。我总结出三种最常见的拆分维度flowchart TD A[单体 Agent 系统] -- B{选择拆分维度} B --|按业务领域| C[领域驱动拆分] B --|按执行阶段| D[管道式拆分] B --|按职责类型| E[角色式拆分] C -- C1[客服Agent] C -- C2[推荐Agent] C -- C3[数据分析Agent] C -- C4[售后Agent] D -- D1[意图识别Agent] D1 -- D2[任务规划Agent] D2 -- D3[工具执行Agent] D3 -- D4[回复生成Agent] E -- E1[协调者Agent/Orchestrator] E -- E2[执行者Agent/Worker] E -- E3[评估者Agent/Evaluator] E -- E4[记忆Agent/Memory Keeper] C1 -.-|通过消息队列通信| C2 C3 -.-|通过消息队列通信| C4 D1 -.-|流水线传递| D2 E1 -.-|任务分发| E2 E1 -.-|结果评估| E3领域驱动拆分是最直观的。你按业务边界切每个 Agent 对应一个业务域。好处是各代理独立演进、独立扩缩容。问题是跨领域协作时需要额外的路由层。管道式拆分是按 Agent 执行流程切。意图识别 → 规划 → 执行 → 生成每一步是一个独立服务。这种拆分让每一步都可以独立优化但引入了天然的串行瓶颈。角色式拆分是我的最爱。你有一个协调者Orchestrator做全局调度多个执行者Worker干具体活评估者Evaluator做质量把关记忆 Agent 管理长期存储。这种架构最灵活但协调者本身成为了新的瓶颈点。2.3 通信模式怎么让拆开的 Agent 还能对话Agent 拆分后通信就成了核心问题。我实践过三种通信模式同步 RPC适合低延迟、强依赖的场景。用 gRPC 或 HTTP asyncio 做同步调用。优点是简单直接缺点是调用链长了延迟会叠加。异步消息适合解耦、削峰的场景。用 Redis Stream 或 RabbitMQ 做消息传递。Agent A 把任务丢到队列就不管了Agent B 从队列里取任务执行。缺点是调试困难——你很难追踪一个请求在多个 Agent 之间的完整路径。事件驱动适合广播通知的场景。用 Redis Pub/Sub 或 Kafka 做事件通知。当一个 Agent 完成某个里程碑它广播一个事件感兴趣的 Agent 自行消费。优点是松耦合缺点是事件丢失的问题需要额外处理。三、生产级代码实现下面给出一个基于 asyncio Redis Stream 的多 Agent 编排实现import asyncio import json import uuid from dataclasses import dataclass, field from typing import Any, Protocol import redis.asyncio as redis # ── 通信协议定义 ────────────────────────────────── dataclass class AgentTask: task_id: str field(default_factorylambda: str(uuid.uuid4())) target_agent: str action: str payload: dict[str, Any] field(default_factorydict) parent_task_id: str | None None # 用于追踪调用链 created_at: float field(default_factoryasyncio.get_event_loop().time) dataclass class AgentResult: task_id: str agent_name: str success: bool False output: Any None error: str duration_ms: float 0.0 # ── Agent 基类 ──────────────────────────────────── class BaseAgent(Protocol): 所有 Agent 都必须实现的接口 name: str async def handle(self, task: AgentTask) - AgentResult: ... # ── 消息总线 ────────────────────────────────────── class MessageBus: 基于 Redis Stream 的异步消息总线 def __init__(self, redis_client: redis.Redis, consumer_group: str agents): self._redis redis_client self._consumer_group consumer_group async def publish(self, task: AgentTask) - str: 将任务发布到目标 Agent 的消息队列 stream_key fagent:inbox:{task.target_agent} msg_id await self._redis.xadd(stream_key, { data: json.dumps({ task_id: task.task_id, action: task.action, payload: task.payload, parent_task_id: task.parent_task_id, created_at: task.created_at, }) }) return msg_id async def subscribe(self, agent_name: str, consumer_id: str): 订阅 Agent 的消息队列 stream_key fagent:inbox:{agent_name} try: await self._redis.xgroup_create(stream_key, self._consumer_group, mkstreamTrue) except redis.ResponseError: pass # 消费组已存在 return self._redis.xreadgroup( groupnameself._consumer_group, consumernameconsumer_id, streams{stream_key: }, count1, block5000, # 阻塞等待最多 5 秒 ) async def ack(self, stream_key: str, message_id: str): 确认消息已处理 await self._redis.xack(stream_key, self._consumer_group, message_id) async def publish_result(self, task: AgentTask, result: AgentResult): 将处理结果发布回调度者 stream_key fagent:result:{task.task_id} await self._redis.xadd(stream_key, { data: json.dumps({ task_id: result.task_id, agent_name: result.agent_name, success: result.success, output: result.output, error: result.error, duration_ms: result.duration_ms, }) }) # ── Orchestrator协调者 ───────────────────────── class Orchestrator: 多 Agent 系统的中央调度器 def __init__(self, bus: MessageBus): self._bus bus self._pending_tasks: dict[str, asyncio.Future] {} async def dispatch(self, task: AgentTask) - AgentResult: 分发任务到目标 Agent 并等待结果 loop asyncio.get_running_loop() future: asyncio.Future loop.create_future() self._pending_tasks[task.task_id] future await self._bus.publish(task) try: result await asyncio.wait_for(future, timeout30.0) return result except asyncio.TimeoutError: return AgentResult( task_idtask.task_id, agent_nametask.target_agent, successFalse, errorf任务超时: {task.task_id}, ) finally: self._pending_tasks.pop(task.task_id, None) async def dispatch_parallel( self, tasks: list[AgentTask] ) - dict[str, AgentResult]: 并行分发多个任务 coros [self.dispatch(task) for task in tasks] results await asyncio.gather(*coros, return_exceptionsTrue) result_map {} for task, res in zip(tasks, results): if isinstance(res, Exception): result_map[task.task_id] AgentResult( task_idtask.task_id, agent_nametask.target_agent, successFalse, errorstr(res), ) else: result_map[task.task_id] res return result_map async def resolve_result(self, task_id: str, result: AgentResult): Agent 回传结果时的回调 future self._pending_tasks.get(task_id) if future and not future.done(): future.set_result(result) # ── Agent 运行循环 ──────────────────────────────── async def agent_loop( agent: BaseAgent, bus: MessageBus, orchestrator: Orchestrator, consumer_id: str, ): 每个 Agent 的主循环持续拉取并处理消息 stream_key fagent:inbox:{agent.name} while True: try: messages await bus.subscribe(agent.name, consumer_id) for stream, entries in messages: for msg_id, fields in entries: if stream.decode() ! stream_key: continue task_data json.loads(fields[bdata]) task AgentTask( task_idtask_data[task_id], target_agenttask_data[agent_name], actiontask_data[action], payloadtask_data.get(payload, {}), parent_task_idtask_data.get(parent_task_id), ) start_time asyncio.get_event_loop().time() result await agent.handle(task) result.duration_ms ( asyncio.get_event_loop().time() - start_time ) * 1000 await bus.ack(stream_key, msg_id) await orchestrator.resolve_result(task.task_id, result) except Exception as exc: # 生产环境中应该上报到监控系统 await asyncio.sleep(1) # 防止错误风暴这套代码的要点每个 Agent 是独立的消息消费者通过 Redis Stream 接收任务Orchestrator 负责分发和等待结果支持超时兜底和并行调度agent_loop是每个 Agent 的主循环可以用asyncio.create_task在进程中跑多个消息确认机制保证任务不会丢失四、边界分析与架构权衡拆分的每一步都伴随着取舍以下是几个关键决策点拆还是不拆如果总 QPS 低于 100、Agent 数量少于 5 个、内存占用稳定单体就够用。超过这些指标再考虑分布式。过早拆分只会引入不必要的复杂度。同步还是异步请求-响应场景用 RPC异步批处理场景用消息队列。实际项目中往往是混合的——Orchestrator 对 Worker 是 RPC 调用Worker 之间通过消息队列异步协作。Orchestrator 怎么不成为瓶颈单个 Orchestrator 确实可能成为瓶颈。解决方案有三个层次第一层Orchestrator 本身用 asyncio 做高并发一个进程可以同时等待上千个 Future第二层按业务域做 Orchestrator 分片每个业务域一个独立的调度器第三层用一致性哈希做无状态调度任意一个 Orchestrator 实例都能接手任意任务。状态在哪里多 Agent 系统最大的坑是把会话状态放在 Agent 的内存里。Agent 重启、扩缩容都会丢状态。应该把会话状态外移到 Redis 或数据库中Agent 本身保持无状态。监控和追踪拆分之后一次用户请求可能经过四五个 Agent。没有分布式追踪出了问题你根本不知道是哪个 Agent 拖了后腿。OpenTelemetry Jaeger 是标配。五、总结多 Agent 系统从单体拆到分布式不是一次技术升级而是一次架构思维方式的转变。你不再写一个能处理所有事情的超级 Agent而是构建一个 Agent 之间的协作网络。核心要点单体不够用时果断拆但不要为了拆而拆拆分维度选业务域最自然选管道式最可控选角色式最灵活通信模式按场景选强依赖用 RPC解耦用消息队列广播用事件Orchestrator 要做超时兜底、并发控制、失败重试状态外移、分布式追踪是分布式 Agent 系统的底线拆完之后你会发现原来那些在单体里互相打架的 Agent变成了一群各司其职、协同工作的好队友。这种感觉就像是把一个混乱的大办公室重新装修成了一个个独立又连通的功能间——虽然多了些门和走廊但每个人都知道自己该干什么效率反而上去了。下一篇预告LangChain 内存管理优化根治 ConversationBufferMemory 的内存膨胀问题。