1. 从单 Agent 到 Coordinator-Subagent 架构：为什么要拆分
先说个背景。我去年做了一个面向企业内部知识的问答 Agent，最初形态就是经典的单 Agent：一个大模型实例，挂一堆工具，用户问什么我就把检索、计算、查询这些能力全塞给同一个系统。前期效果还行，但等知识库扩容到几十个数据源、工具数量突破十五个之后，问题开始集中爆发。
最典型的失控场景是这样的：用户问一句“帮我整理一下上季度各区域的销售数据，顺便对比一下预算完成率”，单 Agent 会把这个任务拆成好几个步骤，然后按顺序调用工具。表面上看没啥问题，实际跑到一半就乱套了——工具返回的 JSON 被截断、中间结果把上下文窗口塞爆、模型在多个工具调用之间切换时忘了最初的目标。到后来甚至出现更离谱的情况：Agent 在调用计算器工具时，把上一个工具的返回值当成指令来执行。
我当时的判断是：这不是模型能力的问题，而是架构形态的问题。单 Agent 的本质是“一个大脑干所有事”，它要同时承担任务理解、工具调度、中间结果维护、最终答案生成多个职责。当工具数量和任务复杂度上去了，这个大脑的注意力就会被稀释。类比一下，一个人既当前台接待、又当项目经理、又当财务核算员，还要负责最终汇报，做砸是大概率事件，跟这个人聪不聪明没关系。
后来我花了大概三周时间，把系统重构为 Coordinator-Subagent 架构。整体思路是：让一个 Coordinator（协调者）Agent 负责理解用户意图、拆解任务、分发给不同的 Subagent（子代理），每个 Subagent 只负责一个领域，拥有独立的上下文和专用的工具集。效果很直接：任务成功率从重构前的 61% 提升到了 89%，token 消耗反而下降了将近三分之一。这篇博文就把设计过程、实现细节和踩过的坑完整记录下来。
2. 架构设计前想清楚的三件事
2.1 不是所有场景都适合上多智能体
动手重构之前，我反复问自己一个问题：是单 Agent 真的不行，还是我没用对？这个问题必须想清楚，因为多智能体架构是有成本的——系统复杂度、调试难度、token 开销、延迟都会增加。
我总结出一个判断标准：如果任务可以拆成“理解-检索-生成”三段式，而且工具之间没有强依赖，那单 Agent 足够用了，不需要折腾。但如果任务呈现以下特征，说明单 Agent 的瓶颈已经逼近：
- 工具数量超过 10 个，模型在工具选择上频繁失误；
- 单个任务需要连续调用 5 次以上工具，中间结果容易丢失；
- 不同工具的返回格式差异大，比如一个返回 Markdown 表格、一个返回嵌套 JSON、一个返回图片链接，模型在格式切换间容易出错；
- 存在多个相对独立的子任务，可以并行处理。
我的项目四条全占。尤其是第一条，工具数量超过 10 个之后，模型在 Function Calling 里的工具选择准确率明显下降。后来看了一些资料，说工具数量与准确率之间大致存在一个倒数关系，工具越多，模型选错的概率越高。把工具按领域拆分给不同的 Subagent，本质上就是把“在 20 个工具里选 1 个”的问题，降维成“在 5 个工具里选 1 个”的问题，准确率自然就上来了。
2.2 Coordinator 不是用来写业务逻辑的
我在设计 Coordinator 的时候踩过一个大坑：一开始我把 Coordinator 当成一个“超级大脑”来设计，希望它能像项目经理一样指导每个 Subagent 怎么做业务。结果系统变得极其难调试，Coordinator 输出的执行计划稍微有一点偏差，下游 Subagent 就跟着跑偏。
正确的思路是：Coordinator 只做任务分发和结果汇总，不做业务决策。它就像一个路由器，职责是判断“这个请求属于哪个域”、“要不要拆成多个子任务”、“子任务之间的依赖关系是什么”，然后把任务原样扔给对应的 Subagent。业务判断全部下沉到 Subagent 层。
这里有一个很微妙的设计点：Coordinator 对 Subagent 的“能力边界”要有明确的认知，但不能知道 Subagent 的“内部实现”。我只需要告诉 Coordinator：“你有 3 个子代理可用，分别是销售分析代理、知识库检索代理、报表生成代理，各自负责某某领域”，然后让模型根据用户请求自动路由。不要试图在 Coordinator 层面写死任务执行的具体步骤，否则就退化成了一个硬编码的 if-else 流程，失去了 Agent 的灵活性。
2.3 上下文隔离是多智能体架构的立身之本
单 Agent 崩溃的根源就是上下文互相污染——工具 A 的输出可能影响工具 B 的调用判断，而且这种影响是不可控的。Coordinator-Subagent 架构的最大优势不是“多了几个 Agent”，而是“每个 Agent 拥有独立的上下文空间”。
我采取的设计是：在内存中为每个 Subagent 维护独立的对话历史。Coordinator 只保留任务分发记录和各个 Subagent 返回的最终结果摘要，不保留子任务的中间过程。每个 Subagent 也只保留自己的对话上下文，看不到其他 Subagent 的原始结果。
这样设计的好处是显而易见的：首先是 token 消耗大幅下降。单 Agent 模式下，每轮工具调用后，历史中都会叠加一堆 JSON 输出，上下文越滚越大，最后光 Prompt 就占了上下文窗口的大半。拆成 Subagent 之后，每个 Subagent 的上下文窗口都被控制在较小的范围内，消耗自然降下来了。
其次是隔离了错误。一个 Subagent 跑偏了，不会影响其他 Subagent 的正常工作。这就像分布式系统里的故障隔离域，单个节点挂掉不至于拖垮整个集群。
3. Coordinator-Subagent 的代码骨架与落地实现
3.1 整体流程设计与状态机
我的实现基于 LangGraph 构建了状态机。选择 LangGraph 而不是 LangChain 的 AgentExecutor，原因是 LangGraph 支持显式的状态管理和节点间的条件跳转，这对多智能体编排来说是刚需。
状态机的核心节点分为四类：
- 入口节点：接收用户输入，调用 Coordinator 模型判断任务类型；
- 路由节点：根据 Coordinator 的输出决定去往哪个 Subagent；
- 执行节点：具体的 Subagent 调用工具、生成结果；
- 聚合节点：收集所有 Subagent 的结果，交给汇总模型生成最终回复。
这里有一个容易被忽视的细节：需要显式区分“路由到子代理”和“直接返回用户”两种情况。很多简单的编排框架里，Coordinator 一旦不能确定任务归属，就会随机选一个 Subagent 执行，导致答非所问。我在路由节点里加了一个判断：如果 Coordinator 给出的任务类型置信度低于阈值，直接走“澄清问题”分支，让用户补充信息，而不是强行分配。
from typing import TypedDict, Literal, List from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class AgentState(TypedDict): user_input: str route: str sub_outputs: dict final_answer: str class RouteDecision(BaseModel): """Coordinator 的路由决策结果""" route: Literal["sales_agent", "kb_agent", "report_agent", "clarify"] = Field( description="路由目标：销售分析/知识库/报表生成/澄清" ) confidence: float = Field(description="置信度 0-1") need_all: bool = Field( description="如果为 True，则多个子代理并行执行后聚合结果" )Coordinator 的 Prompt 模板核心要义就一句话：让模型从预定义的路由集合里选边，不要让它自由发挥。我见过很多失败的案例，就是给 Coordinator 的指令太开放，结果模型输出的路由目标五花八门，跟实际注册的 Subagent 对不上，还得在代码里做模糊匹配，纯属自己给自己找麻烦。
3.2 Subagent 的定义与工具隔离
每个 Subagent 本质上是一个独立的 AgentExecutor，有自己的系统提示词和工具集。我用 dataclass 封装了一个统一的 Subagent 接口，方便路由节点动态调用。
from dataclasses import dataclass, field from typing import List, Callable from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate @dataclass class SubAgent: name: str description: str system_prompt: str tools: List[Callable] llm: object executor: AgentExecutor = None def __post_init__(self): prompt = ChatPromptTemplate.from_messages([ ("system", self.system_prompt), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_tool_calling_agent(self.llm, self.tools, prompt) self.executor = AgentExecutor( agent=agent, tools=self.tools, verbose=False, handle_parsing_errors=True, max_iterations=6, )这里有两个关键参数我必须说一下。
第一个是max_iterations。我默认设成 6，这是多轮试错之后得到的值。设太小，Agent 在复杂任务里容易中途放弃；设太大，一旦模型陷入死循环，token 会哗哗地烧。实际运行中，超过 6 次工具调用还没有结果的场景极少，设置上限可以兜底。
第二个是handle_parsing_errors=True。模型在 Function Calling 模式下偶尔会输出格式错误的 JSON，开启这个开关后，框架会把错误信息回传给模型让它自行修正。这个特性在单 Agent 时代救过我很多次，多 Agent 时代同样适用。
工具隔离方面，我把原来的十几个工具按领域重新归拢。销售分析 Agent 持有查询销售数据的工具，知识库 Agent 持有向量检索和文档读取工具，报表 Agent 持有图表生成和格式化工具。工具之间完全不共享，每个 Subagent 只能看到自己的那套工具，从机制上杜绝了工具误调用的可能。
3.3 并行分发与结果聚合的逻辑
并行编排是 Coordinator-Subagent 架构相比单 Agent 的又一个显著优势。单 Agent 处理“对比 A 区域和 B 区域的销售数据”这类任务时，只能串行地先查 A 再查 B；而多智能体架构下，完全可以让两个 Subagent 同时去查，最后统一汇总。
我在状态机的聚合节点里做了两件事。第一是收集所有已执行 Subagent 的结果，打包成一个结构化字典；第二是把这些结果塞进一个汇总模型的 Prompt，让汇总模型整合成连贯的自然语言答案。
并行执行我用了 Python 的 asyncio 和asyncio.gather来控制。需要注意一个细节：多个 Subagent 共享同一个 LLM 客户端时，要考虑 API 的并发限制，尤其是使用 OpenAI 兼容接口时，超频会导致 429 错误。我在代码里给每个 Subagent 的 LLM 实例配置了独立的请求池和重试机制，避免一个 Agent 的重试风暴影响其他 Agent。
import asyncio from langchain_core.runnables import Runnable async def run_parallel_subagents( routes: List[str], subagent_map: dict, state: AgentState ) -> dict: tasks = {} for route in routes: agent = subagent_map[route] tasks[route] = agent.executor.ainvoke({"input": state["user_input"]}) results = await asyncio.gather(*tasks.values(), return_exceptions=True) return {route: results[i] for i, route in enumerate(tasks.keys())}聚合节点要注意的坑是：子 Agent 返回的结果可能是字符串、也可能是结构化 JSON，在汇总之前最好统一格式。我在每个 Subagent 的 Prompt 里都指定了“最终输出格式”，要求返回{"summary": "...", "data": {...}}结构，这样聚合阶段就不用再处理格式兼容问题。
4. 踩坑实录：六个最值得记录的教训
4.1 协调者幻觉：把不存在的 Subagent 当成目标
第一个坑来得很早。重构上线第一天，有个用户问“帮我查一下最近的工单状态”，Coordinator 竟然把任务路由到了一个叫 “ticket_agent” 的子代理——问题是，我压根没建过这个子代理。
排查后发现，Coordinator 的 Prompt 里描述的是“你有以下子代理可用”，然后列了三个名字。但模型见到用户的问题里有“工单”这个词，就自作主张“脑补”出了一个不存在的工单代理，然后返回了一个无效路由。代码里没做异常处理，路由节点拿到一个不存在的键，直接抛 KeyError，整个请求就挂了。
教训有两条。第一，路由节点必须做有效性校验，任何不在白名单里的路由目标都走兜底分支；第二，Coordinator 的 Prompt 里要显式声明“只能从给定列表中选择路由目标，不要创造新的代理名称”。后来我把路由决策改成了结构化输出模式（也就是前面代码里的RouteDecision），用 Pydantic 强制模型只能输出合法的枚举值，这个坑才算彻底堵住。
4.2 Subagent 之间的“抢答”问题
第二个坑出在结果聚合阶段。当时我设计了三个 Subagent：销售分析、知识库检索、报表生成。本来设想的是：用户问“分析一下上月销售数据并生成报表”，Coordinator 会同时触发销售分析和报表生成两个子代理。结果实际跑起来，Coordinator 偶尔会只路由到其中一个 Subagent，导致回答要么只有分析没有报表，要么只有报表没有分析。
这个问题的根源是 Coordinator 的任务拆解粒度不够细。它能在“这个任务属于哪个领域”上做对判断，但在“这个任务需要哪些子代理协作”上经常漏项。
我的解决方案是在路由决策结构里增加一个need_all字段，当 Coordinator 判断任务涉及多个子代理时，就触发并行执行流程。另外在 Prompt 里给了几个典型的组合示例，比如“用户要求分析+出图”就应该同时路由到销售分析和报表生成。Few-shot 示例的效果比单纯描述规则好得多。
4.3 上下文被摘要算法“洗”没了
第三个坑比较隐蔽，是关于上下文摘要的。我当时为了让每个 Subagent 的上下文不膨胀，在每次子代理返回结果后，都用 LLM 将其结果“压缩”成一段摘要再传给 Coordinator。后来发现，某些场景下 Coordinator 会丢失关键信息，导致最终汇总结果残缺。
深挖原因后发现，问题不是摘要本身的问题，而是摘要的时机不对。有些子任务的结果必须保留原始数据，比如金额、日期、同比环比数字这些，一旦被“压缩”成自然语言摘要，数字精度就没了。模型在摘要时会把“销售额 12345678.90 元”概括成“销售额有所增长”，到汇总阶段就彻底丢掉了精确值。
后来的做法是分级处理：对于需要精确数字的子代理结果，保留原始 JSON 结构，只对不需要精确值的中间过程做摘要；对于文本类的检索结果，才用摘要压缩。一句话总结：别对所有结果一刀切地做摘要处理。
4.4 死循环：Agent 在工具调用之间反复横跳
这个问题我在单 Agent 时代就遇到过，但多智能体架构下它表现得更具迷惑性。某个 Subagent 为了回答一个简单问题，反复调用同一个工具 5 次也不罢休。表面上看，函数调用的入参每次都不一样，但模型的调用意图没有任何新信息。
排查这类问题比较费劲，因为 Agent 的中间推理过程如果没有完整日志，根本看不出来它“为什么”反复横跳。后来我做了两件事：
第一是给每个 Subagent 的 Prompt 加了一条约束：同一个工具调用最多允许出现两次，若例外情况必须明确说明理由。第二是在 AgentExecutor 上开启了verbose=True，把中间推理过程完整打印到日志里——开发环境才开，生产环境关掉，否则日志量太大了。
最重要的兜底措施仍然是max_iterations参数。我把单次请求的最大工具迭代次数从 6 调到了 8，但同时在聚合层加了一个全局超时控制：整个请求最多允许执行 60 秒，超时就主动中断并返回已有结果。Deadline 机制在分布式系统里是标配，多智能体编排同样需要。
4.5 Token 消耗不降反升？问题出在日志和重试上
前文提到重构后 token 消耗下降了约三分之一，但这个过程不是一开始就如此。重构初期，我惊讶地发现 token 消耗不降反升，一度怀疑拆分架构是否有价值。
排查下来的原因有三点：
第一，LangGraph 的调试日志默认会把每轮 Agent 的完整输入输出写入存储，这在开发环境是很有用的诊断工具，但生产环境会产生海量 token 级别的日志开销。我在生产配置里关闭了 verbose 日志，只保留路由决策和最终结果摘要。这里要注意的是，LangGraph 的checkpointer默认会保存每次节点间的完整状态，也包含了中间 Prompt 内容。
第二，重试机制写得不合理。当时所有 Subagent 共享一组重试参数，一旦某个上游 API 出现限流，所有 Agent 的重试风暴会同时发生，token 消耗翻倍。改成给每个 Subagent 独立配置重试次数和重试间隔后，这个现象才消失。
第三，Coordinator 每轮都会携带完整的用户输入和任务下发记录。有些用户消息特别长，Coordinator 反复处理这堆长文本，token 自然水涨船高。优化方法是：Coordinator 只保留脱敏后的用户意图摘要，而不是原文。这种方法会带来一点精度损失，但整体收益远大于损失。
4.6 生产环境下的降级策略
最后是个工程层面的坑。多智能体架构本身引入了新的故障模式，其中最常见的就是“某个 Subagent 对应的大模型 API 不可用”。单 Agent 时代，这种情况就是请求失败；多 Agent 时代，如果销售分析 Agent 挂了，用户问销售问题就直接得到报错，显然不可接受。
我加了一个降级逻辑：当某个 Subagent 连续失败超过两次时，自动把它从可用路由列表中移除，Coordinator 只能把任务分配给剩下的 Subagent。对于被移除的 Subagent 对应的任务类型，系统会给用户返回一个“当前该模块暂时不可用，请稍后再试”的提示，而不是抛出一个毫无解释的异常。
同理，如果所有 Subagent 都不可用，整个系统就退化成一个普通的 LLM 问答接口，不调用任何工具。这种从复杂架构到简单逻辑的降级路线，值得一开始就设计好，否则线上事故发生时你就会手忙脚乱。
5. 问题排查与性能调优的实战技巧
5.1 可观测性建设：日志里要有“任务链路 ID”
多智能体系统的调试难度远大于单智能体。我调试一个问题时，经常需要在日志里翻出 Coordinator 的路由决策、Subagent 的推理过程、工具调用的入参和出参、聚合阶段的 Prompt 拼装结果。如果没有一个贯穿全链路的标志，这些日志根本串不起来。
我后来给每个请求生成了一个request_id，作为日志的唯一追踪标识。在 Coordinator 收到请求、路由分发、每个 Subagent 执行前后、聚合输出等关键节点都打印带有request_id的日志。配合 ELK 之类日志系统，能很快定位请求在哪个环节出了问题。
这个习惯越早养成越好。等系统规模变大、Agent 数量变多之后再补，成本会高很多。
import uuid import structlog logger = structlog.get_logger() def process_request(user_input: str): request_id = str(uuid.uuid4()) logger.info("request_start", request_id=request_id, input=user_input) # 后续每个环节都带上 request_id5.2 用 Eval 集度量架构改造成果
架构改造不能凭感觉说“变好了”，要有数据支撑。我为这个项目搭了一套简单的 Eval 集：整理了 80 条覆盖不同难度和任务类型的测试问题，并标注了预期行为（比如“应该路由到销售分析 Agent”“应该触发并行执行”“应该包含精确数字”）。
每次调整 Prompt 或架构后，我就在这 80 条问题上跑一遍全量回归，记录任务成功率、token 消耗、平均响应延迟三个指标。对比之下，架构改造成果一目了然：成功率从 61% 提升到 89%，平均 token 消耗下降 32%，平均响应时间从 4.8 秒降到了 3.1 秒——并行执行带来的收益在这里体现得很明显。
建立 Eval 集有一些技巧：不要只放“正常问题”，要特意加入边界情况，比如语气模糊的问题、超长文本问题、包含多个子任务的复合问题，这些才是真正考验架构设计的地方。
5.3 性能调优：缓存 + 并发 + 模型分级
最后说三个性能调优点。
第一个是语义缓存。如果两个用户问了同一个问题，且历史中有相同问题的输出，直接返回缓存结果，不用重新走完整流程。我用的是一个简单的向量相似度匹配，达到阈值就命中缓存。这个优化对生产环境很有效，能砍掉大量重复请求。
第二个是模型分级。不是所有 Subagent 都需要用最强的模型。知识库检索 Agent 的任务相对简单，我可以给它配一个较快且便宜的模型；销售分析 Agent 涉及较多计算和推理，才用较强的模型。这样既能控制成本，又不牺牲核心链路的回答质量。Coordinator 因为是任务分发，对推理能力的要求也不是最高的，中等模型就足够。
第三个是并发上限控制。多 Agent 并行看起来爽，但必须设置一个全局并发上限，防止几百个请求同时触发几十个 Agent，把后端 API 打爆。我用的是一个简单的信号量机制，控制同时在执行 Agent 数量不超过 20 个。
注意：所有调优手段都要结合业务实际。如果你的系统是低频请求场景，上面的优化可能不是重点；但如果你的系统每天有成千上万的请求，缓存和并发控制就是基本盘，不做迟早出事。
6. 后续扩展方向
这次架构改造积累了不少经验，也让我看到几个值得继续深化的方向。
第一个方向是让 Coordinator 具备“动态规划”能力，而不是仅仅在预设的路由集合里做选择。比如用户提出了一个当前所有 Subagent 都无法覆盖的任务时，Coordinator 可以创建临时的 Subagent，把可用工具动态组装给它。这会引入类“Agent 工厂”的机制，但随之而来的是工具权限、模型配置、安全边界等一系列新问题。
第二个方向是记忆共享机制。目前的架构中，每个 Subagent 的上下文是隔离的，这对于单次请求没问题，但对跨会话的持续性任务就不好办了。比如用户第一天让销售分析 Agent 生成了报告，第二天想让报表 Agent 基于这份报告做出图表，两个 Agent 之间没有共享记忆，就得让用户重新提供一遍信息。设计一个安全、可控的跨 Agent 记忆层，会是一个值得投入的方向。
第三个方向是多模态场景的延伸。目前的 Subagent 处理的主要是文本，但在实际业务里，用户可能会上传图片要求解读，或者要求生成图表。如果不同的 Subagent 能支持不同的模态能力，Coordinator 的路由判断就会涉及模态匹配，这会比单纯的文本路由更有挑战性，也更有价值。
根据我个人经验，多智能体编排还处于快速演进阶段，这个领域的“最佳实践”每几个月就会更新。架构设计上最重要的不是追求完美，而是保留足够的迭代空间——组件之间解耦、接口定义清晰、依赖显式化，这样无论未来模型能力怎么变，框架都能跟着演进而不需要推倒重来。如果你也正在从单 Agent 往多 Agent 架构转型，建议先小范围试点，用数据说话，评估清楚收益和成本之后再做决策。
