这篇倒不是我做过的最复杂的Agent项目但绝对是我带新团队时最常用来演示“Agent协作到底是怎么跑起来”的一个案例。标题里的“搭建多Agent协作框架实现核心流程”听起来挺大其实落地下来就三件事把任务拆明白、把Agent之间的分工和沟通方式定清楚、再把状态流转用代码固定住。这篇就按我当时从零搭一个多Agent协作框架的全过程来写从思路到踩坑都放进来。1. 为什么必须引入多Agent协作框架而不是写一个大Agent先聊一个最基础的问题单Agent能做的事为什么要拆成多个很多刚接触Agent开发的朋友上来就写一个超大的Prompt把所有任务都塞给一个Agent结果运行时要么上下文爆掉要么角色混乱要么一个分支出错全部重来。我在实际项目里吃过这个亏后来才彻底转向多Agent架构。1.1 单Agent的天花板上下文、专注度与故障隔离一个Agent本质上是一个“大模型Prompt工具集记忆”的组合体。模型本身的上下文窗口是有限的即便现在窗口已经不小但塞进去的内容越杂模型提取关键信息的难度就越大输出质量下降得很快。这有点像让一个人同时干产品经理、后端开发、测试和运维的活不是不可能但效率和质量都会打折。单Agent还有一个隐患是故障隔离能力差。如果Agent在某一步产生幻觉或者工具调用出错错误会沿着同一段上下文继续传播最终输出一个看起来合理但完全不能用结果。多个Agent协作时每个Agent只处理自己那一段出错可以被定位到具体的Agent和步骤修复成本低很多。1.2 多Agent协作的本质把“一个人”变成“一支团队”多Agent协作框架的本质是用多个各有专长的Agent模拟一支团队。团队的协作效率取决于两件事分工是否清晰沟通是否顺畅。分工清晰意味着每个Agent有独立的职责描述、工具集和输出约束沟通顺畅意味着Agent之间传递的消息结构统一、状态流转明确不会出现A输出的字段B看不懂的情况。我第一次搭这个框架时犯过一个典型错误把任务拆给多个Agent之后却没有给它们定义统一的输出格式结果下游Agent拿到上游的结果解析半天都拿不到关键字段。后来我花了整整一天把每个Agent的输出Schema统一成JSON结构所有问题迎刃而解。这件事让我意识到多Agent框架的核心不是“多模型”而是“多角色标准接口”。1.3 什么场景真正需要多Agent什么场景不需要不是所有项目都适合上多Agent。我的判断标准很简单如果你要处理的任务需要多种不同专业能力或者任务的某个环节需要独立验证、独立执行那就值得拆。最典型的例子是内容生产流水线——策划、写作、配图、校对、发布每个环节技能差异大拆成独立Agent非常自然。反过来如果任务本身很简单比如“把这段文本翻译成英文”硬拆成多个Agent只会增加延迟和复杂度性能反而更差。在技术选型上我遵循一个原则先单后多能单不硬多拆多必定义接口。2. 核心流程的设计思路与整体架构拆解这个项目的核心流程定位是“任务拆解→Agent执行→结果汇合”。听起来简单真正落地时涉及的细节远比想象中多。我在这里把当时的思路完整拆开讲。2.1 四种主流协作模式我为什么选了“协同群组”多Agent协作的架构模式大体有四种管线模式、监督者模式、协同群组模式、网状自治模式。管线模式适合流程固定的任务比如“A→B→C”串行处理监督者模式适合任务需要动态决策分配的场景由一个“老板”Agent调度其他Agent协同群组模式是多个Agent通过一个共享的消息总线协作各自认领任务网状自治模式则更自由Agent之间可以直接对话。我最终选了协同群组模式原因是这个项目的核心流程虽然复杂但每个子任务的边界非常清晰不需要一个中心化的调度者来做太多智能决策。共享消息总线的设计让各个Agent之间解耦后续想新增Agent只需要实现接口然后注册不需要改动现有逻辑。项目后期也验证了这个决定的正确性——我们新增了三个Agent零改动接入现有系统。2.2 框架选型为什么没有从零手写编排层当时市面上已经有不少Agent框架轮子比如AutoGen、LangGraph、CrewAI等。我简单拆解了它们的共性和差异之后决定不直接套用而是参考它们的设计思路自己搭一个轻量级编排层。原因有三个第一通用框架为了适配各种场景抽象层比较厚调试和排错时不太直观第二项目的核心流程对消息格式有强约束通用框架未必支持得很好第三团队后续有深度定制需求自研编排层反而可控。但这不意味着从零开始。底层模型调用、工具库、Memory这些基础能力我直接复用了成熟组件。自研的部分主要集中在Agent的注册、消息路由、任务状态机这三块。这个“半自研”的策略让我既拿到了成熟组件的稳定性又保住了业务逻辑的灵活性。2.3 核心流程的四阶段拆解、路由、执行、聚合整个核心流程我拆成了四个阶段。拆解阶段由一个Planner Agent负责把用户提交的原始目标拆成子任务列表并为每个子任务标注需要的Agent类型路由阶段由编排层根据子任务的元信息把任务投递到对应的Agent队列执行阶段每个Agent独立完成自己的子任务把结果按统一Schema写回消息总线聚合阶段一个Reporter Agent负责收集所有子结果合成最终产出。这四个阶段看起来常规但每一步都藏了不少细节。拆解阶段要考虑到子任务的依赖关系路由阶段要处理队列阻塞执行阶段要设置超时与重试聚合阶段要处理部分子任务失败的情况。这些细节我在第3节和第5节会仔细展开。3. 实操环节搭建一个最小可用的多Agent协作框架这一节是全文最核心的部分。我把当时搭建过程中的关键代码、配置、参数选择逻辑都记录下来你跟着走一遍就能跑通自己的最小闭环。示例代码我用Python来写因为生态最成熟测试和调试都方便。3.1 定义Agent抽象与统一消息结构第一步是定义所有Agent都必须遵循的接口。我设计了一个Agent基类核心方法只有一个process(message) - message。这个设计参考了Actor模型——Agent之间不直接调用对方的方法而是通过消息传递来交互。异步、解耦、易于扩展。from dataclasses import dataclass, field from typing import Any, Callable, Dict, Optional import uuid import json dataclass class Message: msg_id: str field(default_factorylambda: str(uuid.uuid4())) sender: str receiver: str msg_type: str task # task / result / error / heartbeat payload: Dict[str, Any] field(default_factorydict) trace_id: str def to_json(self) - str: return json.dumps(self.__dict__, ensure_asciiFalse)消息结构里我特别加了trace_id字段这个字段贯穿整个流程用于全链路追踪。第一步做框架时很多人会忽视这个字段等到线上排障时才后悔——没有trace_id你根本无法定位一条消息在多个Agent之间流转的完整路径。class BaseAgent: def __init__(self, name: str, description: str): self.name name self.description description def process(self, message: Message) - Message: raise NotImplementedError def handle_error(self, message: Message, error: Exception) - Message: return Message( senderself.name, receivermessage.sender, msg_typeerror, payload{error_msg: str(error), original_msg_id: message.msg_id}, trace_idmessage.trace_id, )每个Agent还需要注册自己的能力描述这样Planner才能知道该把任务分给谁。能力描述不是给代码看的是给大模型看的所以要用自然语言写清楚。3.2 实现Planner Agent把目标拆成子任务Planner Agent是整个流程的起点。它接收用户的原始需求输出一个结构化的任务拆解结果。我这里不直接调大模型而是把拆解过程封装成一个独立的函数方便测试和替换。class PlannerAgent(BaseAgent): def __init__(self, name: str, llm_router: Callable): super().__init__(name, 负责将复杂目标拆解为可执行的子任务列表) self.llm_router llm_router def process(self, message: Message) - Message: try: tasks self._decompose(message.payload[goal]) return Message( senderself.name, receiverorchestrator, msg_typeresult, payload{tasks: tasks}, trace_idmessage.trace_id, ) except Exception as e: return self.handle_error(message, e) def _decompose(self, goal: str) - list: prompt f 你是一个任务拆解专家。请把以下目标拆解为3到8个可并行或串行执行的子任务。 每个子任务必须包含 - id: 子任务编号 - desc: 子任务描述 - agent_type: 需要的Agent类型writer/reviewer/translator/summarizer - depends_on: 依赖的子任务id列表没有则填[] 目标{goal} 只输出JSON数组不要输出任何其他内容。 response self.llm_router(prompt) return json.loads(response)这里有一个很关键的工程细节Prompt末尾强调“不要输出任何其他内容”。大模型默认会输出一大段解释性文字如果不用强约束后面解析JSON就会很痛苦。我在实践中还发现加一个“只输出JSON数组”的约束还不够最好在解析时做一层容错用正则提取出类JSON的子串再做解析。3.3 实现编排器任务路由与状态机编排器是整个框架的心脏。它负责接受Planner的输出把子任务投递给对应的Agent然后跟踪每个子任务的完成状态。状态机我用一个字典来维护每个子任务有四个状态pending、running、completed、failed。子任务状态机的核心逻辑看起来简单但处理依赖关系时需要考虑清楚。我的做法是每轮循环把所有pending状态且依赖全部完成的任务挑出来投递如果一个任务的依赖中有failed状态则该任务直接标记为failed防止下游Agent拿到错误上游数据。class Orchestrator: def __init__(self): self.agents {} self.message_bus [] self.tasks {} self.status {} def register_agent(self, agent: BaseAgent): self.agents[agent.name] agent def submit_goal(self, goal: str) - str: trace_id str(uuid.uuid4()) planner self.agents[planner] plan_msg Message(senderuser, receiverplanner, msg_typetask, payload{goal: goal}, trace_idtrace_id) plan_result planner.process(plan_msg) tasks plan_result.payload[tasks] for task in tasks: task_id f{trace_id}-{task[id]} self.tasks[task_id] task self.status[task_id] {state: pending, depends_on: task.get(depends_on, [])} return trace_id def run_pending_tasks(self) - bool: executed False for task_id, task_info in self.tasks.items(): st self.status[task_id] if st[state] ! pending: continue deps_ok all( self.status.get(f{task_id.split(-)[0]}-{dep}, {}).get(state, failed) completed for dep in task_info.get(depends_on, []) ) if not deps_ok: continue agent_name task_info.get(agent_type) agent self.agents.get(agent_name) if not agent: self.status[task_id][state] failed continue msg Message(senderorchestrator, receiveragent_name, msg_typetask, payloadtask_info, trace_idtask_id.split(-)[0]) self.status[task_id][state] running try: result agent.process(msg) if result.msg_type error: self.status[task_id][state] failed else: self.status[task_id].update({state: completed, result: result.payload}) except Exception as e: self.status[task_id][state] failed self.status[task_id][error] str(e) executed True return executed这段代码里把task_id设计成trace_id-子任务编号的格式好处是不需要额外维护关联表直接通过字符串操作就能找到依赖任务。这种小技巧在工程里的价值远比想象中高简化了很多状态同步逻辑。所有依赖都完成后整个流程就收敛到“所有任务completed或failed”的终态Reporter Agent就可以开始聚合了。3.4 实现Reporter Agent把零散结果汇总成最终交付物Reporter Agent负责收尾。它从编排器拿到所有子任务的执行结果按用户需求的格式合成最终输出。实操时我对Reporter的要求是不仅要汇总还要做质量检查——如果某个子任务的状态是failedReporter不能无视它必须在最终结果中明确提示缺失内容。class ReporterAgent(BaseAgent): def __init__(self, name: str, llm_router: Callable): super().__init__(name, 负责聚合所有子任务结果并生成最终交付) self.llm_router llm_router def generate_final_report(self, trace_id: str, orchestrator: Orchestrator) - str: task_results [] failed_tasks [] for tid, status in orchestrator.status.items(): if not tid.startswith(trace_id): continue if status[state] completed: task_results.append(status[result]) elif status[state] failed: failed_tasks.append(tid) prompt f 你有以下子任务的执行结果 {json.dumps(task_results, ensure_asciiFalse, indent2)} 涉及失败任务需要明确标注{json.dumps(failed_tasks, ensure_asciiFalse)} 请将这些结果整合成一份结构清晰的最终报告不要虚构任何内容。 return self.llm_router(prompt)关于Reporter我一直坚持一个原则Agent只能基于已有结果做组织和润色绝不能自行脑补缺失内容。大模型天然有“补全倾向”面对不完整的上下文它会倾向于生成一个看起来完整的答案。所以Prompt里必须强制约束“不要虚构任何内容”这是我在多次踩坑后总结出的经验。3.5 核心参数的选择模型温度、超时与重试策略这个框架里涉及大模型调用的地方主要是Planner和Reporter我给了这两个Agent不同的温度参数。Planner的任务是拆解需要稳定和准确温度设置为0.1Reporter的任务是整合润色稍微需要一点语言灵活性温度设置为0.3。执行Agent如果只做确定性操作温度直接设为0。超时设置上我给每个Agent的执行时间上限是30秒超过就触发重试最多重试2次。这个时间参数是根据当时所用模型的平均响应速度定的如果你的模型更慢需要自行上调。重试策略上我推荐指数退避——第一次重试等1秒第二次等2秒避免Agent同时在等待时产生惊群效应。消息总线队列方面我设置了最大队列长度100超过则拒绝新消息并让发送方重试防止某个Agent处理不过来时内存暴涨。4. 工具选型与底层支撑组件解析框架搭起来之后还需要几个底层组件支撑。我把这部分单拎出来讲是因为很多教程只讲Agent逻辑不讲基础设施结果读者自己实现时在“工具怎么接入”“记忆怎么存”“上游并发怎么扛”这些问题上卡住。4.1 工具注册机制让Agent具备调用外部能力Agent的价值不只是聊天更重要的是调用工具完成任务。我在框架里设计了一个简单的工具注册表每个工具是一个函数附带名称和描述。Agent在执行任务时如果需要调用工具就从工具注册表里按名称查找并调用。工具调用的设计有两个细节值得注意。第一是工具函数的返回结果必须标准化——统一返回JSON结构包含是否成功、数据和错误信息三个字段。这样Agent和后续的人工排查才能一致地处理工具返回。第二是工具注册表要有权限控制不是所有Agent都能调用所有工具比如计费相关工具只允许财务Agent调用。我把这个控制放在注册表内部Agent调用时自动校验它的角色权限。4.2 记忆与上下文管理短期工作记忆与长期知识分离多Agent系统里记忆管理比单Agent复杂很多。每个Agent既需要短期工作记忆来记住当前任务的关键信息也可能需要长期记忆来保存跨任务的经验。我这里的做法是把短记忆放在消息结构里消息的payload中只携带当前任务必要的信息长记忆则用独立的向量数据库存储Agent需要时通过检索接口获取。我特别强调短记忆要“按需携带”不要一股脑把上游所有输出都塞给下游Agent。这既节省上下文也避免无关信息干扰模型判断。实际项目中我把每个子任务的输入输出控制在2000字以内超过的部分先摘要再传递。这样整套系统的上下文开销是可控的运行速度也稳定。4.3 并发控制多个Agent同时运行时怎么扛住热词里有一条“ai agent怎么扛并发”这确实是生产环境必须面对的问题。Agent框架里的并发控制和普通后端服务不太一样核心在于每个Agent实例不是无状态的HTTP接口它有自己的状态和上下文。我采用的方案是“每个Agent类型维护一个工作池”池里有多个Agent实例编排器按任务量动态把消息分发给空闲实例。并发上最怕的是两个问题共享状态被并发读写导致数据错乱以及一个Agent占用了大量计算资源导致其他Agent饿死。前者我用每个Agent实例独立状态、只通过消息总线交互来规避后者我给每个工作池设置了最大并发数默认是4。合理设置并发上限的比例可以让整套系统的吞吐量提升3倍以上实测下来很稳。4.4 Agent安全两个必须做的控制点Agent安全在公共教程里很少被认真讲但一旦出事就是大事故。我的框架里设了两道安全闸门。第一道是输入过滤所有进入Agent的文本先用内容安全规则扫描拦截危险指令第二道是工具调用审计Agent每次调用工具都会记录一条审计日志包含谁调的、什么时间、参数是什么。另外我强烈建议给Agent设计一个“操作白名单”。只有白名单内的工具才允许调用新增工具必须走审批流程。别觉得麻烦线上Agent被注入恶意指令的情况比想象中多得多。后来我看业内一些Agent框架的安全设计思路也类似都是从输入过滤、权限控制和审计日志三个维度下手。5. 常见问题与排查技巧实录这部分是全文的精华。我按实际踩坑频率排序把最常见的问题和排查方式写出来每一条都是真金白银换来的经验。5.1 Agent之间消息格式不兼容下游解析崩溃这是多Agent系统中最常见、也最隐蔽的问题。表面上看A Agent输出了JSONB Agent解析时报错。深入查发现A输出的JSON字段名和B预期的不一致——一个用的是content一个用的是text。原因往往是两个Agent的Prompt对输出格式的描述不够具体模型在字段命名上有细微偏差。解决方案是建立统一的Schema校验层。我在每个Agent的入口加了一个validate_message函数用JSON Schema严格校验消息结构不合法直接拒绝。这个校验在开发阶段看起来多余但上了生产之后它每天能拦住几十条结构异常的消息。5.2 多Agent死锁A在等BB在等A死锁问题在串行依赖复杂的情况下会出现。比如任务A依赖任务BB又间接依赖A状态机的依赖检测逻辑不完善时两个任务会永远处于pending状态。排查方法很简单在编排器里加一个心跳日志定期输出所有任务的状态。一旦发现某个任务超过N轮循环还没变化系统就自动把它标记为failed避免流程卡死。我设置的是5轮实测不会有正常任务被误杀因为正常任务最多两轮就能执行完成。5.3 上下文污染前一个任务的信息干扰后一个任务的判断这个问题的表现是Agent在处理任务时莫名其妙提到与当前任务无关的内容。原因是消息总线复用了同一个Agent实例而Agent的上下文窗口里还残留着上一个任务的信息。修复方案是在任务切换时强制重置Agent实例的上下文。我把每个Agent设计成无状态的消息处理器——它不维护跨任务的记忆每次process调用都从一个干净的上下文开始。如果确实需要跨任务记忆走独立的向量数据库不塞在上下文里。5.4 大模型输出不稳定JSON格式飘忽不定大模型输出不稳定的问题在Agent框架里被放大了因为Agent之间的通信严重依赖结构化数据。我的解决办法是所有模型输出先经过一个解析层做三件事——用正则提取类JSON片段、用宽容模式解析容忍尾逗号、单双引号混用、解析失败时自动重试一次。如果重试还不行就降级为纯文本传递。这个降级机制虽然不完美但保证了流程不被卡死。在框架早期版本里我遇到过连续三次解析失败导致整个流程失败的案例后来加了解析容错和降级策略基本就不再出现这种问题了。5.5 如何快速定位是哪个Agent出了问题多Agent系统最大的排障难点是定位问题。我的经验是链路追踪tracing能力必须一开始就建好。前面提到的trace_id在这里派上大用场。所有Agent在处理消息时都记录日志日志格式包含trace_id、agent_name、message_id、处理耗时和结果状态。排查时只需要拿一个trace_id去日志系统搜索就能把这条任务从Planner到Reporter的完整路径拉出来一眼看出哪个Agent耗时异常或返回错误。这个能力让排障时间从小时级降到了分钟级是我在所有Agent项目里都会保留的基础组件。6. 这个框架后续还能怎么扩展多Agent协作框架搭好之后扩展方向非常明确。我这里分享几个我自己验证过可行的方向以及一个我认为非常有前景的前沿思路。第一个方向是接入因果推断能力。热词里提到的“因果强化学习CRL”思路在实际Agent场景中有一个非常实用的变体——让Agent在执行决策时不仅看“相关性”还尝试判断“因果性”。比如说一个Agent发现某种处理方式经常导致下游任务失败它可以基于这些历史数据做简单的因果分析不只是记住“这样会导致失败”而是进一步判断“在什么条件下会导致失败”。这种能力可以让Agent的决策质量上一个台阶。第二个方向是让Agent具备工具编排能力。现在的框架里工具调用是Agent自己决定的但工具与工具之间的编排还是靠代码写死的。扩展方向是可以让Agent动态生成工具调用序列类似让Agent写一段小流程脚本去执行。这个方向自由度更高但需要更严格的沙箱和权限控制。第三个方向是把人工审批节点插入流程。很多实际业务场景里某些步骤需要人工确认才能继续。我给框架加了一个human_in_the_loop节点Agent执行到该节点时暂停等待人工通过接口确认后继续。这个扩展极大提升了框架在真实业务中的可用性。第四个方向是模型异构。现在框架里所有Agent都调用同一个大模型但实际不同Agent的任务难度差异很大。Planner任务复杂适合用能力更强的模型普通执行Agent任务简单可以用更轻量的模型。在框架里按Agent类型配置不同的模型Endpoint能在几乎不损失质量的情况下把成本降一半。最后分享一个我个人的体会多Agent框架最关键的瓶颈往往不在技术而在任务拆解的合理性。拆得过粗每个Agent的负担就重退化回单Agent拆得过细消息传递开销太大上下文碎片化严重。最理想的状态是每个Agent都能在一个清晰的边界内用最少的上下文完成自己的职责。这个平衡没有标准答案只能根据你的业务场景反复调试。我每接一个新项目至少会花三分之一的时间在任务拆解的调优上这个投入非常值得。