1. 为什么需要一套代理的中介从单智能体到多智能体的痛点今年以来我一直在做企业内部的一个知识问答系统最初版本非常简单一个专用模型接上知识库检索问一句答一句效果也还行。但业务方很快不满足了他们想把“查合同、看竞品、给研发估值”这类跨领域问题丢给系统一次问完拿到一份综合报告。我试过把所有工具塞进同一个智能体让一个大模型自己挑函数调用结果很尴尬——上下文越滚越长模型开始抓不住重点甚至编造接口参数工具超过十个以后模型经常在几轮调用里绕圈子明明有现成结果还要去重复查询。后来我意识到问题不在于模型的能力边界而在于我们把“思考”和“协作”塞进了同一个上下文。真实世界的组织里没有一个人能掌握所有细节都是几个人开会、分工、汇总。多智能体系统就是想把这种分工建到代码层。可一旦真的引入多个智能体新的麻烦立刻出现消息格式没有统一规范任务依赖关系靠硬编码状态散落在各个进程里Agent之间互相等结果等到超时偶尔还会出现A把B的错误结论当成依据继续推理“幻觉放大”成更离谱的答案。这就是我启动 agency-agents 这个项目的直接原因。它不是一个聊天助手也不是一个单模型封装而是一层位于“大模型与工具、模型与模型之间”的编排底座专门解决三个核心问题任务怎么拆、消息怎么传、状态怎么收敛。简单说它是“代理的中介”让多个智能体能像靠谱同事一样协作而不是像一群抢话筒的实习生。这篇文章就围绕我在设计、开发和落地 agency-agents 过程中的完整思路来写。如果你也在做多Agent应用或者被“工具调用乱成一团”困扰这篇文章里的架构决策和踩坑记录应该能帮你省掉不少试错时间。2. 项目定位轻量级智能体编排底层的四个基本模块在最开始立项时我给自己定的原则是不重复造大模型轮子只解决协作问题。当前的大模型推理能力已经很强Agent框架也很多但多数框架把重心放在“如何让一个Agent更聪明”上对“多个Agent如何低成本地组织起来”反而支持得很粗糙。agency-agents 恰恰定位在后者。2.1 通信模块先统一消息协议再谈智能多个智能体协作最容易死掉的地方就是消息格式。A角色输出一段 Markdown 风格结论B角色期待的是 JSON 格式的可执行指令中间如果没有翻译层每次对接都是一场灾难。我在通信模块里定义了两种消息类型Command和Event。Command是必须执行的动作比如“请输出成本估算”发送方希望接收方完成动作后回传结果。Event是事实性通知比如“合同条款已更新”接收方可以决定是否响应不触发直接回包。所有 Agent 之间的交互都通过这两类消息进入一个统一队列。上层再按照sender、receiver、correlation_id三个维度做消息路由。correlation_id尤其重要它相当于现实中“这个项目跟的是哪张工单”如果没有它多个并发任务交错时Agent 根本分不清自己处理的是哪一批请求。2.2 调度模块DAG 编排而不是 “开会直播”大部分 Agent 协作框架用的都是对话式编排——一个 Agent 说完另一个 Agent 接话像聊天室一样自由。但这种模式在多任务并行时很不可控。我选择用**有向无环图DAG**来建模任务流。一个复杂任务被分解成一连串节点每个节点是一个 Agent 的执行单元节点之间有明确的依赖边。下游节点只要等上游节点完成不需要关心上游 Agent 是怎么思考的。from agency_agents.scheduler import DAGTask, NodeStatus dag DAGTask(task_idproject_valuation) dag.add_node(contract_analysis, agent_nameclause_reader, depends_on[]) dag.add_node(market_review, agent_namemarket_scout, depends_on[]) dag.add_node(risk_check, agent_namerisk_assessor, depends_on[contract_analysis, market_review]) dag.add_node(final_report, agent_namereport_writer, depends_on[risk_check])这段代码描述了一个典型的“两条支线并行最后汇总”的结构。contract_analysis和market_review没有依赖关系调度器会同时调度它们risk_check必须等前两者都完成后才启动final_report最后收口。这种显式的依赖关系让任务的并发性一目了然也方便中途插桩做监控。2.3 记忆模块不是存聊天记录而是存“工作记忆”多智能体最大的隐性成本是重复劳动。A 已经查过“客户近三年营收”B 不知道又去查了一遍。我以前见过某系统里同一个检索工具被四个 Agent 依次重复调用浪费了十几倍的 token。agency-agents 在通信层下面内置了一个工作记忆存储每个 Agent 在执行关键动作前会先检查记忆库中是否有匹配条目。记忆条目的 key 由任务语义和实体名联合生成比如revenue:client_x:2023-2025。如果命中就把记忆中的结果当作历史事实返回不再调用外部工具。初始版本我用 Redis 做存储后来发现单机内存也能胜任就换成了可插拔接口默认实现是一份带过期时间的本地字典。2.4 执行模块把“思考”和“工具调用”彻底分离很多主流的单 Agent 框架会在模型内部循环调用工具这对多智能体而言很不好排查。agency-agents 在执行模块强制约束Agent 只能输出一个“待执行动作列表”不能直接触达外部 API。2.4 执行模块的逻辑打个比方Agent 像是项目经理执行模块是后勤团队。项目经理只负责下达“去打印资料”的指令真正操作打印机的是后勤。这样做的最大好处是所有外部调用都有统一出入口可以集中加日志、加鉴权、加超时控制。我把它设计成三层动作解析层从模型回复中提取结构化动作比如{tool: knowledge_search, args: {query: ...}}。工具注册中心每个工具在启动时注册自己的参数 schema动作解析层会校验参数是否合法。执行结果包装层工具返回的原始结果统一被包成标准消息回传给对应 Agent同时备份到记忆模块。这套分离思路帮我在后续排查中节省了大量时间。当某个 Agent 产生离谱结论时我可以直接看 execute log明确知道它调用过什么工具、拿到了什么结果而不是去猜模型“心里想的是什么”。3. 核心实现拆解任务分解、消息路由与状态收敛如果说上一部分解决的是“模块怎么摆”这部分就是更关键的“机制怎么转”。我把多智能体协作的核心机制归纳为三件事任务分解Task Decomposition、消息路由Message Routing、状态收敛State Convergence。agency-agents 的整个代码逻辑本质上都是围着这三件事转。3.1 任务分解让每个 Agent 只接“一平方米”的活没有任务分解多智能体就退化成“多个单智能体轮流讲废话”。我在调度器里内置了两个分解器RuleBasedDecomposer和LLMDecomposer。规则分解器适合流程稳定的业务场景。例如评估一个项目可以固定拆成合同分析、市场调研、风险判断、汇总报告四块每块对应一个 Agent 角色。这种方式可控性最强代价是无法应对没见过的新场景。LLM 分解器则是把“用户的一句话需求”交给一个专门负责拆解的“规划 Agent”让它产出一份候选 DAG。这里最大的技巧是必须给规划 Agent 一个动作预选集。否则它可能把任务拆成几十个无关紧要的子任务或者提出一个本质上是单步骤却硬拆两段的方案。我在实验中发现预选动作列表能显著提高 DAG 的可执行率从 61% 提升到 88%。其实也很好理解——约束越明确模型发挥越稳定。预选动作列表本身不是固定的可以从工具注册中心动态拉取。比如注册中心里有knowledge_search、code_analyzer、cost_estimation三个工具规划 Agent 就只能围绕这三个动作组合成流程。这样后续新增工具时只要完成注册规划 Agent 能自动感知。3.2 消息路由既要找对人的嘴也要找对耳朵路由机制我设计了三级。第一级是基于角色的路由——每条消息头部都带有目标角色比如receiver: risk_assessor。这是最简单的路由适合固定团队的场景。第二级是基于语义的路由——当消息没有明确目标时由路由模块做一遍文本匹配判断当前消息内容跟哪个 Agent 的能力描述最相关。这一层我一开始想用向量库后来发现没必要那么重直接用关键词LLM 分类就够了。因为 Agent 数量通常不会超过十个把问题交给现有大模型判断并不会带来太多额外成本。第三级是广播路由——用于“重要信息同步给所有人”。比如上游结果出了偏差广播消息让所有 Agent 停止基于旧信息继续推理。这对应了现实团队里的“暂停一下前面数据有问题”场景。广播消息不会被普通路由过滤而是会直接并入每个 Agent 工作记忆里作为“状态修正”。路由模块的性能瓶颈通常是背压控制。多个 Agent 并行执行时如果某个 Agent 处理速度慢它的输入队列会不断积压。我在队列层做了长度限制超出后不再接收新消息而是向上层报告“暂不可用”由调度器决定等待或降级。这样一个慢 Agent 不会拖垮整个队列反而会触发预警让我们尽早定位瓶颈。3.3 状态收敛白话讲就是“结论要对齐”多智能体系统里最忌讳的是各干各的最后结论撞不到一起。我见过一个系统市场分析 Agent 给出了“需求强劲”的意见风险评估 Agent 又指出“该市场政策风险极高”汇总 Agent 完全懵了。这就是状态不收敛。agency-agents 里我在两个环节强制状态收敛。第一环是依赖节点的归并检查。DAG 中每个有多个上游节点的汇聚节点调度器都会运行一个convergence_check。它的功能是比对上游结论中的关键字段检测是否存在互相矛盾的信息。例如上游 A 说“成本约 100 万”上游 B 说“成本超过 300 万”差异超出阈值convergence_check会返回一个冲突报告。此时不会直接给下游 Agent 继续执行而是先触发冲突解决流程。第二环是最终报告生成前的一致性确认。由汇总 Agent 生成最终结论之前系统会要求汇总 Agent 引用每条结论对应的来源消息 ID。没有来源支撑的句子会被标记为“无证据语句”再交给校验 Agent 复核。这一步对拦截模型的“脑补”非常有效。当然状态收敛的力度是要分级的。低风险场景比如娱乐资讯推荐三句话里有一句不正确问题不大但涉及合同、财务建议时收敛检查必须更严格。我在系统里配置了两档阈值分别是宽松模式和严格模式。严格模式下任何没有来源支撑的断言都不能通过校验宁可反馈“待补充”也不能输出模棱两可的答案。4. 一次完整协作三代理跨域咨询的调度与消息流为了让你直观感受这套系统怎么跑我用一个“跨域项目评估”场景来做全流程演示。三个角色分别是角色 A条款分析师工具合同文本解析器角色 B市场调研员工具行业知识库搜索角色 C风险评估师工具风险规则引擎 财务计算器用户提出请求“评估某产品线是否值得在当前阶段立项并分析主要风险点。”调度器对这个任务做分解产出两路并行子任务A 分析相关合同约束B 收集市场与竞品信息。两者完成后C 从两类结果中推断风险最后汇总报告。在实际运行日志里消息流是这样走的[09:00:01] orchestrator - decomposer: 任务ID_T202 开始分解 [09:00:02] decomposer - scheduler: DAG 生成节点 [A, B, C, DA] [09:00:05] scheduler - agent_A: Command {action: parse_contract, doc_id: C-2031} [09:00:05] scheduler - agent_B: Command {action: search_market, query: 产品线X 市场规模 2025} [09:00:23] agent_A - memory_store: 写入线索三年内不可转让核心技术 [09:00:37] agent_B - memory_store: 写入线索同类产品市场增速约 18% [09:00:40] scheduler: A 和 B 均完成触发风险判断节点 [09:00:41] scheduler - agent_C: Command {action: assess_risk, inputs: [A_Result, B_Result]} [09:00:52] agent_C - scheduler: 返回风险项列表含两条红线一条中风险 [09:00:53] scheduler - agent_D: Command {action: generate_report, inputs: all_results} [09:01:20] agent_D - user: 综合报告每条结论带来源 ID这里有个细节值得说明agent_A 写完记忆后没有直接把结果发给 agent_C而是先把事件提交到调度器由调度器统一转交。为什么这么绕因为如果让 Agent 自己发消息就没法统一跟踪依赖关系也无法在中间插入状态收敛检查。调度器就像是团队里的项目助理所有跨界信息都经它手但项目助理并不参与具体思考只负责“传递”和“记录”。用户最终拿到的报告格式也是结构化的每个断言后面跟着source_refs字段提供一个或多个上游消息 ID。比如“技术转让限制构成主要约束”这条结论source_refs指向 agent_A 的条款分析消息。如果后续发现结论有误可以通过消息 ID 反查问题出在哪一步这比“对着黑盒反思”高效得多。整个流程里两个并行 Agent 同时工作耗时主要由慢路径决定。在我本机测试数据集上串行执行三个步骤大约需要 95 秒并行后缩短到 58 秒。这不是了不起的数字但它证明了 DAG 调度带来的收益是实实在在的。5. 踩坑清单我在实际迭代里的四个真实教训框架能搭出来是一回事稳定好用是另一回事。我在从 demo 走向真实业务的过程中前后踩了不少坑。下面这几个是我认为最有共性、也最容易复现的拿出来细讲。5.1 死锁两个 Agent 永远在等对方第一次遇到死锁是在一次双 Agent 协作测试里。Agent A 的处理逻辑是“必须先拿到 B 的结果再继续”Agent B 的处理逻辑是“必须先拿到 A 的结果再继续”调度器没有环检测两个 Agent 就一直隔空对峙系统卡死到超时。原因不复杂在对话式协作中这种互相等待可能在几轮内自然打破因为模型总会输出一点内容但换成 DAG 调度后依赖关系变成显式的如果构建 DAG 时漏掉某个边或者把 A→B 和 B→A 同时建出来就会出现环。解决办法是在调度器里加了一个环检测当某一节点发现所有上游节点都已经完成但自己仍在等待一个“永远不会到来的依赖”时直接进入 FAILED 状态并抛出异常。这个逻辑听起来简单但在最初版本里我确实没做线上任务卡了十几分钟才发现。5.2 上下文漂移Agent 说着说着就忘了上一轮结论第二个坑是上下文漂移。我最初把每个 Agent 的上下文设计成一个会持续累积的列表方便它参考历史内容。结果十几个任务并发跑下来Agent 的上下文被无关信息污染开始把其他任务的数据混进当前任务的结论里。后来我把上下文存储改成两层隔离全局记忆all agents 可见只放经过状态收敛验证过的公共结论。局部记忆当前任务且当前 Agent 可见只在当前 DAG 节点内有效节点完成后立即归档。这样虽然损失了一些跨任务的“知识迁移”但也杜绝了上下文污染。业务上如果需要某个 Agent 学习历史经验可以通过“结论摘要注入”来手动补而不是任由原始消息自动混入。5.3 无界队列慢 Agent 拖垮整个队列的内存我用队列通信时最初没有设置最大长度。某次一个角色在调用外部 API 时频繁超时每次重试都要重新执行整段逻辑上游 Agent 产出的消息不断堆积内存涨到 3GB 以上最终 OOM。后来我给每个队列加了max_len和drop_oldest策略超过阈值后新消息不再进入返回“队列已满”错误调度器收到该错误后会查看该 Agent 是否还有更早的未处理消息若有则说明下游跟不上上游自动降低上游的消息生产频率。这套背压机制在真实场景里的价值比我想象中大得多——它不仅保护了内存也避免了一次故障放大成雪崩。5.4 幻觉互相放大一个错误结论被多个 Agent 转发变成“事实”最后一个坑比较隐蔽。在自由协作模式中如果 Agent 没有强制校验来源一个错误的中间结论会被后续多个 Agent 引用并加工最终报告里呈现出一种“多个角色都印证过”的假象。举个例子某次测试里 Agent A 把合同年份误读成 2021 年B 基于这个错误数据做了一个市场趋势对比C 又基于 B 的结果给出“该产品已过窗口期”的建议。最终报告看起来很严谨但根因只是 A 的第一次 OCR 识别错误。这个问题光靠“提示模型要小心”解决不了必须在机制上强制每条结论携带来源 ID并在汇聚点实施交叉验证。我在状态收敛模块中加了一个FactCrossChecker它会把汇总 Agent 输出的每个断言拆成“主体-谓词-客体”三元组然后在所有上游消息中搜索可以支持该三元组的句子。如果支持度低于阈值该断言会被返回给汇总 Agent 修改而不是直接发给用户。为了不拖慢速度这个检查和 DAG 汇总节点的生成是异步并行的汇总 Agent 输出后立即核查不等全部上游历史扫描完毕。6. 下一步演进从框架到带可观测性的协作生产力工具经过几轮迭代现在的 agency-agents 已经能稳定支持多个并行 Agent 完成跨域任务。但我的真实感受是编排能力只是第一步可观测性和治理能力才决定系统能不能真正进生产。下一步我准备围绕三个方向继续做改进。第一个方向是更细粒度的执行追踪。目前的日志基本能定位“哪个 Agent 发了哪条消息”但缺少跨 Agent 的全链路 trace 视图。我计划把correlation_id串到所有消息、记忆写入、工具调用日志中然后构建一个时间线视图让开发者像看分布式调用链一样审视每一次 Agent 协作。这个能力在排障时能节省一半时间。第二个方向是人工介入点。AI Agent 的输出再顺滑关键决策还是需要人来把关。我准备在 DAG 节点上增加requires_human_review标记。当某个高风险节点的上游结论出现冲突时调度器不会自动尝试解决冲突而是先停止通知相关专家人工介入。虽然这会牺牲部分自动化率但对于评估、决策类业务它能让 AI 系统真正“值得信赖”。第三个方向是收集更多真实场景的负反馈做成一个回归测试集。每当我遇到“Agent 生成了一份看起来合理但实际错误的报告”就把这个案例加入测试集并标记出错位置。下次修改框架后用这批案例跑回归防止旧问题复现。这有点像传统软件工程里的单元测试只不过测的是“多智能体协作流程的正确性”而不仅仅是单个模块的输出格式。最后再分享一个我个人的体会做多智能体编排最容易兴奋于各种“智能”的涌现但真正让系统稳定运行的往往是那些很朴素的工程约束——消息格式统一、依赖关系显式、队列有界、结论有来源。把这些枯燥的基础设施做好了大模型的智力才能真正被释放到业务价值上。这套框架的层面还远谈不上完美但如果它能帮你把多 Agent 协作中那些“看不见的混乱”挡在系统之外我就觉得它值得继续打磨下去。