★ FEATURED ARTICLE 1. 从agency-agents这个标题说起一个多智能体协作框架的完整拆解第一次看到agency-agents这个标题的时候我脑子里蹦出来的第一反应是这大概率是一个围绕代理和智能体做文章的项目而且从命名习惯来看它更偏向于一个多智能体协作框架或者智能体编排系统而不是单点的工具库。为什么这么说因为agency这个词在软件语境里通常指向代理能力或者代理机构而agents是复数说明它不是一个智能体而是一群智能体在协同干活。这种命名方式在近两年的开源项目里非常常见基本可以判断它的核心诉求是让多个具备不同职责的智能体像一家公司里的员工一样分工协作共同完成一个复杂任务。我之所以对这个方向特别感兴趣是因为过去一年里我陆陆续续试过不少单智能体的方案踩过的坑非常集中——单个智能体一旦任务链条变长就会出现上下文爆炸职责混乱自我纠错能力差这三座大山。而多智能体协作恰好是绕开这些问题的思路之一把一个大任务拆成若干子任务每个子任务交给一个专职智能体再由一个协调者负责调度和汇总。这就像一个人干不完的活交给一个团队去干每个人只盯自己那一块效率和质量都会明显不一样。这篇文章我打算把agency-agents这类项目从里到外拆一遍它到底解决什么问题、核心架构怎么设计、智能体之间怎么通信、任务怎么分发和回收、实操中怎么落地、以及我在类似项目里踩过的那些坑。不管你是刚接触智能体概念的新手还是已经写过几个Demo想往工程化方向走的老手应该都能从里面找到能直接抄作业的部分。我会尽量用大白话把原理讲清楚同时把关键参数、目录结构、代码骨架都给到位让你看完就能自己搭一个能跑起来的最小版本。2. 多智能体协作到底解决什么问题需求拆解与方案选型2.1 单智能体的三个死穴在聊多智能体之前得先把单智能体为什么不够用说清楚不然你会觉得多智能体是为了复杂而复杂。我自己实测下来单智能体在真实任务里最容易崩的三个点第一是上下文窗口的物理限制。一个任务如果涉及十几个步骤、几十个文件、上百轮对话所有历史信息都往一个上下文里塞很快就会触顶。触顶之后要么截断历史导致失忆要么压缩历史导致细节丢失。我试过一个代码重构任务单智能体跑到第30轮的时候已经开始忘记前面定义过的接口名了输出的代码前后对不上。第二是职责混杂导致的人格分裂。你让一个智能体同时扮演需求分析师架构师程序员测试员它在不同阶段切换角色时提示词里的约束会互相打架。比如程序员角色要求快速产出可运行代码测试员角色要求严格挑刺这两个目标在一个上下文里会互相稀释最后产出的东西四不像。第三是错误累积无法隔离。单智能体一旦在某一步推理错了这个错误会污染后续所有步骤而且它自己很难发现——因为它没有第二双眼睛。这就像一个人写文章自己校对永远看不出自己的错别字。2.2 多智能体协作的核心思路agency-agents这类框架的核心思路说白了就是分而治之 专职专责 交叉验证。把复杂任务拆解成子任务每个子任务分配给一个专家智能体专家只关心自己那一亩三分地上下文自然就短了同时引入一个协调者负责拆解任务、分发任务、收集结果、处理冲突必要时再引入评审者对关键产出做二次校验把错误隔离在单步之内。这个思路不是凭空来的它借鉴了人类组织协作的成熟模式。一家公司为什么能完成一个人完不成的大项目因为有分工、有层级、有流程、有审核。多智能体框架本质上就是把公司的组织架构搬到了软件系统里。2.3 方案选型为什么是代理而不是流水线这里有个关键选型问题同样是多智能体你可以做成固定流水线A做完交给BB做完交给C也可以做成动态代理协调者根据任务情况动态决定派谁上。agency-agents从命名看更偏向后者也就是强调agency——代理的自主决策能力。固定流水线的优点是可控、可预测、好调试缺点是僵化遇到流程外的任务就抓瞎。动态代理的优点是灵活能处理开放式任务缺点是调试难度大、容易出现智能体互相踢皮球或者无限循环的问题。我的建议是新手从固定流水线起步跑通之后再逐步引入动态调度。一上来就搞全动态你会被各种边界情况折磨到怀疑人生。下面这张表是我总结的两种模式对比你可以根据自己的场景选对比维度固定流水线模式动态代理模式任务类型流程明确、步骤固定开放式、步骤不确定调试难度低每步可单独测高需要追踪调度日志灵活性低改流程要改代码高调度逻辑可配置资源消耗可预测波动大可能空转适用阶段原型验证、MVP生产环境、复杂任务典型风险流程外任务失败死循环、踢皮球2.4 适用场景与不适用场景多智能体协作不是银弹它有明确的适用边界。适合的场景包括长链条的代码生成与重构、多步骤的研究报告撰写、需要多视角评审的决策分析、复杂的数据处理管道搭建。这些场景的共同特点是任务可拆、子任务之间有明确依赖、单步产出需要校验。不适合的场景也很明确简单的单轮问答、实时性要求极高的任务多智能体调度有额外开销、子任务之间高度耦合无法拆分的任务。我见过有人拿多智能体去做一个简单的文本分类结果调度开销比推理本身还大纯属杀鸡用牛刀。3. 核心架构拆解一个多智能体系统由哪些部件组成3.1 四大核心角色一个能跑起来的多智能体系统最少需要四类角色缺一不可协调者Coordinator整个系统的大脑负责接收用户任务、拆解任务、决定派谁执行、收集结果、判断任务是否完成。它自己不干具体活只做调度。协调者的提示词设计是整个系统里最关键的它需要清楚每个下属智能体的能力边界。执行者Worker真正干活的智能体每个执行者只负责一类子任务。比如代码生成者文档撰写者数据分析者。执行者的提示词要窄而深不要让它什么都干。评审者Reviewer对执行者的产出做质量校验发现问题就打回重做。评审者的存在是多智能体相比单智能体最大的质量优势来源。我实测下来加了评审环节之后最终产出的可用率能从六成提到八成以上。记忆模块Memory存储任务历史、中间产出、智能体之间的通信记录。没有记忆模块智能体之间就没法共享上下文协作就无从谈起。3.2 通信机制智能体之间怎么说话智能体之间的通信机制是这类框架的技术核心。常见的有三种消息传递式智能体之间通过结构化的消息对象通信每条消息包含发送者、接收者、消息类型、载荷。这种方式清晰可控适合固定流程。消息对象一般长这样class AgentMessage: def __init__(self, sender, receiver, msg_type, payload, task_id): self.sender sender # 发送者ID self.receiver receiver # 接收者ID self.msg_type msg_type # 类型task/result/feedback self.payload payload # 实际内容 self.task_id task_id # 关联的任务ID self.timestamp time.time() # 时间戳共享黑板式所有智能体读写同一块共享内存黑板谁需要什么自己去取。这种方式灵活但容易产生竞态需要加锁机制。混合式关键路径用消息传递保证可控辅助信息用共享黑板提高效率。生产环境里混合式最常见。3.3 任务分发与回收任务分发有两种策略推模式和拉模式。推模式是协调者主动把任务塞给执行者适合任务量可预测的场景拉模式是执行者空闲时主动从任务队列里领任务适合任务量波动大的场景。任务回收的关键是状态管理。每个任务要有明确的状态机pending - assigned - running - completed / failed。协调者需要能查询任意任务的状态并在任务失败时决定是重试、换人还是上报。我踩过的一个坑是早期版本没做状态持久化程序一崩所有任务状态全丢重启后不知道哪些任务做完了哪些没做只能全部重来。后来加了状态落盘才解决。3.4 目录结构参考一个清晰的项目结构能让后续维护省一半力气。我常用的结构是这样的agency-agents/ ├── core/ │ ├── coordinator.py # 协调者实现 │ ├── worker.py # 执行者基类 │ ├── reviewer.py # 评审者实现 │ └── message.py # 消息定义 ├── agents/ │ ├── coder.py # 代码生成智能体 │ ├── writer.py # 文档撰写智能体 │ └── analyst.py # 数据分析智能体 ├── memory/ │ ├── short_term.py # 短期记忆当前任务 │ └── long_term.py # 长期记忆跨任务 ├── config/ │ ├── agents.yaml # 智能体配置 │ └── prompts/ # 提示词模板 ├── tests/ └── main.py # 入口这个结构的好处是职责边界清晰core放框架逻辑agents放具体智能体memory放记忆实现config放配置。新人接手的时候一眼就能找到该改哪里。4. 实操落地从零搭一个能跑的最小系统4.1 环境准备与依赖选择先把环境搭起来。我推荐用Python因为生态最成熟各种大模型SDK都有现成的。核心依赖其实不多pip install pydantic pyyaml tenacitypydantic用来做消息和配置的数据校验pyyaml读配置文件tenacity做重试。大模型调用部分你可以用任意一家提供API的服务这里不绑定具体厂商用统一的接口抽象层封装就行。提示不要一上来就引入复杂的编排框架先用最朴素的方式把流程跑通。框架会掩盖很多细节等你需要排查问题时才发现自己什么都不懂。4.2 定义智能体基类所有智能体都继承同一个基类基类负责统一处理日志、重试、消息收发这些横切关注点class BaseAgent: def __init__(self, agent_id, role_prompt, llm_client): self.agent_id agent_id self.role_prompt role_prompt self.llm llm_client self.inbox [] # 收件箱 self.outbox [] # 发件箱 def receive(self, message): self.inbox.append(message) def send(self, receiver, msg_type, payload, task_id): msg AgentMessage(self.agent_id, receiver, msg_type, payload, task_id) self.outbox.append(msg) return msg def think(self, context): # 子类实现具体推理逻辑 raise NotImplementedError def run_once(self): if not self.inbox: return None msg self.inbox.pop(0) result self.think(msg.payload) return result这个基类的设计要点是收件箱和发件箱分离智能体处理消息和发送消息解耦方便做异步。think方法留给子类实现基类不关心具体怎么推理。4.3 协调者的任务拆解逻辑协调者是整个系统里提示词最讲究的地方。它的核心任务是把用户的一句话需求拆成一组带依赖关系的子任务。我常用的提示词结构是这样的你是一个任务协调者。你的职责是把用户需求拆解成可执行的子任务。 可用的执行者 - coder: 负责编写和修改代码 - writer: 负责撰写文档和说明 - analyst: 负责数据分析和结论提炼 拆解规则 1. 每个子任务必须明确指定执行者 2. 子任务之间如果有依赖用 depends_on 字段标注 3. 子任务粒度控制在单个执行者一次能完成的范围内 4. 输出必须是JSON格式包含 tasks 数组 输出格式 { tasks: [ {id: t1, agent: analyst, desc: ..., depends_on: []}, {id: t2, agent: coder, desc: ..., depends_on: [t1]} ] }这里有个实操心得拆解粒度宁小勿大。我一开始让协调者把任务拆成三四个大块结果每个执行者拿到手还是一脸懵因为它面对的还是一个大任务。后来改成拆成八到十个小块每个块执行者一次就能搞定整体成功率明显上升。代价是调度次数变多但这点开销换来质量提升非常值。4.4 执行者的提示词设计执行者的提示词要遵循窄而深原则。以代码生成智能体为例你是一个代码生成专家。你只负责根据给定的需求编写代码。 约束 1. 只输出代码不要输出解释性文字 2. 代码必须包含必要的注释 3. 如果需求不明确在代码开头用注释标注 TODO 4. 不要假设任何未在需求中说明的依赖 输入{task_description} 上下文{relevant_context}关键点是明确告诉它不要做什么。大模型有个通病就是喜欢热心地帮你做额外的事比如你只要一个函数它给你写一整个模块。在提示词里把边界划死能省掉大量返工。4.5 评审者的校验机制评审者是质量守门员。它的提示词核心是挑刺你是一个严格的评审者。你的职责是找出产出中的问题。 检查清单 1. 是否完整覆盖了任务要求 2. 是否存在逻辑错误或前后矛盾 3. 是否违反了任何约束条件 4. 是否有明显的遗漏或偷工减料 输出格式 {pass: true/false, issues: [问题1, 问题2], suggestion: 修改建议}评审不通过时协调者会把问题反馈给原执行者重做。这里要设置最大重试次数我一般设3次超过3次就上报给用户避免无限循环烧钱。4.6 完整流程串起来把上面这些部件串起来主流程大概是这样def run_task(user_request): # 1. 协调者拆解任务 tasks coordinator.decompose(user_request) # 2. 按依赖顺序执行 results {} for task in topological_sort(tasks): # 等待依赖完成 context {dep: results[dep] for dep in task.depends_on} # 分配给执行者 worker get_worker(task.agent) output worker.think(task.desc, context) # 评审 review reviewer.check(task.desc, output) retry 0 while not review.pass and retry 3: output worker.think(task.desc, context, feedbackreview.issues) review reviewer.check(task.desc, output) retry 1 results[task.id] output # 3. 汇总结果 return coordinator.summarize(results)这个骨架跑通之后你就有了一个最小可用的多智能体系统。接下来所有的优化都是在这个骨架上做加法。5. 常见问题与排查技巧实录5.1 智能体互相踢皮球怎么办这是动态代理模式最常见的问题。协调者把任务派给AA觉得这不是自己的活又转回给协调者协调者再派给BB又转回来死循环。排查思路是看调度日志找到循环的起点。解决办法有三个一是给每个智能体明确的能力边界在提示词里写清楚你只处理X类任务遇到Y类任务直接返回失败而不是转交二是设置转交次数上限一个任务被转交超过2次就强制由协调者自己处理或上报三是引入任务归属锁定任务一旦分配给某个智能体除非它明确返回失败否则不允许转交。5.2 上下文还是太长怎么办即使拆了任务单个执行者的上下文有时还是会长。这时候要做上下文裁剪只把与当前子任务直接相关的信息传给执行者其他信息用摘要代替。我常用的做法是让协调者在分发任务时额外生成一段任务相关上下文摘要只把摘要传给执行者而不是把全部历史都塞过去。另一个技巧是分层记忆短期记忆存当前任务的详细过程长期记忆只存结论和关键决策。执行者默认只读短期记忆需要历史结论时再主动查长期记忆。5.3 评审者太严或太松评审者的尺度很难一次调准。太严会导致大量无意义返工太松则起不到质量把关作用。我的经验是分阶段调整原型阶段把评审标准调松先保证流程能跑通稳定之后再逐步收紧每次只加一条检查项观察通过率变化。还有一个技巧是给评审者分级轻度问题只记录不阻塞重度问题才打回重做。这样既保证了质量又不会因为小瑕疵卡住整个流程。5.4 常见问题速查表问题现象可能原因排查方向解决手段任务卡住不动依赖环或死锁检查依赖图拓扑排序检测环智能体反复转交能力边界不清看调度日志明确边界转交上限产出质量差提示词太宽泛检查提示词收窄职责加示例成本飙升无限重试看重试计数设最大重试次数结果前后矛盾上下文不一致检查记忆模块统一上下文来源评审总不通过标准过严看评审意见分级评审放宽标准5.5 几个我踩过的坑坑一忘了做幂等。任务重试的时候如果执行者有副作用比如写文件、调外部接口重试会导致重复执行。解决办法是给每个任务加唯一ID执行前先检查这个ID是否已经执行过。坑二日志打太少。多智能体系统的调试全靠日志。我早期版本日志打得太少出问题只能靠猜。后来改成每个消息收发、每次推理、每次评审都打日志排查效率提升十倍。日志要包含时间戳、智能体ID、任务ID、消息类型方便过滤。坑三没做超时控制。某个智能体卡住不返回整个流程就挂起。后来给每个环节加了超时超时后走降级逻辑比如用上一次的结果或者跳过这一步。坑四配置硬编码。智能体的提示词、模型参数一开始都写在代码里改一次要重新部署。后来全部抽到yaml配置文件改配置不用动代码迭代速度快了很多。6. 性能优化与扩展方向6.1 并发执行有依赖关系的任务必须串行但没有依赖关系的任务完全可以并行。把任务图做拓扑分层同一层的任务并发执行整体耗时能大幅下降。实现上用线程池或者异步IO都行注意并发写共享状态时要加锁。6.2 缓存复用相同的子任务如果重复出现可以直接复用之前的结果。给每个子任务算一个内容哈希执行前先查缓存。这在迭代开发场景下特别有用改了一个小地方大部分子任务的结果都能复用。6.3 智能体能力扩展系统跑通之后扩展新能力就是加新的执行者。比如加一个翻译者处理多语言需求加一个格式化者处理输出格式。每个新执行者只需要实现think方法注册到配置里就能用不用改框架代码。这就是好的架构带来的扩展性红利。6.4 人工介入点生产环境里关键节点最好留人工介入的口子。比如任务拆解完成后先让人确认一下拆得对不对再执行或者最终产出提交前让人过一眼。完全自动化的系统在复杂任务上还是不够可靠人机协同才是现阶段最务实的方案。我在实际项目里发现多智能体系统最大的价值不是完全替代人而是把人从重复劳动里解放出来让人只做关键的判断和决策。把这一点想清楚你在设计系统的时候就知道哪些环节该自动化、哪些环节该留给人。这个思路比任何具体的技术选型都重要。
阅读完成 · 觉得有帮助？
