ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

多Agent协作失控?用Agency模式构建有序调度机制

2026/8/30 22:48:16 拓冰建站 浏览量
多Agent协作失控?用Agency模式构建有序调度机制 如果你最近正在折腾 AI Agent大概率遇到过这个场景单个 Agent 回答问题时非常聪明能写代码、能总结文档、能调用工具可一旦让它和另一个 Agent 协作结果就开始失控——任务重复、上下文混乱、互相覆盖文件、甚至两个 Agent 在同一份配置里反复改来改去。这个问题不是模型能力不够而是组织方式出了问题。Agent 应用进入真实工程之后瓶颈往往不是“某个 Agent 聪明不聪明”而是“一群 Agent 怎么被有序地管理”。Agency-Agents 这个项目核心就是在回应这个问题。本文不打算只罗列 GitHub 仓库的功能清单而是从工程视角拆解这类项目背后的关键设计什么是 Agency 模式为什么多 Agent 协作需要这一层“调度机构”以及如何用一套可落地的最小方案把多个 Agent 跑起来。读完你至少能回答三件事这类项目适合解决什么问题、不适合解决什么问题、以及在自己的代码里应该从哪里开始搭建第一个多 Agent 协作流程。1. 这篇文章真正要解决的问题先给一个判断当项目里 Agent 数量超过两个你真正要管理的就不是 Prompt而是任务流转、上下文边界和工具权限。很多人是从“写一个 Agent 调用大模型 API”开始接触这个领域的。最初的代码往往很简单构造 system prompt拼接 user message把返回结果打印出来。这个阶段不存在“编排”因为只有一个 Agent它自己就能完成从理解到输出的全过程。但真实业务不会只有一个 Agent。比如一个技术写作任务可能需要一个 Agent 负责资料收集一个 Agent 负责提炼要点一个 Agent 负责写初稿一个 Agent 负责校对。如果四个 Agent 各自拿着完整上下文去跑很快你就会发现同一个资料被四个 Agent 分别读了一遍Token 消耗翻了好几倍写稿 Agent 和校对 Agent 对同一段事实的理解不一致某个 Agent 把任务结果写进了一个共享目录另一个 Agent 根本没看到一个 Agent 陷入死循环不断调用工具费用快速上涨。这些问题的共同根源是缺少一个“中间层”来接收任务、拆分任务、调度执行、汇总结果。Agency-Agents 这类项目试图提供的正是这个中间层——我把这个模式称为“代理机构模式”。它不是让一个 Agent 去扮演其他 Agent而是让一个总控 Agent 像团队负责人一样把任务分给若干个专业性更强的子 Agent再对结果做统一收口。这篇文章适合的读者是已经写过一个简单 Agent知道如何调用模型接口但不确定多 Agent 该怎么设计或者正在做一个内部工具需要让多个 Agent 协作完成报告生成、代码审查、测试用例生成等任务。如果你还没写过任何 Agent建议先跑通一个单 Agent 的调用再回来看这篇。2. Agency-Agents 的核心概念与适用场景Agency-Agents 从项目名字就能看出两层含义Agency 是“代理机构”Agents 是“智能体”。放在一起它表达的是一种协作关系多个智能体在一个统一调度实体下协作完成任务。这里要澄清一个常见误解很多人以为多 Agent 就是把多个大模型调用写在一个循环里。不是的。Agency 模式强调的是“机构”的概念——机构本身不负责具体业务它负责分配资源、定义流程、监督结果。对应到技术设计上就是引入一个 Orcherstrator调度器或者 Coordinator协调器。2.1 Agent、Tool、Agency 三者的边界先从基础概念说起Agent智能体一个能感知输入、做出决策、执行动作并返回结果的程序单元。在 AI 场景里它通常由大模型驱动但也可以包含规则引擎或普通代码。Tool工具Agent 可以调用的外部能力比如搜索引擎、数据库查询、文件写入、命令行执行。工具让 Agent 从“只会说话”变成“可以做事”。Agency代理机构一个管理多个 Agent 的协调层。它接收用户任务决定任务应该由哪个 Agent 执行处理执行顺序汇总各 Agent 的输出。用一个类比来理解Agent 是员工Tool 是员工能使用的办公系统和设备Agency 是部门负责人。你可以写一个极其聪明的员工但如果没有人分配任务、协调资源、检查结果整个部门还是乱成一团。2.2 单体 Agent、多 Agent、Agency 模式的区别很多项目代码里确实有多个 Agent但它们之间往往是“链式调用”A 的输出作为 B 的输入B 的输出作为 C 的输入。这种方式适合任务流程固定的场景但不适合任务需要动态判断的场景。下表对比了三种常见形态形态任务分配方式优点缺点典型场景单体 Agent全部任务由同一个 Agent 完成实现简单上下文连贯Token 消耗高职责不清难扩展简单问答、单步骤生成链式多 Agent固定顺序A 完成后 B 再执行流程可控逻辑清晰无法动态调整一个环节失败会中断整个链路固定流水线任务Agency 模式调度器动态拆分和分配任务灵活、可扩展、职责清晰需要额外的编排和治理成本复杂任务、多工具协作、团队协作自动化Agency-Agents 更适合作为第三种形态的参考实现。从项目结构来看它的重点不是“如何写一个更强的 Agent”而是“如何把多个 Agent 组织成一个可管理的团队”。这意味着你需要关注的是接口定义、任务队列、状态管理、审计日志而不是反复调 Prompt。2.3 适用场景与不适合的场景从社区里这类项目的用法来看比较适合的场景包括研究报告自动生成一个 Agent 查资料一个 Agent 列大纲一个 Agent 写正文一个 Agent 核对引用。代码审查流水线一个 Agent 扫描代码规范一个 Agent 分析潜在 Bug一个 Agent 汇总审查意见。客服工单分类与回复一个 Agent 先判断工单类型再分给不同的处理 Agent最后由一个 Agent 审核回复内容。数据分析自动化一个 Agent 写 SQL一个 Agent 执行并校验数据一个 Agent 生成分析结论。不适合的场景也更值得注意如果任务流程固定、步骤少链式多 Agent 更简单没必要引入 Agency 层。如果对延迟要求极高多 Agent 的多次模型调用会显著增加耗时单体 Agent 可能更合适。如果团队没有做好日志和监控多 Agent 的排错成本会很高反而比单体方案更痛苦。核心原则是Agency 解决的是复杂度和可扩展性问题而不是单次响应质量。如果你的问题一个 Agent 就能解决就不要硬拆成多个。3. 环境准备与前置条件这一节给出一个通用的环境准备清单。需要说明的是Agency-Agents 这类开源项目在不同阶段的依赖变化较快具体版本请以实际克隆的仓库为准。本文的重点是演示一套通用的搭建思路而不是绑定某个固定版本。3.1 运行环境建议准备以下环境操作系统Linux 或 macOS 优先Windows 也能跑但在环境变量和命令执行上需要多花一点时间。编程语言Python 3.10 或以上因为新版 Agent 项目普遍使用类型注解和异步语法。Node.js 18 或以上如果项目是 TypeScript 实现则需要用到。包管理工具pip、venv 或 poetryPythonnpm 或 pnpmNode。Git用于克隆仓库和切换版本。如果你只是验证思路不一定要先克隆项目。你可以先创建一个独立的 Python 虚拟环境把 Agent 依赖装进去后面再决定是否引入开源框架。3.2 模型服务与 API Key多 Agent 协作的底层依然是模型调用。常见的模型服务有三种OpenAI 兼容接口包括 OpenAI 官方 API以及大量兼容 OpenAI 接口的国内外模型服务。Anthropic 接口部分项目原生支持 Claude 系列模型。本地模型服务通过 Ollama、vLLM 等工具在本地拉起模型适合对数据隐私要求较高的场景。准备 API Key 时要注意最小权限原则。如果模型服务支持创建多个 Key建议为不同环境创建不同 Key并设置调用额度。不要在一个共享文档里放生产环境的 Key更不要用export OPENAI_API_KEY后随手粘贴进聊天工具。3.3 依赖安装示例如果你选择 Python 作为实现语言一个最小依赖集合大致如下# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下为 .venv\Scripts\activate # 安装依赖 pip install openai python-dotenvopenai用来调用模型接口python-dotenv用来加载.env文件避免把 Key 硬编码到代码里。创建.env文件# .env OPENAI_API_KEYsk-your-key-here AGENCY_MODELgpt-4o-mini WORKER_MODELgpt-4o-mini这里把调度模型和工作模型分开了。实际项目中调度器可以使用能力较强的模型因为拆分任务、理解全局上下文更复杂具体的执行 Worker 可以使用速度更快的模型以便控制成本和延迟。4. 核心流程拆解从任务到结果的完整链路搞清楚了概念和环境接下来需要理解一个 Agency 模式的完整执行链路。无论你用什么框架核心流程都不会差太多。4.1 接收任务与目标解析整个流程的第一步是调度器拿到用户的原始任务。这个阶段不要急着把任务丢给子 Agent而是要做两件事一是确认任务边界二是判断任务是否可以拆分。例如用户输入“帮我写一份关于微服务监控方案的技术文档”调度器需要识别出这个任务包含资料收集、方案整理、文档撰写、格式校对四个子任务。如果调度器直接把整段文字丢给“写手 Agent”那么“写手 Agent”可能没有资料收集能力生成的内容就会泛泛而谈。这个阶段的关键是设计一个“任务解析”模块它可以是纯 Prompt 提示调度器输出 JSON也可以是一段规则代码。社区项目里更推荐前者因为大模型对语义拆解更高效。4.2 子任务分配与上下文管理拆分完成后调度器需要决定每个子任务交给哪个 Agent并控制每个 Agent 能看到什么内容。这里最容易犯的错误是把所有历史上下文全部传给每个 Agent。正确的做法是给每个 Agent 只传它需要的上下文。比如“资料收集 Agent”只需要知道主题和关键词“文档撰写 Agent”只需要拿到资料收集的结果和写作规范“格式校对 Agent”只需要拿到正文和校对规则。上下文隔离是多 Agent 工程最重要的设计之一。它既控制 Token 成本也避免不同 Agent 之间互相干扰。4.3 执行、反馈与重试子 Agent 执行时可能有三种结果成功、失败、超时。调度器应该为每种情况定义行为成功将结果写入任务上下文进入下一步。失败如果是临时性错误比如网络超时可以重试一次如果是任务本身不可执行应该把错误信息返回给调度器由调度器决定是否重新拆分任务。超时需要设置最大执行时间或最大步骤数防止 Agent 死循环。从工程角度看执行反馈是决定项目是否可信的关键。如果调度器只是“把任务丢出去然后等结果”那和链式调用没有区别。真正的 Agency 模式要求调度器能根据反馈调整计划这也是引入这一层带来的核心价值。4.4 结果汇总与交付所有子任务完成后调度器需要把结果汇总成一个交付物。汇总不是简单拼接而是要做一致性检查各子任务之间的结论是否冲突、信息是否有遗漏、格式是否符合预期。如果是生成文档类任务调度器还可以调用一个“审查 Agent”做最终校验。这一步在实践中能显著提升输出质量代价是增加一次模型调用。是否使用需要根据任务重要性和成本预算权衡。5. 完整示例代码实现跑通一个最小 Agency下面我用一个最小可运行的 Python 示例演示 Agency 模式的核心链路。这个示例不依赖复杂框架只使用 OpenAI 的 Chat Completions 接口通过“调度器 两个 Worker”的结构完成任务。5.1 示例目标我们要完成的任务是让一个“研究员 Agent”整理某个技术主题的关键信息再让一个“写作者 Agent”基于整理结果输出一篇简短技术文档。调度器负责接收任务、按顺序调用两个 Worker、汇总结果。# demo_agency.py import os from openai import OpenAI client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) ROLE_TEMPLATE 你是{role}。请只处理分配给自己的任务不要尝试扮演其他角色。 def call_model(role: str, task: str, model: str | None None) - str: 调用大模型返回文本结果。 model model or os.getenv(WORKER_MODEL, gpt-4o-mini) response client.chat.completions.create( modelmodel, messages[ {role: system, content: ROLE_TEMPLATE.format(rolerole)}, {role: user, content: task}, ], temperature0.2, ) return response.choices[0].message.content def researcher(topic: str) - str: Worker 1负责收集关键事实和风险点。 return call_model( 研究员, f请整理与「{topic}」相关的关键事实、实现难度和已知风险控制在500字以内。, ) def writer(topic: str, material: str) - str: Worker 2负责基于调研材料撰写文档。 return call_model( 写作者, f请基于以下材料围绕「{topic}」写一篇200字左右的技术摘要\n\n{material}, ) def agency(topic: str) - dict: 调度器按顺序调用两个 Worker并汇总结果。 print(f[Agency] 开始处理任务{topic}) print([Agency] 任务已分配给研究员 Agent...) material researcher(topic) print([Agency] 研究员 Agent 已完成正在分配给写作者 Agent...) document writer(topic, material) print([Agency] 全部子任务已完成正在汇总结果。) return {topic: topic, material: material, document: document} if __name__ __main__: result agency(用Python实现一个多Agent调度器) print(\n 调研结果 ) print(result[material]) print(\n 文档结果 ) print(result[document])这个示例虽然只有几十行但它已经包含了一个最小 Agency 的核心要素角色隔离每个 Worker 只被设置为一个专业角色职责单一调度顺序agency函数决定了任务的执行顺序先调研后写作参数传递writer接收material作为上下文但没有接收整个历史对话结果汇总统一返回一个字典结构方便后续做校验或持久化。5.2 运行与验证在运行前确保你已经配置好.env。然后在同一个终端里执行# 加载 .env 中的环境变量 set -a source .env set a # 运行脚本 python demo_agency.pyWindows 用户可以在 PowerShell 中执行Get-Content .env | ForEach-Object { if ($_ -match ^(.*?)(.*)$) { [Environment]::SetEnvironmentVariable($matches[1], $matches[2]) } } python demo_agency.py预期结果是脚本依次输出“研究员”的调研结果和“写作者”的文档结果。如果模型调用成功你应该在终端看到类似的流程日志[Agency] 开始处理任务用Python实现一个多Agent调度器 [Agency] 任务已分配给研究员 Agent... [Agency] 研究员 Agent 已完成正在分配给写作者 Agent... [Agency] 全部子任务已完成正在汇总结果。出现这四行日志说明最小 Agency 已经跑通。5.3 扩展为异步并发上面的示例是串行执行。如果子任务彼此独立比如“资料收集 Agent”和“代码扫描 Agent”可以同时跑那么可以改为异步并发显著降低整体延迟。下面是一个用asyncio并发执行两个独立 Worker 的示例# demo_agency_async.py import asyncio import os from openai import AsyncOpenAI client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) async def call_model(role: str, task: str) - str: response await client.chat.completions.create( modelos.getenv(WORKER_MODEL, gpt-4o-mini), messages[ {role: system, content: f你是{role}。}, {role: user, content: task}, ], ) return response.choices[0].message.content async def collect_material(topic: str) - str: return await call_model(研究员, f围绕「{topic}」收集技术要点。) async def scan_risks(topic: str) - str: return await call_model(风险分析师, f围绕「{topic}」列出实施风险。) async def main() - None: topic 微服务监控方案 material, risks await asyncio.gather( collect_material(topic), scan_risks(topic), ) print(调研材料, material) print(风险清单, risks) if __name__ __main__: asyncio.run(main())异步模式在多 Agent 场景里非常实用但它会带来一个新的问题并发执行时多个 Worker 可能同时写入同一个共享状态需要引入锁或独立目录。这也是为什么真实项目不会只用一段脚本而会引入任务队列和状态存储。6. 运行结果与效果如何验证6.1 判断成功的标准一个简单的脚本跑通并不代表一个多 Agent 项目已经成功。从工程角度看需要从四个维度验证效果功能正确性最终交付物是否满足任务要求过程可观察性能否从日志中看到每个 Agent 的输入、输出和耗时资源可控性每次任务消耗多少 Token、多少成本是否存在失控调用失败可恢复性某个子 Agent 失败时整个任务是否会卡死如果你只是运行了第 5 节的示例那么你只能验证第一项。要验证后三项需要补齐日志、计数和异常处理。6.2 给脚本加一个最小审计日志一个很实用的做法是在每个 Worker 调用前后记录执行时间。使用 Python 的time模块即可import time def timed_researcher(topic: str) - str: start time.monotonic() result researcher(topic) cost_ms (time.monotonic() - start) * 1000 print(f[Audit] researcher 耗时 {cost_ms:.2f} ms) return result这看起来简单但在排查“为什么多 Agent 任务很慢”时非常关键。没有耗时统计你很难判断瓶颈是在模型调用、工具执行还是任务排队。6.3 失败后的第一步排查顺序如果脚本运行失败建议按顺序检查环境变量是否加载成功。.env文件是否存在key 是否被正确读取。打印os.getenv(OPENAI_API_KEY)的前几位可以确认。模型名称是否可用。不同模型服务支持的模型 ID 不一样报错信息通常会提示。网络是否能连通模型服务。代理、防火墙、DNS 都会导致超时需要先排除基础网络问题。角色 Prompt 是否导致模型输出异常。如果返回内容为空或不符预期可以先用普通对话接口测试同一模型。从我的经验看90% 的初次运行失败都来自前两项而不是编排逻辑本身。不要看到openai报错就立刻怀疑大模型先确认请求有没有真正发出去。7. 常见问题与排查思路多 Agent 项目一旦开始接入真实工具和多人协作问题会变得非常具体。下表整理了我在实际项目中遇到过的高频问题。问题现象可能原因排查方式解决方案多个 Agent 同时修改同一文件内容互相覆盖缺少共享状态锁或写入队列查看文件修改时间检查任务日志中是否有并发写入给写操作加锁或者让每个 Agent 写入独立目录一个 Agent 陷入无限循环费用持续上涨缺少最大步骤数或最大轮次限制查看模型调用日志检查循环条件是否恒为真在调度器中增加max_steps参数超限强制终止上下文长度超出模型限制子任务结果未经压缩就传给下一个 Agent查看传给模型的 messages 长度在汇总层增加摘要模块或者按关键字段截断多个 Agent 对同一事实输出矛盾每个 Agent 使用不同版本的知识或上下文不一致检查各 Agent 的 system prompt 和输入上下文统一事实来源让后续 Worker 只基于前序结果输出某个 Agent 调用工具时权限过大共享同一个 API Key 或工具白名单缺失检查工具执行记录确认是谁调用了高危操作为每个 Agent 配置独立的最小权限工具执行前增加审批代码部署后无法复现本地结果模型版本或参数不一致对比本地和生产环境的环境变量、模型 ID、随机参数固定模型版本把所有参数纳入配置管理一个子任务失败导致整个任务中断没有异常捕获和降级策略查看异常堆栈是否来自单一 Worker在调度器中增加失败重试和降级分支返回部分结果以上问题不是“可能遇到”而是在真实项目里几乎一定会遇到。建议在开发多 Agent 项目时先把这些问题对应的基础设施搭好再往上加业务逻辑。8. 最佳实践与工程建议8.1 每个 Agent 只做一件事Agent 的 Prompt 写得再复杂也不要让一个 Agent 承担两种完全不同的职责。比如“研究员 Agent”只负责收集资料“写作者 Agent”只负责输出文档。这样做的最大好处是当某个任务质量出现问题时你能快速定位是哪一个 Agent 出了问题而不是在长 Prompt 里反复找原因。8.2 调度器不做业务只做编排调度器的职责是拆任务、派任务、汇总结果。它不需要理解“微服务监控方案”的具体细节也不需要知道怎么写代码。一个好的架构是调度器输出的是结构化任务清单Worker 执行的是业务动作。8.3 用结构化协议定义 Agent 输入输出如果 Agent 之间通过 JSON 传递数据建议使用 Pydantic 或 TypeScript 的类型定义来约束结构。这样在调度器解析结果时不需要写一堆if result[xxx]的防御代码。from typing import TypedDict class ResearchResult(TypedDict): topic: str key_points: list[str] risks: list[str] confidence: float定义好协议后即使底层模型输出偶尔不标准你也可以在解析层做校验和修复而不是把脏数据一路传递下去。8.4 每次任务都带 Trace ID多 Agent 项目排错难是因为一次完整任务会横跨多个模型调用、多个工具执行。如果没有统一的trace_id你很难把一次用户请求对应的所有日志串起来。import uuid TRACE_ID str(uuid.uuid4()) def log(agent_name: str, message: str) - None: print(f[{TRACE_ID}] [{agent_name}] {message})生产环境建议把日志输出到结构化日志系统用trace_id作为索引字段。8.5 工具调用必须是白名单Agent 能调用的工具越多出事的概率越大。不要给所有 Agent 一个“万能执行器”。每个 Agent 只能获取它完成任务所需的最小工具集合。比如“资料收集 Agent”可以访问搜索 API但不能访问删除文件的 Shell 命令“测试生成 Agent”可以写测试文件但不能修改生产配置。如果 Agent 执行的是高危操作应该增加人工审批环节。宁可损失一点自动化程度也不要让你的 Agent 拥有过度的系统权限。8.6 上线前先跑成本测试多 Agent 意味着多次模型调用。一次任务如果是 5 个子任务那么 Token 消耗可能是一个单 Agent 任务的 5 到 10 倍。上线前一定要用小规模测试集跑一遍统计平均成本、平均延迟、平均失败率。不要等月底账单出来才意识到问题。可以从三个维度控制成本选择更便宜的模型承载高并发、低难度 Worker尽量复用中间结果避免同一数据被多个 Agent 重复获取对重试次数做硬限制防止失败任务反复触发调用。8.7 先跑单体再拆多体最后一条建议也是最重要的一条不要一开始就设计六个 Agent。先用一个单体 Agent 把任务跑通观察输出质量、成本和延迟。当数据证明“任务确实需要拆开”时再把环节逐步拆成独立的 Agent。很多时候单 Agent 加上一个工具集就够了。过早引入多 Agent只会让调试复杂度急剧上升。Agency-Agents 这类项目是有价值的但它的价值发挥在合适复杂度之上。9. 总结与后续学习方向Agency-Agents 这个项目给开发者的启发不在于“多了一个 Agent 框架”而在于它把一个问题摆到了台面上Agent 多了之后编排和治理比模型能力更关键。本文从多 Agent 协作的痛点出发介绍了 Agent、Tool、Agency 三者的边界梳理了 Agency 模式的核心流程并用一段 Python 示例跑通了一个最小调度链路。如果你现在正准备在公司内部做一个多 Agent 应用建议先按第 8 节的建议把日志、成本控制、权限白名单和数据协议四件事做好再开始堆业务功能。下一步可以沿着三个方向继续深入尝试引入真正的任务队列中间件让调度器具备并发、优先级和持久化能力把工具调用从简单的文本函数升级为带权限校验的远程服务研究多 Agent 的评测方法建立一套包含正确率、成本、延迟和失败率的可量化指标。一个多 Agent 项目能不能从 Demo 走向生产最终看的不是 Agent 的 Prompt 有多花哨而是你愿不愿意在它周围建立真正的工程护栏。建议先把这个最小例子复制到本地跑通然后给它加上一个工具、一个并发分支、一条审计日志。你会发现真正的挑战从现在才开始。