资讯详情

agency-agents多智能体协作框架:角色、消息与调度机制解析

📅 2026/10/24 9:17:53 | 华诺云谱 👁 阅读
agency-agents多智能体协作框架:角色、消息与调度机制解析
1. 从“agency-agents”这个名字说起它到底在解决什么问题第一次看到agency-agents这个项目名很多人会愣一下——agency 和 agents 两个词叠在一起到底指什么我最初的反应是这大概率是一个围绕“代理agent”构建的框架而 agency 强调的是“能动性”或者“代理机构”这层含义。把这两个词放在一起其实透露了项目的核心定位它不是一个单纯的 agent 运行时而是一套让多个 agent 具备自主决策与协作能力的组织化系统。换句话说如果你之前接触过单个 agent 的编排比如让一个大模型调用工具、执行任务那么agency-agents想做的事情更进一层——它试图回答一个问题当你有多个 agent每个都有自己的职责、记忆和工具集时它们之间怎么像一家“代理机构”那样分工、汇报、协作最终完成一个复杂目标这个思路在自动化工作流、多智能体协作、任务编排这些场景里非常关键。我之所以对这个项目感兴趣是因为在实际工作中单 agent 的局限性太明显了。一个 agent 既要理解需求、又要规划步骤、还要调用工具、最后还得校验结果上下文一长就容易“精神分裂”。而agency-agents这种多 agent 组织化的思路本质上是把“一个人干所有事”变成“一个团队各司其职”。这跟现实中公司运作的逻辑是一样的有接需求的、有做规划的、有执行的、有审核的。项目名里的 agency 正是这个意思——它模拟的是一个代理机构的运作方式。这篇文章适合谁看如果你正在做 AI 应用开发、自动化流程设计、或者对多智能体系统感兴趣那agency-agents的设计思路和落地细节会很有参考价值。即使你暂时不打算用它理解它背后的组织模型也能帮你在设计自己的 agent 系统时少走弯路。接下来我会从核心架构、协作机制、实操配置、常见坑这几个角度把这个项目拆开讲透。2. agency-agents 的核心架构角色、消息与调度三层拆解2.1 角色抽象每个 agent 不是一个函数而是一个“岗位”很多多 agent 框架在实现时习惯把 agent 定义成一个带有 system prompt 的函数输入输出都是文本。这种做法的好处是简单但坏处是 agent 之间没有明确的“岗位边界”。agency-agents在这点上做得更彻底它把每个 agent 抽象成一个角色Role角色包含几个关键属性——职责描述、可用工具集、记忆存储、以及汇报关系。我实测下来这种设计最大的好处是可替换性。比如你有一个负责“数据采集”的 agent它的职责描述和工具集是固定的那么当你想换一种采集方式时只需要替换这个角色的实现而不影响其他 agent。这就像公司里换一个岗位的人业务流程不用变。角色抽象还带来一个隐性收益调试变得容易。当系统输出不对时你可以快速定位是哪个角色的职责没履行好而不是在一大坨 prompt 里大海捞针。具体来说一个角色通常包含以下字段基于常见实践补充字段作用实操建议name角色唯一标识用英文短名避免中文和空格description职责说明写清楚“做什么”和“不做什么”tools可调用的工具列表按最小权限原则分配memory记忆存储配置区分短期上下文和长期记忆supervisor上级角色用于汇报和任务分解这里有个容易忽略的点description 里一定要写“不做什么”。我踩过的坑是早期只写了职责结果一个负责“内容生成”的 agent 跑去调用数据查询工具因为它觉得“多做一些总没错”。后来在描述里明确加了“你不负责数据获取如需数据请向数据角色请求”这类越界行为就少了很多。2.2 消息传递为什么不是简单的函数调用agency-agents里 agent 之间的通信不是直接函数调用而是通过消息Message。这个设计选择乍看有点绕但理由很充分函数调用是同步的、强耦合的而消息传递是异步的、松耦合的。在多 agent 场景下你经常需要“发一个请求出去等对方处理完再回来”如果直接用函数调用整个链路就会阻塞。消息机制还带来一个好处可观测性。每条消息都可以被记录、审计、回放。我在调试一个多 agent 流程时就是靠消息日志发现某个 agent 反复发送同样的请求原因是它的记忆里没有正确更新任务状态。如果换成函数调用这种问题很难被发现。消息的结构一般包含发送者、接收者、意图、载荷、以及上下文引用。意图intent这个字段很关键它让接收方知道“这条消息是要我干什么”而不是靠解析自然语言去猜。常见实践是定义一组有限的意图枚举比如REQUEST_DATA、PROVIDE_RESULT、ASK_CLARIFICATION、REPORT_ERROR。这样 agent 的处理逻辑就可以基于意图做分支而不是全靠大模型理解。2.3 调度器谁来决定下一个该谁干活调度器是agency-agents里最容易被低估的部分。很多人以为多 agent 系统就是“让大模型自己决定调用谁”但实际跑起来会发现完全靠模型决策会导致流程不稳定、死循环、或者任务被反复转手。agency-agents的调度器提供了一个混合策略既支持模型驱动的动态调度也支持规则驱动的静态编排。我的经验是核心流程用规则编排边缘决策用模型调度。比如一个“生成报告”的流程步骤是固定的——采集数据、清洗数据、分析数据、撰写报告、审核。这五步用规则串起来稳定可靠。但每一步内部比如“分析数据”时该用哪种分析方法可以交给模型动态决定。这样既保证了流程可控又保留了灵活性。调度器还需要处理一个现实问题超时和重试。某个 agent 卡住了怎么办agency-agents的做法是给每个任务设置超时超时后触发重试或降级。这里有个细节重试时最好换一个 agent 实例而不是让同一个 agent 重试因为如果它第一次失败是因为上下文污染重试大概率还是失败。3. 多 agent 协作的三种模式流水线、辩论与层级3.1 流水线模式适合步骤明确的场景流水线模式是最直观的协作方式agent A 的输出是 agent B 的输入B 的输出给 C依次类推。这种模式在agency-agents里对应的是顺序编排。我把它用在“内容生产”类任务上效果很好——采集 agent 拿到素材整理 agent 做结构化撰写 agent 出初稿审核 agent 做校验。但流水线模式有个致命问题错误会累积。如果采集阶段漏了关键信息后面所有环节都会受影响而且越往后越难发现。我的应对策略是在每个环节加一个轻量校验不追求完美只检查“必要字段是否存在”。比如整理 agent 输出后检查是否包含标题、正文、关键词这三个字段缺任何一个就打回重来。这个校验成本很低但能拦住大部分低级错误。另一个经验是流水线不要超过五步。超过五步后调试难度指数上升而且每一步的延迟会累加。如果确实需要更多步骤考虑拆成两条流水线并行最后合并。3.2 辩论模式让多个 agent 互相挑刺辩论模式是我觉得agency-agents里最有意思的设计。它的思路是对于同一个问题让多个 agent 分别给出方案然后互相评审最后综合出一个更可靠的结论。这在需要高准确率的场景下特别有用比如事实核查、方案评估、风险分析。具体怎么操作我通常会配置三个角色一个“提议者”、一个“质疑者”、一个“裁决者”。提议者给出方案质疑者找出漏洞和反例裁决者根据双方论据做最终判断。实测下来这种模式能显著减少“模型自信地胡说”的情况因为质疑者会强制挑战假设。但辩论模式也有代价成本高、延迟大。三个 agent 各自跑一轮token 消耗是单 agent 的三倍以上。所以我的建议是只在关键决策点用辩论模式日常任务用流水线就够了。另外质疑者的 prompt 很关键不能让它为了反对而反对要明确“只指出有依据的问题没有依据的猜测不要提”。3.3 层级模式模拟真实组织的汇报关系层级模式是agency-agents最贴近“agency”这个词的设计。它模拟的是一个组织架构有一个“主管”agent 负责接收目标、分解任务、分配给“下属”agent下属完成后向主管汇报主管再决定下一步。这种模式适合目标比较模糊、需要动态分解的场景。我试过用层级模式做一个“竞品分析”任务。主管 agent 拿到“分析某类产品的市场情况”这个目标后自己分解成“收集产品信息”“分析功能差异”“总结优劣势”三个子任务分别派给三个下属。下属完成后主管汇总并输出报告。整个过程不需要我手动编排步骤主管 agent 会根据目标动态决定怎么做。层级模式的关键在于主管 agent 的分解能力。如果它分解得太粗下属不知道怎么做分解得太细又失去了灵活性。我的调优方法是给主管 agent 提供一些“分解示例”让它在 few-shot 的引导下学会合理的粒度。另外主管 agent 需要有一个“最大递归深度”限制防止它无限分解下去。4. 落地实操从零搭一个 agency-agents 流程4.1 环境准备与依赖安装假设你已经有了基本的 Python 环境搭建agency-agents流程的第一步是安装依赖。虽然项目本身可能提供了一些封装但核心依赖通常包括一个 agent 运行时库、一个消息队列或事件总线、以及一个用于存储记忆的数据库。我个人的习惯是先用最简配置跑通再逐步加组件。# 创建虚拟环境 python -m venv agency-env source agency-env/bin/activate # 安装核心依赖基于常见实践 pip install agency-agents pip install redis # 用于消息和状态存储这里有个坑不要一上来就用生产级配置。我见过有人直接上分布式消息队列、向量数据库、监控系统结果光环境就搭了两天还没跑通一个 demo。正确的做法是先用内存存储和同步消息跑通流程确认逻辑没问题后再替换组件。4.2 定义你的第一个角色定义角色是agency-agents里最核心的配置工作。我以一个“资料整理”角色为例展示常见的配置结构from agency_agents import Role, Tool class OrganizerRole(Role): name organizer description 你负责将原始资料整理成结构化内容。 你只做整理不负责采集也不负责最终审核。 如果资料不足向 collector 请求补充。 tools [ Tool(nameformat_text, funcformat_text), Tool(nameextract_keywords, funcextract_keywords), ] memory_config { short_term: {max_tokens: 2000}, long_term: {enabled: False} } supervisor manager注意description里的三句话做什么、不做什么、遇到问题找谁。这三句话能极大减少 agent 的越界行为。另外tools列表要尽量精简只给这个角色真正需要的工具。我早期犯的错是给所有角色都配了“搜索”工具结果每个 agent 都想去搜一下流程变得很乱。4.3 编排一个三 agent 流水线有了角色之后下一步是编排流程。agency-agents通常提供两种编排方式代码编排和配置编排。我推荐先用代码编排因为调试方便。下面是一个三 agent 流水线的示例from agency_agents import Agency, Pipeline agency Agency() # 注册角色 agency.register(CollectorRole()) agency.register(OrganizerRole()) agency.register(ReviewerRole()) # 定义流水线 pipeline Pipeline([ {role: collector, input: topic, output: raw_data}, {role: organizer, input: raw_data, output: structured_data}, {role: reviewer, input: structured_data, output: final_result}, ]) result agency.run(pipeline, topic某类产品的市场情况)这个例子里每个步骤的输入输出是显式声明的。这样做的好处是数据流清晰出问题时能快速定位是哪个环节的输出不对。我建议在每一步之后加一个日志打印把中间结果存下来方便回看。4.4 跑通之后的第一个优化点流程跑通后你大概率会发现一个问题速度慢。三个 agent 串行执行每个都要调用一次大模型总延迟是三者之和。第一个优化点就是并行化。如果某些步骤之间没有依赖关系可以让它们并行执行。比如“采集”和“背景调研”可以同时进行最后再合并。agency-agents通常支持并行编排你只需要在定义流水线时标记哪些步骤可以并行。但要注意并行会带来状态同步问题。如果两个 agent 同时写同一个记忆存储可能会冲突。我的做法是给每个并行分支分配独立的临时存储最后再合并到主存储。另一个优化点是缓存。如果某个角色的输出在多次运行中变化不大可以加一层缓存避免重复调用。比如“背景调研”的结果可能一周内都不会变缓存起来能省不少成本。5. 踩坑记录多 agent 系统里那些让人头疼的问题5.1 死循环为什么 agent 会互相踢皮球死循环是多 agent 系统里最常见的坑。表现是agent A 把任务转给 BB 觉得这不是自己的事又转回 AA 再转给 B无限循环。我遇到过一次两个 agent 互相请求对方“先提供更多信息”结果谁都不肯先动手。根因是职责边界模糊。当两个角色的 description 有重叠时它们就会互相推诿。解决办法是在每个角色的描述里明确“如果遇到 X 情况你应该自己处理不要转交”。另外调度器要加一个转交次数上限比如同一个任务被转交超过三次就强制终止并报错。这样至少不会无限跑下去。还有一个隐蔽的死循环是记忆污染导致的。某个 agent 的记忆里存了一个错误的任务状态导致它每次都做出同样的错误决策。这种问题很难从日志里看出来我的排查方法是定期清空短期记忆看问题是否复现。如果清空后正常那就是记忆污染。5.2 上下文爆炸token 消耗失控的排查过程多 agent 系统的 token 消耗很容易失控因为每个 agent 都有自己的上下文而且消息传递还会带上历史记录。我有一次跑一个五 agent 流程单次任务消耗了将近十万 token成本高得离谱。排查过程是这样的先看每个 agent 的输入输出长度发现“审核” agent 的输入特别长因为它收到了前面所有环节的完整输出。问题是它其实只需要看最终结果不需要看中间过程。于是我在消息传递时加了一个字段过滤只传必要字段。这一下就把 token 消耗降了六成。另一个 token 大户是系统提示词。如果每个 agent 的 prompt 都写得很长累加起来很可观。我的经验是系统提示词控制在 500 字以内把详细规则放到工具描述或外部配置里按需加载。5.3 输出格式不一致如何让 agent 说“同一种语言”多 agent 协作时上游 agent 的输出格式如果和下游的预期不一致下游就会解析失败。我遇到过“整理” agent 输出的是 JSON但“审核” agent 期望的是 Markdown 表格结果审核 agent 直接把 JSON 当文本处理输出了一堆废话。解决办法是定义统一的数据契约。在流程开始前明确每个环节的输入输出格式最好用 schema 约束。agency-agents通常支持在角色定义里声明输出格式我建议强制开启这个约束。如果某个 agent 的输出不符合 schema直接打回重做而不是让下游去猜。另外对于自然语言字段要约定好“不要加额外解释”。我见过 agent 在 JSON 外面包了一层“好的以下是整理结果”导致解析失败。在 prompt 里明确写“只输出 JSON不要任何额外文字”能避免大部分这类问题。6. 性能与成本调优让多 agent 流程跑得又快又省6.1 模型分级不是所有 agent 都需要最强模型多 agent 系统里如果每个 agent 都用最强模型成本会非常高。但实际上很多角色的任务并不需要那么强的推理能力。比如“格式转换”“字段提取”这类任务用轻量模型就够了。我的做法是给角色分三级决策级用最强模型执行级用中等模型格式级用轻量模型。实测下来这种分级能降低一半以上的成本而效果几乎不受影响。关键是要判断哪些角色真的需要强推理。一般来说涉及“规划”“审核”“辩论”的角色需要强模型而“采集”“格式化”“简单问答”用轻量模型即可。6.2 批处理与流式根据场景选择执行方式如果你的流程是处理批量任务比如一次处理一百条数据那么批处理比逐条处理效率高得多。agency-agents通常支持批量输入让 agent 一次处理多条。但要注意批处理时上下文会变长需要控制每批的数量避免超出模型限制。对于需要实时反馈的场景流式输出更合适。让 agent 边生成边返回用户能更快看到结果。但流式输出会给消息传递带来复杂性因为下游 agent 需要处理“不完整的输入”。我的建议是只在最终输出环节用流式中间环节还是等完整结果再传递。6.3 监控指标哪些数据值得盯跑多 agent 流程有几个指标必须监控单任务 token 消耗、单任务耗时、各角色调用次数、失败重试率。这几个指标能帮你快速发现异常。比如某个角色的调用次数突然飙升很可能是它在死循环token 消耗突然变大可能是上下文爆炸。我习惯在流程里埋一个简单的统计模块每次运行后输出这些指标。不需要复杂的监控系统一个日志文件加一个汇总脚本就够了。关键是持续看而不是出了问题才去看。7. 这套东西还能怎么扩展agency-agents的思路其实不局限于它本身的实现。理解了角色抽象、消息传递、调度编排这三层之后你完全可以在自己的项目里借鉴这套模型。比如你做一个自动化客服系统就可以用“接待角色”“查询角色”“升级角色”来组织而不是写一个巨大的 if-else。另一个扩展方向是跨流程复用角色。如果你把角色定义得足够独立那么同一个“审核”角色可以用在内容生产流程里也可以用在数据分析流程里。这能大幅减少重复配置的工作量。我目前就在整理一套通用角色库按领域分类新流程直接拿来组装。最后分享一个我在实际使用中的体会多 agent 系统的价值不在于 agent 多而在于边界清。我见过有人为了“看起来厉害”硬凑了十几个 agent结果流程乱成一团。真正好用的系统往往是三到五个角色每个角色职责清晰、工具精简、输入输出明确。与其追求数量不如把每个角色的边界打磨清楚这样系统的稳定性和可维护性会好很多。
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑