Agent 系统全年架构演进总结从单体 Agent 到多 Agent 平台的路线图一、深度引言与场景痛点大家好我是赵咕咕。一年前我们的 Agent 系统长这样agent ReActAgent(llmGPT4, tools[search, db, email]) agent.run(帮我做周报)一个 Agent 实例三个工具一个循环。简单能跑但很快就暴露了所有单体架构的通病一个慢工具比如查数据库超时卡死整个 Agent。增加新工具要改 Agent 核心代码。多个用户并发请求时Agent 之间互相抢资源。没有监控不知道哪个步骤最慢、哪个工具最容易出错。一年后的今天这套系统已经演化成了一个多 Agent 平台5 个专职 Agent、共享工具池、事件驱动流水线、幻觉检测层、Prompt 脱敏中间件、两级 embedding 缓存。这篇文章是我对这一年架构演进的总结。如果你正在从 0 搭建 Agent 系统希望这篇路线图能帮你少走一些弯路。二、底层机制与原理深度剖析2.1 一年演进的完整路线图回顾这一年的演进历程我们的架构主要经历了四个关键里程碑的迭代。每个阶段的升级都是由前一个版本暴露出的核心痛点所驱动具体路径如下里程碑 1单体 Agent初期采用标准的 ReAct Agent配备 3 个工具运行在单进程中。里程碑 2拆解为专用 Agent由于单体架构存在单点故障且无法扩展我们将系统拆解为检索专用、任务分解和工具执行三个专职 Agent。里程碑 3基础设施层专用 Agent 之间出现了碎片化、重复建设、耦合及异常传播等问题同时面临幻觉和质量不稳定、安全 PII 泄露等挑战。为此我们引入了共享工具池 MCP、事件驱动流水线、幻觉检测层以及 Prompt 脱敏中间件。里程碑 4生产级平台基于近一年的成果总结与展望最终构建了具备多租户隔离、可观测性、成本控制及自我优化能力的生产级平台。接下来我们将详细拆解第一个里程碑的具体实践与教训。2.2 里程碑 1单体 Agent这是我们最开始的版本——一个标准的 ReAct Agent。技术栈LangChain GPT-4 3 个自定义工具。做对了什么快速起步。从 0 到 demo 只用了 3 天验证了 Agent 这个方向在业务上确实能解决用户问题。做错了什么没有考虑扩展性。ReAct 循环没有 max_steps 限制有一次 Agent 跑了 47 步Token 账单 13 美元。工具调用没有超时控制一个卡住的数据库查询让整个 Agent 挂掉。所有逻辑堆在一个 800 行的文件里。最大的教训Demo 能跑不等于生产能跑。单体 Agent 的最长生命周期是 2 周——超过这个时间还没开始拆分架构债就会开始滚雪球。2.3 里程碑 2拆解为专用 Agent当单体 Agent 需要处理的工具超过 5 个、场景超过 3 种时我们发现一个 Agent 处理所有事的模式行不通了。于是做了一次大重构按职责拆成 3 个专用 Agent。Retriever Agent只做一件事——根据用户问题检索相关文档。内置了向量检索 BM25 混合检索 重排序。对外暴露search(query) - docs接口。Planner Agent拿到用户任务后分解为子任务分配给合适的 Executor。类似 Plan-Execute 模式中的 Planner 角色。Executor Agent×2执行具体的工具调用。一个负责内部数据查询数据库 API一个负责外部操作发邮件、创建 Issue、更新 CR 系统。关键收益每个 Agent 的代码量从 800 行降到 200 行。Retriever Agent 可以独立升级比如换 embedding 模型不影响其他 Agent。并发场景下Executor Agent 可以横向扩展多实例Planner 做负载均衡。新问题每个 Agent 各自维护了自己的工具实现查数据库这个功能在 Retriever 和 Executor 里各写了一遍。Agent 之间是直接函数调用耦合严重。Retriever 挂了会导致整个链路不可用。没有统一定义工具的接口和生命周期。2.4 里程碑 3构建基础设施层这个阶段我们做了四件事把各自为政的 Agent 集群变成有共同基础设施的平台。第 3 周完成的关键工作就是这周共享工具池MCP 协议用 MCPModel Context Protocol标准化了工具接口。所有工具以 MCP Server 形式提供Agent 通过 MCP Client 调用。新增一个工具只需要多起一个 MCP Server不需要改任何 Agent 代码。事件驱动流水线asyncio.Queue用 asyncio.Queue 把 Agent 之间的调用从直接函数调用改成消息传递。每个 Agent 是独立协程通过队列通信。一个 Agent 崩溃不会导致整个链路不可用。幻觉检测层在 Agent 输出答案之前加了一层交叉验证。NLI 模型 BM25 关键词共现验证答案中的每条声明是否被检索文档支撑。幻觉率从 12% 降到了 4%。Prompt 脱敏中间件在 Prompt 发给 LLM 之前正则 高熵检测自动识别并替换 PII 和密钥。LLM 只能看到占位符原始值在返回给用户前还原。这个阶段的教训基础设施不是一次建成的。MCP 协议是先用 stdio Transport 跑通再逐步补上 HTTPSSE 支持。幻觉检测器是先上线规则引擎正则 BM25再逐步接入 NLI 模型。基础设施的引入必须对上层透明。Agent 代码不需要知道我正在通过 MCP 调工具或我的 Prompt 被脱敏了。这些是中间件不是业务逻辑。2.5 里程碑 4生产级平台进行中第四季度要推进的方向多租户隔离不同客户的 Agent 实例在资源Token 配额、工具权限、数据隔离上完全独立。可观测性统一的 tracing每个 Agent 调用链路可追踪、metrics延迟、命中率、幻觉率、logging结构化日志。成本控制基于模型路由的智能降级——简单问题用gpt-4o-mini复杂问题用gpt-4o。Token 消耗实时监控和预算告警。自我优化Reflexion 机制让 Agent 从失败中学习。缓存的 feedback 累积后自动触发 Prompt 优化。三、生产级代码实现这里给出当前多 Agent 平台的骨架代码——各个模块的接口和编排方式import asyncio import logging from dataclasses import dataclass, field from typing import Any logger logging.getLogger(__name__) # ─── 基础消息体 ─── dataclass class AgentMessage: Agent 之间传递的消息。 task_id: str role: str # planner | retriever | executor content: dict[str, Any] error: str | None None metadata: dict[str, Any] field(default_factorydict) # ─── Agent 基类 ─── class BaseAgent: 所有 Agent 的基类。 def __init__(self, agent_id: str): self.agent_id agent_id self._running False async def start(self): self._running True logger.info(Agent %s 启动, self.agent_id) async def stop(self): self._running False logger.info(Agent %s 停止, self.agent_id) async def handle(self, msg: AgentMessage) - AgentMessage: raise NotImplementedError # ─── 平台编排器 ─── class AgentPlatform: 多 Agent 平台的编排核心。 def __init__(self): # 解耦队列 self._planner_queue: asyncio.Queue[AgentMessage] asyncio.Queue(maxsize100) self._retriever_queue: asyncio.Queue[AgentMessage] asyncio.Queue(maxsize100) self._executor_queue: asyncio.Queue[AgentMessage] asyncio.Queue(maxsize100) self._output_queue: asyncio.Queue[AgentMessage] asyncio.Queue(maxsize200) # Agent 实例 self._planner: BaseAgent | None None self._retriever: BaseAgent | None None self._executors: list[BaseAgent] [] # 基础设施中间件 self._sanitizer None # PromptSanitizer self._hallucination_detector None # HallucinationDetector self._cache None # EmbeddingCache # 结果追踪 self._pending_tasks: dict[str, asyncio.Future] {} async def startup(self): 启动平台Agent 基础设施。 self._planner PlannerAgent(planner-01) self._retriever RetrieverAgent(retriever-01) self._executors [ ExecutorAgent(executor-01, data_query), ExecutorAgent(executor-02, external_action), ] agents [self._planner, self._retriever] self._executors await asyncio.gather(*[a.start() for a in agents]) # 启动 Worker 协程 self._workers [ asyncio.create_task(self._run_worker(self._planner_queue, self._planner)), asyncio.create_task(self._run_worker(self._retriever_queue, self._retriever)), ] for executor in self._executors: self._workers.append( asyncio.create_task(self._run_worker(self._executor_queue, executor)) ) logger.info(Agent 平台启动完成: %d Agents, %d Workers, len(agents), len(self._workers)) async def execute(self, task: str, user_id: str anonymous, timeout: float 60.0) - dict[str, Any]: 执行一个用户任务。 task_id ftask_{user_id}_{asyncio.get_running_loop().time()} # 1) Prompt 脱敏 if self._sanitizer: task, _ await self._sanitizer.sanitize(task) # 2) 分发到 Planner msg AgentMessage( task_idtask_id, roleuser, content{task: task, user_id: user_id}, ) await self._planner_queue.put(msg) # 3) 等待结果 future: asyncio.Future asyncio.get_running_loop().create_future() self._pending_tasks[task_id] future try: result_msg await asyncio.wait_for(future, timeouttimeout) except asyncio.TimeoutError: self._pending_tasks.pop(task_id, None) return {status: error, message: 任务处理超时} finally: self._pending_tasks.pop(task_id, None) # 4) 幻觉检测 if self._hallucination_detector and result_msg.content.get(answer): retrieved_docs result_msg.metadata.get(retrieved_docs, []) report await self._hallucination_detector.validate( result_msg.content[answer], retrieved_docs ) result_msg.metadata[hallucination_report] { is_safe: report.is_safe, ratio: report.hallucination_ratio, } return { status: success, task_id: task_id, answer: result_msg.content.get(answer, ), metadata: result_msg.metadata, } async def _run_worker( self, queue: asyncio.Queue, agent: BaseAgent ) - None: Worker 主循环从队列取消息 → Agent 处理 → 路由到下游队列。 while self._planner is not None and self._planner._running: msg await queue.get() if msg is None: # 停止信号 break try: result await agent.handle(msg) await self._route_result(result) except Exception as e: logger.error(Agent %s 处理失败: %s, agent.agent_id, e) # 错误消息通知等待者 future self._pending_tasks.get(msg.task_id) if future and not future.done(): future.set_exception(e) finally: queue.task_done() async def _route_result(self, msg: AgentMessage) - None: 根据消息的角色路由到下游队列。 routing_map { planner: self._retriever_queue, retriever: self._planner_queue, executor: self._planner_queue, } target routing_map.get(msg.role) if target is not None: await target.put(msg) # 如果是最终产出通知等待者 if msg.role planner and msg.content.get(final, False): future self._pending_tasks.get(msg.task_id) if future and not future.done(): future.set_result(msg) async def shutdown(self): 优雅关闭平台。 for worker in self._workers: worker.cancel() for agent in [self._planner, self._retriever] self._executors: if agent: await agent.stop() queues [ self._planner_queue, self._retriever_queue, self._executor_queue, self._output_queue, ] for q in queues: await q.put(None) await asyncio.gather(*self._workers, return_exceptionsTrue) logger.info(Agent 平台已关闭) # ─── Agent 实现示例骨架 ─── class PlannerAgent(BaseAgent): async def handle(self, msg: AgentMessage) - AgentMessage: if msg.role user: # 用户新任务 → 需要检索 return AgentMessage( task_idmsg.task_id, roleplanner, content{action: retrieve, query: msg.content[task]}, ) elif msg.role retriever: # 检索完成 → 生成答案 docs msg.content.get(docs, []) return AgentMessage( task_idmsg.task_id, roleplanner, content{ final: True, answer: f根据 {len(docs)} 份文档的分析结果..., sources: [d.get(id, ) for d in docs[:3]], }, metadata{retrieved_docs: docs}, ) return msg class RetrieverAgent(BaseAgent): async def handle(self, msg: AgentMessage) - AgentMessage: query msg.content.get(query, ) # 查 embedding 缓存 # 做混合检索向量 BM25 # 做重排序 docs [{id: fdoc_{i}, content: f相关文档 {i}} for i in range(5)] return AgentMessage( task_idmsg.task_id, roleretriever, content{docs: docs, query: query}, ) class ExecutorAgent(BaseAgent): def __init__(self, agent_id: str, domain: str): super().__init__(agent_id) self.domain domain async def handle(self, msg: AgentMessage) - AgentMessage: return AgentMessage( task_idmsg.task_id, roleexecutor, content{result: f{self.domain} 操作完成}, )架构设计要旨AgentPlatform是唯一的编排入口。它持有所有队列、Agent 实例和中间件。把Agent 通信和Agent 逻辑彻底分离。_route_result做消息路由。Planner 发回来的消息根据内容决定发给 Retriever 还是 Executor。这是一个简单的规则路由未来可以升级为动态路由基于消息内容和 Agent 负载。每个 Agent 的handle是纯逻辑——接收消息处理后返回新消息。Agent 不需要知道消息从哪来、到哪去。四、边界分析与架构权衡4.1 什么时候应该拆 Agent不是每个项目都需要多 Agent。拆分的判断标准单个 Agent 的工具超过 8 个→ 拆。工具太多会导致 LLM 选择工具的正确率下降。有一个明确的独立子任务→ 拆。比如检索和执行 SQL 查询是两种完全不同的技能天然适合独立 Agent。需要独立扩缩容→ 拆。如果一个子任务的 QPS 是其他子任务的 10 倍拆开后可以水平扩展该子任务对应的 Agent。不到三个理由别拆。每拆一个 Agent 就多一个队列和一组生命周期管理代码。4.2 多 Agent 通信的开销消息在队列间传递的延迟在 0.1-1ms。一个请求经过 Planner → Retriever → Planner → Executor → Planner5 跳的额外延迟约 2-5ms。对于端到端延迟在 1-3 秒的 Agent 系统来说这个开销几乎可以忽略。真正的开销是上下文传递。Planner 和 Executor 之间需要传递当前任务的上下文——检索结果、执行计划、历史对话等。如果上下文很大几万 Token消息体的序列化开销就不可忽视了。优化方向Agent 之间传递引用如task_id而不是全量上下文通过共享的TaskStore来获取上下文。4.3 单体 vs 多 Agent 的选择矩阵维度单体 Agent多 Agent 平台实现成本低1-2 周高1-2 月工具上限约 8 个无上限异常隔离差好独立扩展不支持支持调试难度低高Token 消耗低略高Agent 间通信适合场景原型、内部工具生产服务、多租户4.4 未来演进Agent-to-Agent Protocol目前 Agent 之间的通信还是定制的消息体 规则路由。未来有两个方向值得关注A2A Protocol类比 MCP 定义了 Agent 到工具的协议A2AAgent-to-Agent可能会定义 Agent 之间通信的标准格式。Agent Registry 动态发现Agent 启动后注册到 RegistryPlanner 根据能力描述动态选择合适的 Agent 来执行子任务而不是硬编码路由规则。这两个方向都还在社区早期探索中但在 2025 年下半年可能会有成熟方案出来。五、总结回顾这一年Agent 架构的演进可以归纳为一条主线和三条支线。主线单体 → 拆分 → 平台化。从能跑到能扛从快糙猛到稳准狠。每个阶段的驱动力都是当前的架构已经不能满足业务需求了。如果你还没到那个阶段别急着拆——过早优化是万恶之源。三条支线工具标准化MCP 协议把工具从Agent 的私有财产变成平台的共享资源。这是一次从 0 到 1 的范式转变。流程解耦事件驱动架构asyncio.Queue让 Agent 之间的耦合从函数调用变成消息传递。异常隔离、独立扩展才成为可能。质量守门幻觉检测和 Prompt 脱敏是两道质量门禁。Agent 可以犯错但错误不能直接到达用户。多 Agent 平台不是终点。下一个里程碑可能是 Agent 的自我优化和自主协作——但这要等上述三条支线的基础设施都稳定下来之后。无论如何从一个 800 行的单体 Agent 到现在 5 个专职 Agent 4 层基础设施的平台这一年是我职业生涯中学到东西最多的一年。希望这篇总结能帮你在 Agent 架构的迷雾中找到一点方向。本系列完结。感谢一个月来的陪伴。有任何问题欢迎评论区交流。