ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

多智能体协作编排框架实战:从角色分工到任务调度

多智能体协作编排框架实战:从角色分工到任务调度 第一次看到 agency-agents 这个名字你可能以为它跟网络代理有关但这里的 agent 其实是 AI 智能体。这是我自己的一个多智能体协作编排项目把多个具备不同能力的智能体组织起来像一家小公司一样分工干活。有人负责拆解目标有人负责搜集资料有人负责写初稿还有人专门挑毛病。整个过程不需要人手动转发消息全部交给调度器自动完成。如果你正被单个 Agent 搞不定的复杂任务折磨又不想人工把需求拆成几十段提示词那这个项目的思路也许正好对你有用。下面我会把设计逻辑、角色定义、调度实现和踩坑记录都拆开讲尽量让你也能照着搭一套自己的智能体团队。1. 项目概述1.1 这个项目到底是什么agency-agents 本质上是一个“多智能体运行框架”。它不关心你用的是哪个模型服务也不绑定某个特定API只提供一个组织智能体协作的壳定义了一组标准数据结构比如任务、消息、智能体注册表、共享记忆实现了一个调度循环负责把任务派发给合适的智能体收集输出校验结果再决定下一步动作最后还封装了工具调用、重试、超时、预算控制这些让智能体真正能干活的关键细节。为什么叫“agency”因为我发现单个大模型智能体很像一个能力不错但精力有限的实习生。你让它写一段代码它能写你让它做一个需要查资料、分析、验证、出报告的全流程任务它很容易写到一半就跑偏。与其逼着一个人扛所有事不如组一支小队让每个人只负责自己最擅长的环节。small team 的协作效率通常远高于一个超长提示词驱动的“全能选手”。agency-agents 就是把这种团队协作模式变成了一套可复用、可配置、可观测的程序化框架。1.2 它能解决我遇到的什么具体问题最早我遇到一个很典型的场景需要快速输出一份行业调研报告。用单个 Agent 直接干第一步让它搜索资料它会给我一堆链接和摘要第二步让它分析它容易漏掉关键信息第三步让它写报告它经常自己脑补数据。最难受的是整个过程在一个上下文里滚动前面的低级错误会一路污染到最后。用 agency-agents 之后我把流程拆成了四个角色资料搜集员、分析提炼员、报告撰写员、质量审校员。搜集员把原始材料汇总到共享区分析员只基于这些材料做结构化梳理报告员拿到的是干净的分析结果审校员最后检查事实和格式。每个角色都是一个独立的 Agent每次调用只处理自己需要的信息上下文干净、职责清晰、错误边界清楚。实际跑下来一份几千字的初稿从过去的三四个小时压缩到不到半小时质量还平稳很多。1.3 适合谁来参考如果你是有一定编程能力、能调用 LLM API 的开发者想自己搭一套多智能体系统那这篇文章基本就是按你的视角写的。如果你刚接触 Agent 概念也问题不大我会先把最简单的模型讲清楚再给你可以直接抄的配置和伪代码。如果你是产品经理想判断多智能体方案能不能落地也可以重点看第二、三章了解这种模式的能力边界和成本隐患。2. 整体设计与思路拆解2.1 为什么我坚持用多智能体而不是一个超级智能体很多人第一反应是把任务写详细点给一个 Agent 配上所有工具不也能完成吗我试过而且试过不止一次。问题在于单 Agent 处理复杂任务会同时遇到三个坎上下文长度有限、指令重心漂移、工具选择困难症。上下文长度有限意味着你没法把几十个网页原文都塞进去指令重心漂移是指提示词越长模型越容易只关注开头和结尾中段的重要约束经常被忽略工具选择困难症更明显当你给单个 Agent 挂了搜索、代码执行、文件读写、数据库查询等二十个工具它每次行动都会纠结该用哪个偶尔还会用错工具。多智能体相当于把“一个全才的复杂决策”拆成“多个专才的简单决策”。每个 Agent 只需要在自己的角色范围内做相对简单的判断成功率自然高很多。我整理过一个很直观的对比摘在下面。它不是严谨评测但足够说明问题。维度单智能体多智能体协作上下文压力随着任务推进持续膨胀每个智能体只看到自己需要的部分职责边界依赖长提示词约束容易漂移靠角色定义隔离边界清晰工具管理工具数量多易选错按角色绑定少量工具决策简单故障隔离一步出错带偏全局单点失败可以重试或替换角色成本调用次数相对少但单次输入长调用次数变多单次输入短开发调试改提示词是黑盒调优每个环节可单独看日志和结果当然多智能体也有代价开发更复杂调用成本可能上升链路一长更容易出现死循环或消息风暴。所以这个方案不是银弹它最适合的是“流程稳定、环节清晰、每个子任务都有明确产出”的场景。2.2 角色定义与职责边界怎么划agency-agents 里每个智能体都对应一张“角色卡片”。卡片上写清楚四件事你是谁、你负责什么、你可以用哪些工具、你最终要交什么格式的结果。不要小看这四件事绝大多数多智能体项目跑飞都是因为角色边界没划清。举个例子我一开始把“资料搜集员”定义为“查找与主题相关的信息”。它确实去找了但经常把搜索结果直接原样返回里面还夹着一堆广告和冗余文字。问题出在我没告诉它“该返回什么”。后来我规定它必须返回结构化条目每个条目包含标题、来源、核心观点、可信度评分并且只保留与目标强相关的内容。输出结构一明确后面的分析员终于不用从垃圾材料里捞金子了。职责边界背后其实是一种约束力。模型天然有“多管闲事”的倾向给它的角色描述越模糊它越会替下游决策。所以我在设计角色卡片时特别强调“不要做下一步的工作你只要完成本环节”。这句话听着笨但在实际运行里能明显减少越权行为。2.3 协作拓扑为什么采用中心化调度而不是让智能体自由对话多智能体协作有两种常见拓扑一种是完全去中心化所有智能体可以互相发消息谁都能跟谁对话另一种是中心化调度所有消息都经过一个调度器由它决定下一步派给谁。agency-agents 选择的是典型的“中心化调度去中心化执行”。去中心化模式很灵活像是开一个全员微信群谁有问题就 别人。问题在于群里消息一多没人能掌握全局而且容易陷入无意义的来回讨论。遇到过吗两个 Agent 互相否定对方的观点绕了八个回合还在第一层问题上打转。这种情况在去中心化模式里特别难控制你甚至说不清是谁先开始的。中心化调度就不一样。调度器像一个项目经理所有任务都进队列所有结果都回报到它这里。它不做具体工作只负责派单、检查进度、处理失败。这样整个系统有明确的信息流发起任务 - 调度器 - 执行智能体 - 结果回调度器 - 再派下一单。日志里按照 task_id 全链路追踪问题定位快得多。代价是调度器本身成了瓶颈所以我把它设计得尽量轻只做状态流转不承载领域逻辑。2.4 消息与任务协议是协作的“共同语言”智能体之间的协作最怕信息格式不统一。A 返回一段 MarkdownB 返回一段 JSONC 直接甩了个文件路径下游处理起来会非常痛苦。agency-agents 设计了一套统一的数据单元核心是 Task 和 Message。Task 是调度的基本单位。它包含几个关键字段task_id 唯一标识、type 表示任务类型、payload 放输入数据、expected_output 描述期望输出格式、assignee 指定负责的智能体、priority 表示优先级、timeout 设置超时时间。这套字段看起来简单但它让调度器不需要理解任务内容只需要按字段路由。Message 是执行结果。它包含 sender、receiver、task_id、content_type、content、metadata。其中 metadata 很关键可以放模型名、token 消耗、工具调用记录等辅助信息。代码实现里所有智能体的输入输出都必须转换成 Message 格式。这样做的好处是新人接手项目时不用理解每个智能体内部的提示词只要学会看 Message 就能追踪业务链路。我觉得这套协议有点像快递面单面单标准化了物流系统才能高效运转。3. 核心细节解析与实操要点3.1 智能体注册表给每个角色建一张“简历”智能体的定义我全部放在配置文件里不在代码里写死。配置中心的核心是一份“智能体注册表”相当于每个角色的简历。注册表里每个条目包含 name、description、system_prompt、tools、model、temperature、max_retries 等字段。description 是最容易被低估的字段。它是调度器选人的依据好比招聘网站上的自我介绍。你写得越具体调度器越容易把任务派给正确的人。我这里有一个真实教训某次我定义了“代码审查员”和“代码修复员”但两个 description 都写了“负责处理代码相关问题”。结果调度器随机分配审查员去改代码、修复员去提意见输出完全错乱。后来我把 description 写成“专门检查代码中的安全漏洞和性能问题不直接修改代码”“根据审查意见生成修复后的代码”分工一下就清楚了。system_prompt 是角色的核心人格和规则。我会写上角色背景、工作流程、输出要求、禁止事项。“禁止事项”非常管用比如“不要编造不存在的结论”“不要输出与任务无关的内容”。相比正面鼓励模型对明确的负向约束更敏感。一个典型的智能体配置长这样{ name: analyst, description: 负责对资料条目进行结构化分析和提炼输出结论与证据列表。, system_prompt: 你是一名资深分析员。你只能基于提供的资料分析不得补充外部信息。输出必须包含结论和对应的证据编号。, tools: [extract_claims, read_shared_memory], model: default-model, temperature: 0.2, max_retries: 2 }3.2 工具能力与权限控制白名单和人工审批智能体没有工具就像员工没有电脑只能空谈。但工具给错了比不给还危险。我记得有一次给一个搜索型智能体挂了代码执行工具它搜到一个命令后直接自说自话地执行了虽然只是无害的 echo但也把我吓出一身冷汗。agency-agents 里的工具管理分为两层。第一层是工具白名单也就是每个智能体只能调用自己配置里可见的工具。第二层是操作分级把工具分成只读类、写入类、高风险类。只读类如搜索、读文件执行前无需额外确认写入类如写文件、发消息执行后要记录日志高风险类如删除文件、执行任意 shell 命令、调用付费外部服务必须经过人工审批。我知道很多人觉得“人工审批”太麻烦破坏了自动化。但做项目要分场景。自己本地跑一个小任务全自动没问题一旦接入了真实接口或生产环境没有审批环节就是在赌运气。我在工具接口里留了一个 abstract 方法 confirm_before_execute如果工具定义里 enable_manual_review 为 true调度器就会先把待执行的动作挂起然后通过回调通知我确认。多花两秒点一下能避免一小时的事故恢复。3.3 上下文隔离与共享记忆别把对话历史当公共汽车多智能体最容易忽略的问题就是上下文污染。很多人天真地以为只要把所有 Agent 的输出都塞给下一个 Agent 就万事大吉。结果上下文越来越长模型越来越迷糊还会把不属于当前任务的历史信息当依据。我在设计里做了一个强制规定任何 Agent 都不能直接读取另一个 Agent 的完整对话历史。信息传递必须通过“共享黑板”完成。共享黑板是一个结构化的临时存储区只有被标记为“可共享”的信息才能写进去。写入的条目需要包含四个字段write_author 谁写的、content 内容、visibility 对哪些角色可见、ttl 过期时间。具体执行时资料搜集员写一批资料条目到黑板上设置 visibility 为 [“analyst”]分析员读完并产出结构化结论后这些结论会被写入下一块区域只有报告撰写员能读。这样每个智能体看到的只是与自己相关的最新信息而不是一大堆噪声。如果你已经感受到了上下文开销对成本和准确性造成的双重挤压这一条也许就是最值得抄走的设计。3.4 模型与温度参数不同角色不能用一套参数打天下很多人配置智能体时所有角色都用同一个模型、同一个 temperature0.7。这在多智能体场景里是一种浪费。不同角色的决策性质不一样参数也应该不一样。资料搜集员需要一定探索性temperature 可以设到 0.5 到 0.7让它多找几个角度分析员和审校员需要稳定和严格temperature 我通常设在 0.1 左右负责创意的角色可以更高一点但高到 1.0 以上往往就开始胡说八道了。模型选择上也是同理分析环节可以追求强模型简单格式转换环节用便宜模型就够。这里的核心思想是“成本花在最需要智能的环节”。如果你的任务链路里有十个 Agent 全用最强模型最终成本大概率不可控但如果你全用最弱模型前面链条的错误又会一路放大。我在配置里支持按角色覆盖模型参数注册表里的 model 字段可以直接指定某个模型别名。每个模型别名对应一个实际的模型服务配置包括上下文长度、输入输出 token 上限、单价等。这套机制让我能在不改代码的前提下针对不同任务动态调整成本。4. 实操过程与核心环节实现4.1 最小可运行架构三步搭起骨架如果你要从零开始实现一个类似效果的系统我建议先搭一个“能跑通任何任务”的最小骨架再慢慢填充角色。最小骨架只需要三块智能体配置、交通队列、调度循环。我用 Python 实现时核心调度循环只有几十行。它做的事情很朴素从队列里取一个 Task根据 assignee 找到 Agent让 Agent 执行一轮拿到结果后校验再决定是结束、重试还是生成新 Task。这是我第一次调通整个流程时的伪代码你可以参考着改from collections import deque import json class Orchestrator: def __init__(self, agent_registry): self.registry agent_registry self.queue deque() self.results [] def submit(self, task): self.queue.append(task) def run(self): while self.queue: task self.queue.popleft() agent self.registry.get_agent(task.assignee) try: message agent.execute(task) task.status done task.output message self.results.append(task) except Exception as e: task.retry_count 1 if task.retry_count agent.max_retries: task.status failed self.handle_failure(task, e) else: self.queue.append(task) return self.results你看到这里没有复杂的路由规则因为第一步就是把链路走通。链路通了后面再往队列里加条件分支。我见过太多人一上来就设计几十个类、上千行代码结果第一次运行连“把一句话传给一个角色”都没成功。最小骨架的价值是让你把模型调用、工具解析、结果返回这个基本闭环跑顺畅。4.2 配置一个调研报告型智能体团队光看调度循环不够我们配一个实际可运行的团队。这个团队的流程是搜集资料 - 提炼观点 - 撰写报告 - 质量审校。我通常在配置文件里按顺序定义四个角色然后由调度器按依赖关系依次触发。配置片段如下{ pipeline: research_report, steps: [ { task_type: collect, assignee: collector, next: analyze }, { task_type: analyze, assignee: analyst, next: write }, { task_type: write, assignee: writer, next: review }, { task_type: review, assignee: reviewer, next: finish } ], agents: { collector: { tools: [web_search, fetch_url], temperature: 0.6, output_structure: list[source_title, source_url, key_points, relevance_score] }, analyst: { tools: [read_shared_memory], temperature: 0.2, output_structure: list[conclusion, evidence_ids] }, writer: { tools: [read_shared_memory], temperature: 0.4, output_structure: markdown_report }, reviewer: { tools: [read_shared_memory, check_fact], temperature: 0.1, output_structure: review_comments, verdict } } }实际跑这个流程时调度器会在每个步骤完成后把上一步的输出写到共享黑板然后创建下一步的 Task。整体运行过程中我不需要干预任何一环只需要在最后看一眼审校报告。如果你按这套配置走大概率会遇到一个坑搜索工具返回的内容格式不稳定。我建议在 collector 环节加一个“内容清洗”的小步骤让模型把搜索原文转换成统一格式后再写入黑板这会省掉后面所有环节的大量麻烦。4.3 调度循环与状态机让流程可观测、可重试任务一多调度循环就不能只是一个 while 队列了。我在项目里给每个 Task 引入了明确的状态pending、running、waiting、done、failed。pending 是刚创建还没派发running 是某个 Agent 正在处理waiting 是任务因为需要人工确认或依赖外部输入而暂停done 和 failed 是终态。状态流转的规则很直接。调度器从队列取出 pending 任务时将其置为 running并发给对应的 Agent。Agent 内部如果遇到工具需要人工审批任务会变成 waiting。审批完成后重新回到 pending等待调度器再次派发。重试逻辑使用指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒最多不超过配置的 max_retries。这个机制能有效避免模型服务临时限流导致的“重试风暴”。我还给每个 Agent 的执行过程加了一个“拆分子任务”的能力。当一个任务太大比如“分析这 50 份资料”Agent 可以返回一个专门的 continue 信号里面携带拆好的子任务列表。调度器会把这些子任务重新入队。这个能力本来就是多智能体的优势但很多人只在配置里写死了线性流程浪费了这个特点。4.4 成本控制与并发参数别让多智能体变成烧钱怪兽多智能体系统的成本往往比单 Agent 高因为调用次数变多了。如果你不做成本控制一个长任务可能跑到一半预算超了。我在调度器里加了三个控制参数max_steps 限制整个任务链路的最大步骤数max_retries 限制单 Agent 重试次数max_cost 限制总费用。实现 max_cost 的思路很简单每次模型调用结束后从返回值里读取 token 用量乘上模型单价累加到一个计数器。当计数器超过阈值调度器会停止派发新任务并把当前状态保存为 failed_reasonbudget_exceeded。这个字段一定要记录否则排查时根本不知道是预算用尽还是程序出错。还有一个常被忽略的参数是并发数。多智能体流程天然适合并行但并行度过高模型服务的限流和成本都会迅速上升。我在项目里做了一个简单的信号量限制同一时刻最多只能有 3 个 Agent 在运行。单个任务跑得慢一点没关系稳定不报错更重要。并发数、超时时间、重试次数我建议都先设成“保守档”跑稳了再慢慢放宽。参数推荐初始值说明max_steps15防止任务链路无线延伸max_retries2防止单点不断失败max_cost按任务预算设定达到即终止并返回提示concurrency3控制模型服务压力与成本task_timeout60秒超过则转入失败处理5. 常见问题与排查技巧实录5.1 智能体陷入循环同一个任务反复执行这是多智能体系统里最让我头疼的问题没有之一。现象是日志里同一个 Task 被派发了几十次每次 Agent 都返回了内容但内容几乎一样调度器却总认为“还没完成”。我排查后发现根源有两个。第一个是完成条件太宽松只是“返回了内容”就算成功导致 Agent 给一个空泛的结果也能过关第二个是反馈回路太粗糙Agent 每次拿到上一次的输出再原样加工一遍形成原地踏步。我的解决办法是增加“信息增量校验”。每次执行结果的摘要会和上一次执行的结果计算相似度如果相似度高于 0.9就认为这次执行没有带来新信息不再继续重试而是直接判定为任务异常交给人工处理。同时我给每个 Task 设置一个最大执行次数比如 3 次。别相信模型自己会“哪天想通”靠机制兜底才靠谱。5.2 工具调用失败与重试风暴Agent 调用工具失败常见原因有三个模型生成的参数不符合工具 schema目标服务请求超时模型服务本身返回限流错误。我之前踩过一个低级的坑工具解析没做好模型明明想调用 search但在参数里多了一个多出来的字段解析器直接抛异常。异常被上层捕获后又触发重试模型重新生成参数还是多这个字段陷入死循环。后来我把工具调用接到了一个统一解析层。这个解析层会先做 JSON 格式修复和字段匹配只把解析成功的工具调用交给真正执行器。对于解析失败的调用我会记录失败原因并且在下一次系统提示里把失败原因带回去让模型知道要改什么。同时连续失败 3 次后我会触发熔断不再自动重试直接把问题抛给日志和人工确认。这样处理之后重试风暴基本消失了。5.3 上下文污染与幻觉记忆上下文污染的现象很隐蔽。某次我跑一个代际链路比较长的任务最后审校环节的报告里突然出现一条跟主题无关的历史信息。打开日志找了半天发现是中间分析员读取共享黑板时把一个之前项目留下的旧摘要也读进去了。原因是我没有给共享黑板的条目设置 ttl旧数据一直躺着没清理。从那以后我强制所有共享条目必需带过期时间过期条目会被后台任务定时清理。另外我还在读取共享黑板时加了“按任务过滤”每次任务创建时会生成一个 task_context_id写入黑板的条目都会带上这个 ID读取时只返回匹配当前任务上下文的条目。这个方法从机制上避免了跨任务信息串扰。如果你不想重新设计存储系统最简单的做法是在共享黑板表里加两列task_id 和 write_time每次读取都按这两个字段过滤。别嫌土真的有效。5.4 结果质量不稳定加一个审校角色比改提示词更有效多智能体链路里越靠近下游错误越容易被放大。搜集阶段一个小偏差经过分析、撰写最后可能变成一篇失真的报告。我最早的做法是反复优化每个环节的提示词但发现很难穷尽。后来我引入了一个专职审校角色。它不负责改写只负责挑错包括事实一致性、格式完整性、任务相关性、证据缺失这几项。审校结果如果是不通过调度器会产生一个“返修”任务还给相应的撰写或分析角色。这个闭环非常管用。审校角色的 temperature 一定调低优先级也设置高一点。另外对于关键结论我会同时跑两遍让两个不同配置的分析员独立执行再让审校员对两份结果做一致性比对。成本会上升但只建议用在影响较大的节点比如对外发布的报告核心金额或技术方案选型。5.5 排查问题速查表现象常见原因定位方法解决动作任务反复执行完成条件过于宽松查看相似度日志增加信息增量校验与最大执行次数Agent 不调用工具工具描述不清晰查看模型输出原始内容在系统提示中补充工具用途示例工具参数报错模型生成的 JSON 不合法解析统一入口日志增加 JSON 修复层和失败反馈上下文不停膨胀共享黑板写入过多历史检查读写日志按 task_id 过滤并可加 ttl结果偏离任务主题环节太多导致信息衰减检查各环节输出摘要增加审校环节和人工确认节点成本突然飙高未设置 max_cost查看费用计数器设置预算阈值并熔断模型返回限流错误并发度太高查看服务端错误码降低并发数启动指数退避重试下游读不到上一步结果消息格式不统一检查 Message content_type统一使用 Message 协议并加字段校验6. 经验总结与扩展建议6.1 我在实际项目里的几条经验第一配置比代码重要。把智能体角色、工具、参数都放在配置文件里之后改角色行为根本就不需要改代码只要动 JSON再刷新一下注册表就行。这让调试周期从分钟级缩短到秒级。我甚至会把不同任务的智能体团队配置单独存文件切换场景就是切换文件非常方便。第二先跑通一条链路再扩充智能体数量。我第一次用这个框架时一口气定义了十个智能体有产品经理、架构师、开发、测试、文档工程师看着很像那么回事。实际跑起来角色之间互相干扰日志乱成一锅粥。后来我删掉了一半智能体只保留三个核心角色把单条链路跑顺再逐步加回来。智能体从来不是越多越好必要的角色才是最贵的角色。第三日志一定要带 task_id 和 parent_task_id。多智能体系统是个天然的树状结构一个根任务会衍生出很多子孙任务。没有 parent_task_id 的话你根本不知道某个结果是从哪条分支冒出来的。我习惯在每个日志里打印这两个字段再加一个 exec_trace_id这样一条完整链路从根到叶子都看得清清楚楚。6.2 最后分享一个小技巧给关键环节留一个人工确认入口哪怕自动化做得再好我也建议你在任务链路的关键位置保留一个“人工确认”的暂停点。比如对外发送消息、删除数据、确认最终数字这类高风险动作。你不需要每步都盯着但要在异常时候能及时插进去。我的做法是让调度器支持“暂停在等待状态”我可以在命令行或 Web 面板里直接改任务状态。这个设计看起来不起眼但它在真实场景里救过我很多次。如果后续要扩展我大概会往两个方向走一个是给智能体加动态记忆检索让长期经验沉淀下来另一个是把这套编排能力包成 API让外部系统也能提交任务并接收进度回调。这两种扩展都不需要改变核心调度模型只要在现有注册表和消息协议上做加法。希望你也能从这套思路里找到适合自己的切入点而不是把多智能体玩成一场华丽的提示词拼贴。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进