从单Bot到多智能体协作：Agent编排框架的设计与落地

在把单一聊天机器人包装成“服务”去对外接单的时候我越来越觉得一个人同时扮演客服、策划、文案、审校、交付几个角色效率迟早会撞墙。于是就有了这个代号叫agency-agents的项目一个把大模型能力拆成多个可分工、可路由、可回退的智能体协作框架。这篇文章会完整拆解它的设计动机、系统骨架、最小可跑Demo以及我在真实压测中踩过的几个典型坑适合正在做多智能体编排或想用AI能力接单变现的开发者直接参考。1. 为什么叫“agency-agents”从单Bot到多智能体协作的转变如果你只是写一个ChatBot你不需要这个项目。真正让我决定动手的原因是当任务从“陪人聊天”变成“帮客户完成一项完整的交付”时单一大模型的短板会暴露得非常明显。1.1 单个大模型解决不了“完整交付”问题什么是完整交付客户说“帮我写一套小红书种草文案包含5篇图文笔记”听起来是一个能力实际拆开却是四件事先搞明白产品卖点是什么再找竞品/评论区的风格素材然后生成不同口吻的正文最后还要检查有没有违规词、有没有过度承诺。如果全部丢给同一个对话线程会出现三个典型问题第一上下文会越长越乱聊到后面模型开始遗忘第一轮的产品信息第二不同环节需要的能力侧重点不一样一个模型参数很难同时满足“创意发散”和“严格审校”第三没有分工就意味着没有问责一旦输出质量下降你根本不知道是哪一个步骤先坏的。1.2 “代理机构”隐喻带来的架构启发“agency”这个词本身是代理、中介的意思。我借用的是它“机构化分工”的那层含义一个真正的服务型外包公司里有客户经理负责接单拆需求有执行人员负责干活有质检员负责验收还有项目经理盯进度。agency-agents 就是把这一套角色搬到代码里Router路由分身接收用户任务分析意图拆解成子任务并分派。Worker执行分身按照角色预设完成具体的生成、提取、格式化工作。Critic质检分身检查Worker的结果是否符合要求不合格就打回重做。Tool Executor工具分身负责调用外部API、搜索、读写文件、操作数据库等非纯文本能力。每个分身都是独立的Agent单元共享一套消息总线。这样做的好处是任何一个环节的模型能力升级或Prompt调整都不会影响其它环节而且每个环节可以独立测试。对于一个个人开发者或小团队来说这种模块化的可测试性比什么都重要。1.3 这个框架适合谁如果你只是想给公众号加一个自动回复机器人那确实用不上这套东西棒槌打鼓。但如果你在做下面这几类事情我希望你从里面的思路里能拿走点东西想用AI替客户做内容代运营或自动化流程搭建的独立开发者给企业内部做知识库问答、工单分拣、报告生成这类复杂业务系统的工程师研究或学习多智能体编排想理解Agent协作机制的学生和研究者。这套东西最忌讳做的事就是“为了多Agent而多Agent”。如果你的任务在一次Prompt里就能完成得很好别硬拆那是浪费token和延迟。这是我整个项目里最重要的一个判断准则后面所有设计都围绕它展开。2. 系统骨架设计角色、路由与任务总线的基本盘多智能体系统的设计本质上是在回答三个问题智能体有哪些角色任务怎么流转信息存在哪里这一节我会把agent-agents的骨架拆开讲清楚这些决策你在重构任何Agent项目时都能复用。2.1 角色的三段式抽象我最后把Agent抽象成三个基本单元而不是搞一个很复杂的基类继承树单元类型职责输入输出Planner/路由器读懂任务目标拆分子任务决定调用链用户原话 系统上下文任务清单 依赖关系Worker执行具体内容生成/加工可搭配工具子任务描述 所需资料结构化的结果JSONCritic基于评分标准检验Worker的结果原始任务 Worker输出通过/打回 修改意见这三个抽象对应着真实流水线上的“领导、干活的人、质检员”。段位低的框架会写一堆长得差不多的Agent类区别只在Prompt里段位高一点的框架会干脆把角色差异全部收敛到配置里——同一个Worker类换一组工具注册表和一个角色Prompt就变成完全不同的员工。2.2 路由的主动权放在“意图匹配”而不是“关键词”很多初学者做路由喜欢先写一个意图分类函数用一堆if else或正则去判断用户输入里有没有“写”“画”“总结”这类词。这套方案在真实场景里基本就是纸糊的——客户说“帮我出一份活动复盘”里面既没有“总结”也没有“写”你怎么匹配在 agency-agents 里路由本身也是一个Agent调用。Router收到原始输入后会先让大模型输出一个结构化JSON包含任务类型、目标受众、交付格式、约束条件四个字段然后代码再根据这个结构化结果去匹配执行链。{ task_type: content_generation, goal: 撰写5篇小红书种草笔记, audience: 25-35岁护肤人群, format: markdown, constraints: [避免绝对化用语, 每篇不少于400字] }这一步看起来多花了一次模型调用但它带来的稳定性收益极大。一来大模型做意图拆解的准确率远高于正则匹配二来这个JSON本身就是任务的可追溯档案后面Critic在审校时可以直接拿着约束条件逐条核对而不是凭感觉打分。路由的延迟代价在绝大多数场景里是可以接受的因为它避免了错误分派人导致的重跑成本。2.3 消息总线用队列而不是直接函数调用这是我对整个框架最满意的一个设计决策Agent之间不直接互相调函数而是通过一个内存消息总线发布任务和结果。直呼函数有什么问题同步、强耦合。A想调B就得知道B的类名、方法签名、异常类型调起来一时爽Debug火葬场。而且一旦某个环节执行时间长了整个调用链都得阻塞在那里等。改成消息总线之后每个Agent变成了订阅者Router发布一个task.created事件。Worker订阅这个事件处理完发布task.completed。Critic订阅task.completed判断是否通过不通过就发布task.rejectedRouter再接住重新派活。异步的好处是你可以自然地把不同Agent放进不同线程或进程里跑单个环节失败了也不至于拖垮全链路。坏处也很明显你要处理的消息状态突然多了一个数量级“消息没送达”“消息重复消费”“消息丢了一半”这些问题全来了。所以我强烈建议如果你要在生产环境跑消息总线里至少要带一个最简单的状态流转表记录每个taskId当前处于什么状态。2.4 任务状态的五态流转任务的流转状态我设计成五个queued已入队、running执行中、waiting_critic待校验、approved已通过、failed执行失败。这五个状态是硬编码在框架里的。每个状态还配了一个超时时间比如Worker执行超过90秒没回消息总线就自动标记running失败触发一组兜底逻辑先重试一次换一个更小的模型再失败就降级为直接返回错误信息给用户。没有这套状态机的时候我遇到过Agent卡死之后整个任务悬在半空、用户那里既没结果也没报错的情况那是所有线上事故里最让人抓狂的一类。3. 让Agent学会用工具函数协议与外部能力的接入缝纯靠大模型吐文字它永远只能做“内容顾问”做不了“自动化执行方”。Agent真正值钱的地方在于它能调用工具、操作真实系统。这一节我从协议设计讲到具体注册方式都是可以直接抄走的部分。3.1 为什么工具调用不能靠让模型“直接执行代码”在最早的原型里我走过弯路让模型直接输出Python代码再由我在沙箱里执行。这个方案缺点很致命——输出代码的格式千奇百怪模型偶尔还会在代码里夹带它自己的假想数据而且你几乎没法对它做精细的权限控制一旦代码有恶意倾向沙箱只能兜底不能预防。后来换成了现在主流的函数调用方案我把每一个能力包装成带有明确参数Schema的函数模型只负责根据用户意图“选择函数并填入参数”真正的执行由框架代码完成。这相当于把模型从“操作员”降级成了“调度员”它手里拿着我预先核准过的操作按钮能按哪个、不能按哪个全部由我决定。3.2 工具注册表的设计在 agency-agents 里我没有为每个Agent单独实现一套工具逻辑而是建了一个全局工具注册表。注册表里存的是工具名称、描述、输入Schema、以及执行函数四个部分。tools_registry {} def register_tool(name, description, schema): def decorator(func): tools_registry[name] { name: name, description: description, schema: schema, handler: func, } return func return decorator register_tool( web_search, 在互联网上进行信息搜索返回相关网页摘要列表, { query: string, max_results: int } ) def web_search(query, max_results5): # 内部调用某搜索API解析结果后返回 return search_results这样Worker Agent的系统Prompt里我只注入每个工具的名称、描述、参数说明不给完整函数实现。模型的输出会被框架解析成{tool_name: web_search, parameters: {...}}随后框架查注册表调用对应的handler再把结果拼回上下文交还模型。正是这个“注册表Schema”的结构让后面加新工具变得极其简单。任何同事要给框架增加一个“查订单”能力只要写一个注册函数填好Schema告一截儿。能力边界和参数约束都由Schema兜着根本不担心模型乱传参。3.3 工具结果需要先粗加工再送回模型直接把100条搜索结果全文塞回给模型是新手最容易犯的错。哪怕上下文窗口再大塞进去的有效信息密度都很低模型反而会迷失在冗余信息里。我现在的做法是在工具handler里就预先做粗加工搜索类工具只保留标题、来源、发布时间、前200个字符。数据库查询类工具只保留前10条记录并转成紧凑的JSON数组。文件读取类工具只保留关键段落或特定Sheet列。做完粗加工还要加一行“元信息注释”。比如搜索结果的末尾补一句“以上内容来自公开网页检索未经验证请注意甄别事实”这能显著降低模型把搜索内容当确定事实直接引用的概率。虽然不能完全杜绝但至少能提高它对信息源不确定性的警觉这是我在多次幻觉事故里总结出来的血泪经验。4. 跑通一个最小闭环从任务派发到成稿审改的完整日志工具和角色都有了理论讲得再多不如跑一遍真实流程。我在这个项目里选的最小验证场景是我当时正在接的一个真实需求给客户做一套活动预告文案。下面就是完整跑通的日志每一步我都会标出里面的关键设计点。4.1 输入与首轮路由拆解我输入的原始需求只有一句话“帮我写5篇不同风格的活动预告推文目标人群是大学生要突出早鸟票优惠。”这句话信息量很少连活动主题、时间地点都没有但放在真实接单场景里客户往往就是先给你一句话看你的反应。Router第一轮做的事是“需求澄清”而不是直接动笔。它输出的结构化JSON如下{ task_type: content_generation, goal: 为早鸟票活动撰写5篇不同风格的预告推文, missing_info: [活动名称, 活动时间与地点, 票价档次], execution_plan: [ 先确认缺失信息若无回复则基于通用假设继续生成并标注假设内容, 生成5篇不同风格的推文初稿, 按平台规则审查违规词与绝对化用语, 输出最终版本 ], contingency: 若客户未补充信息生成时使用【活动名称待定】等占位符并在交付说明中单独列出待确认项 }这里有个非常实用的技巧执行计划里加入了回调方案contingency。因为在真实场景里客户经常不回信息而交付截止时间又卡在那里你不能因为缺信息就停工。在占位符基础上继续生成再把缺失项单独列出远好过干等。模型这个时候的“心里有数”完全来自于Router那一条“即使缺信息也要给可落地执行路径”的Prompt约束。4.2 Worker执行的模型选型与参数控制拆解完成后Router把5篇推文的任务分发给了5个独立的Worker实例。这里我特意没有用一个Worker循环生成5篇而是起了5个并行Worker因为风格之间如果放在同一上下文里很容易相互污染——模型会在写第二篇的时候不自觉地模仿第一篇的语气最后交出来全是一个调调。每个Worker的模型参数也不一样创意发散任务的大模型温度调到了0.85尽量放飞风格而负责统一事实信息比如早鸟票价格、活动地点的那个Worker温度调到0.2确保输出稳定准确。参数创意写作Worker事实校验Worker模型温度0.850.2上下文长度4000 tokens2000 tokens超时时间120秒30秒输出格式Markdown正文JSON事实清单很多人搭建Agent时把所有Worker用同一套参数这是不对的。模型温度本质上是输出的确定性开关高温度适合发散创意低温度适合稳定抽取。如果反过来用创意Worker会收敛到无聊的套话事实校验Worker则会给出各种花哨但不靠谱的事实表述。4.3 Critic的评分细项与打回逻辑5篇初稿完成后全部进入Critic审核。Critic做的事不是泛泛地打一个“写得好不好”的分而是拿着硬性规则逐条过是否包含指定的优惠信息是否有“全网最低价”“百分百有效”等违规绝对化用语是否有大学场景可感知的语境表达5篇之间的开头句式是否重复其中两篇被打回了。一篇因为用了“史上最强优惠”这种涉嫌诱导并且也可能违规的说法另一篇是因为开头连续三篇都用了“还在为XX发愁吗”的句式风格区分度太低。Critic在打回时不是简单说“不行”而是输出了一条结构化修改意见{ decision: reject, reason_code: forbidden_word_detected, problem_snippet: 史上最强优惠, suggestion: 替换为限时早鸟价前200名有额外礼赠 }Worker拿到这个修改意见后重新生成再交给Critic复检直到通过。整个链路很像真实工作流里的“写稿-审校-返工-确认”。这个反馈闭环是质检Agent存在的真正价值——如果没有它错误内容就会直接送达用户那你的专业度口碑就一次清零了。4.4 一次典型完整跑通的耗时数据我测下来的平均耗时大概是这个水平路由拆解0.3秒本地小模型5个Worker并行执行平均下来在12秒左右Critic审校3秒打回重做一次4秒。整体从用户提交到拿到终稿大约20秒。这里有个值得注意的结论并行起了很大作用。如果Worker串行跑每篇推文要等上一篇完成总耗时可能翻倍到40秒以上。代价则是并发成本假设一次单任务的输入输出token总和是万级5个Worker并发大概消耗五倍左右的token成本。对个人接单来说单次任务的API开销折算后仍然利润可观但对高并发的大规模任务你需要仔细平衡并发层级——不是所有子任务都需要并行。5. 实测避坑上下文污染、循环调用与并发控制的经验清单项目跑了快两个月这个框架在真实场景里给我上了一课又一课。这一节不藏私把我遇到过的、最容易在同类系统里复发的三类问题全列出来每个都附上我的处理方案。5.1 上下文污染多子任务共用一个上下文的后果最早我图省事让同一个Agent上下文里连着处理两三个子任务结果就是灾难。第二个子的任务生成时模型不仅带了第一个子的任务风格还会把第一个子任务的输出内容当成背景素材引用导致第二篇推文里出现第一篇文章里的产品名称。后来我做了严格隔离每个Worker处理完一个taskId后它的上下文会自动关闭并归档新任务到来时Worker拿到的上下文只包含系统Prompt、该任务的拆解结果、相关工具返回结果其它的历史消息一概不注入。这个“一次任务一上下文”的规矩救了无数次交付质量代价是每次任务都要重新灌系统Prompttoken消耗略有上升但相比出错返工的成本这点开销完全值得。5.2 重试逻辑模型有时会钻牛角尖我发现一个特别有意思的现象Critic打回一次之后Worker如果继续用同一个上下文重写大概率第一次能改好但如果打回两次以上模型会开始“摆烂”——它不再认真修改而是把原文压缩一下改几个形容词就交回来甚至会在修改意见里找漏洞。解决办法是引入所谓的“轮次衰减策略”第一次返工时沿用原上下文第二次返工时清空上下文重新组装任务描述第三轮就换一个更低温度或换一个完全不同的模型来处理。连续打回超过三次系统不再自动返工而是走人工介入通道。这个阈值看起来很死板但在真实业务里超过三次还搞不定的内容基本是任务描述本身就有问题继续让模型白耗已经没有意义。5.3 并发控制与限流别把API额度烧干净并行Worker很爽但也容易失控。我遇到过最惊险的一次是给一个紧急需求开了8个并发Worker每个Worker内部还调了3次搜索工具和2次生成调用结果一分钟内打了近50次API请求直接把账户的并发配额打满其它正在跑的任务全部排队阻塞。后面我在框架里加了一个全局令牌桶限流器核心逻辑很简单import time import threading class RateLimiter: def __init__(self, max_tokens, refill_rate): self.max_tokens max_tokens self.tokens max_tokens self.refill_rate refill_rate self.lock threading.Lock() self.updated_at time.time() def acquire(self, cost1, timeout30): deadline time.time() timeout with self.lock: while time.time() deadline: now time.time() self.tokens min( self.max_tokens, self.tokens (now - self.updated_at) * self.refill_rate ) self.updated_at now if self.tokens cost: self.tokens - cost return True time.sleep(0.05) return False每个任务在发起API调用前都必须先向桶里申请额度拿不到就排队等待。这个设计救了整个系统的稳定性允许任务有弹性但绝不允许多个任务同时打爆上限。你在自己实现时这套令牌桶的代码可以直接按需改一改搬走。6. 从框架到服务把agent-agents打包成可对外交付的代理服务最后聊一下项目落地的形态。如果这套东西只是自己玩其实到这里已经够了。但我的目标是让它变成一个能对外接单赚钱的“AI代理服务”所以还做了几个面向交付的改造这些经验对想把Agent项目产品化的人应该特别有用。6.1 封装成请求入口同步接口与回调回调事件对接单场景来说客户最常用的是同步等待结果。我封装了一个简单HTTP接口用户POST一段任务描述系统在内部跑完整个Agent链路实时返回最终生成的内容。这样客户不需要理解什么是Router、什么是Critic它只看到一个输入框和一份交付结果就好。不过对时长较长的后台任务我加了一个任务ID机制提交任务后立刻返回taskId客户可以拿着这个ID轮询进度接口看到每个子任务的实时雷达状态。这一点很重要因为它把“黑盒跑40秒”变成了“能看见步骤的透明流水线”。在一个什么都要追求可解释性的时代哪怕只是看见“正在审核”四个字客户的耐心都会成倍提升。6.2 成本控制按任务类型区分模型级别主动控制成本是非常重要的产品化能力。我给三类任务的默认模型做了分级任务类型推荐模型档次理由写营销文案、头脑风暴中等能力大模型 高温度追求创意不追求极端事实精度信息抽取、格式转换小快速套装模型高确定性低延迟成本低复杂推理、逻辑判断强推理大模型 低温度一步错步步错值得用好模型这一分级让我单次任务的API成本下降了约40%而交付质量几乎没有变化。原因是很多子任务并不需要顶级模型的智力它们只是按模板做格式化小模型更快更便宜。很多个人开发者会在这个地方无谓烧钱——什么都往最大的模型上怼交付没变好利润倒是少了。6.3 交付模板与客户期望管理我在最后交付层做了一层“精修包装”这不是画蛇添足而是接单业务里非常关键的一环。系统输出的内容会经过一个最后整理Agent把格式统一、把占位符高亮标出、并在文末自动生成“交付说明”里面写着哪些信息是客户需要补充的、哪些内容是系统假设的。它的存在把很多模糊的期望直接摊在了桌面上客户不会觉得你偷工减料反而会觉得这是一套专业的交付体系。我也遇到过客户拿Agent生成的结果直接发布结果出了问题回头找我。现在我在交付说明里会明确写一句“AI生成内容发布前请人工复核尤其是涉及事实和数据的地方”。这不是推卸责任而是真正的敬业——告诉客户工具的边界才是对自己的专业信誉负责。我现在的体会是agent-agents这套思路真正颠覆我工作方式的点不在于它产出的内容有多惊艳而在于它把混乱多变的“接单干活”流程变成了一个每个环节都可观测、可干预、可复盘的流水线。如果你也想自己搭一套类似的东西别急着抄代码先花一个晚上把你的交付流程拆成角色、路由、质检三层肚子里的结构理顺了代码层面的事反而都是水到渠成。这个框架本身还会继续迭代我接下来打算给它加上更细粒度的反馈学习和按项目维度沉淀知识记忆一旦跑通了再回来写篇续篇。