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( agentagent, toolsself.tools, verboseFalse, handle_parsing_errorsTrue, max_iterations6, )这里有两个关键参数我必须说一下。第一个是max_iterations。我默认设成 6这是多轮试错之后得到的值。设太小Agent 在复杂任务里容易中途放弃设太大一旦模型陷入死循环token 会哗哗地烧。实际运行中超过 6 次工具调用还没有结果的场景极少设置上限可以兜底。第二个是handle_parsing_errorsTrue。模型在 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_exceptionsTrue) 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 上开启了verboseTrue把中间推理过程完整打印到日志里——开发环境才开生产环境关掉否则日志量太大了。最重要的兜底措施仍然是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_idrequest_id, inputuser_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 架构转型建议先小范围试点用数据说话评估清楚收益和成本之后再做决策。