Agent 架构从单体到分布式：多 Agent 系统怎么拆分和编排部署

Agent 架构从单体到分布式：多 Agent 系统怎么拆分和编排部署

一、深度引言与场景痛点

大家好，我是赵咕咕，一名每天在 Agent 堆里摸爬滚打的工程师。

我有一个项目，最开始只有一个 Agent —— 负责回答用户的产品咨询问题。那时候它住在一台 4C8G 的小服务器上，吃得好睡得香，QPS 不过百，日子过得相当滋润。

后来业务加码了：老板说"我们要支持实时数据分析"，于是加了一个数据分析 Agent。市场部说"用户画像很重要"，又加了一个推荐 Agent。运营说"售后也需要智能化"，于是售后 Agent 也来了。到我反应过来的时候，这台小服务器上已经跑着七个 Agent，内存占用 90%+，启动一个任务能把 CPU 拉满，响应延迟从 200ms 飙到了 5 秒。

单体 Agent 架构的甜蜜期结束了。拆还是死，这是一个问题。

今天我们就聊透：多 Agent 系统怎么从单体拆到分布式，以及拆分后的编排、部署、监控怎么搞。这不是学院派的理论堆砌，而是我从一次次凌晨故障中总结的实战经验。

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

2.1 单体的天花板在哪里？

单体 Agent 架构就是所有 Agent 逻辑跑在同一个进程中：同一个 Python 解释器、共享同一块内存、共用同一条消息总线。它的问题不是功能上的，而是物理上的：

• 内存瓶颈：每个 Agent 的上下文、工具配置、会话状态都吃内存。七八个 Agent 加上各自的长上下文，轻松突破 16G
• CPU 争抢：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_factory=lambda: str(uuid.uuid4())) target_agent: str = "" action: str = "" payload: dict[str, Any] = field(default_factory=dict) parent_task_id: str | None = None # 用于追踪调用链 created_at: float = field(default_factory=asyncio.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 = f"agent: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 = f"agent:inbox:{agent_name}" try: await self._redis.xgroup_create(stream_key, self._consumer_group, mkstream=True) except redis.ResponseError: pass # 消费组已存在 return self._redis.xreadgroup( groupname=self._consumer_group, consumername=consumer_id, streams={stream_key: ">"}, count=1, block=5000, # 阻塞等待，最多 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 = f"agent: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, timeout=30.0) return result except asyncio.TimeoutError: return AgentResult( task_id=task.task_id, agent_name=task.target_agent, success=False, error=f"任务超时: {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_exceptions=True) result_map = {} for task, res in zip(tasks, results): if isinstance(res, Exception): result_map[task.task_id] = AgentResult( task_id=task.task_id, agent_name=task.target_agent, success=False, error=str(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 = f"agent: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[b"data"]) task = AgentTask( task_id=task_data["task_id"], target_agent=task_data["agent_name"], action=task_data["action"], payload=task_data.get("payload", {}), parent_task_id=task_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 的内存膨胀问题。