大概半年前我在做一个批量整理用户反馈的工具。最初版本只靠一个Agent从头干到尾提示词堆到两千多字结果它经常干着干着就忘了最初的目标有时还会在某个分支决定里来回打转。后来我把任务拆给多个各司其职的Agent让它们通过一块共享的“黑板”传递中间结果再配合一个很轻的调度层把流程串起来。这套东西我起了个名字叫 agency-agents。它解决的根本问题不是“单次调用大模型能产生多聪明的回答”而是“怎么让一群大模型Agent有条理地长期协作”。如果你也在从单Agent方案往多Agent结构迁移或者被长提示词、上下文爆炸、AI反复横跳折磨过这篇文章应该能从架构、代码和排错三个层面给你一点可复制的东西。1. 单Agent把号练废之后我开始拆团队1.1 单Agent方案的三个失灵点我最早的处理方式很简单把所有规则写进一个大提示词让模型一次性完成“读取反馈—归类问题—生成报告”三步。看起来很像人干活实际上一跑就露馅。第一个失灵点出现在上下文上。200条评论拆解出的临时结论会全部堆在同一段上下文里token一多后面的步骤就开始忽略前面的约束。第二个失灵点更难防一步错步步错。只要中间某个判断出现偏差后续所有输出都会踩在这个偏差上而且问题到底出在哪一步很难定位。第三个失灵点其实是工程问题——没法并行。一次只调一个模型哪怕前面两步明明可以同时做也只能老老实实排队。把这三个失灵点摊开看会发现一个共性我把“流程”和“执行者”耦合到了一起。所有职责都在同一个Agent里自然缺少可替换、可重试、可审计的边界。1.2 什么时候才值得拆团队不是说所有任务都必须上多Agent。我的判断标准很简单任务是否包含三个以上互不相同的子步骤且每一步需要的领域知识或输出格式差别很大。如果只是“总结一段文字”这种单步工作拆团队纯属浪费token。当时我手上那个反馈分析工具正好满足条件。提取用户反馈是一套逻辑归因到产品模块是另一套逻辑生成给团队看的报告又有一套独立的语气和结构要求。三个环节各自为政串在一起又互相依赖。这种情况下拆成三个Agent让每个Agent只回答一个窄问题比训练一个“全能Agent”要可靠得多。我后来甚至觉得单Agent方案的很多“不听话”不是模型智力问题而是职责过载。一个Agent又要提取、又要判断、又要写作任何一步的格式偏好都会污染其他步骤。1.3 “agency”到底指什么项目名里的agency不是“代理机构”而是强调每个Agent要有自己的“自主行动空间”。我给Agent的指令是“你要达成什么结果”而不是“你必须按第1、2、3步做”。后者不够灵活现实数据稍微偏离预期就卡死前者给模型留出选择具体方法的余地反而更容易产出稳定结果。当然自主行动不等于放任不管。我会在后面架构部分专门讲怎么用消息契约和黑板结构给这份“自主”套上边界。2. 分工、黑板和消息格式架构里最能省时间的三个决定2.1 先把每个Agent当成一个“岗位”而不是“模型”很多多Agent方案喜欢让Agent讨论我的做法正好相反尽量少讨论多干活。每个Agent对应一个岗位岗位说明书写清楚三件事——输入边界、输出格式、质量红线。比如负责提取反馈的岗位只需要接收原始文本输出固定结构的JSON包含涉及的模块名称、用户情绪和原文引用。它不需要知道报告长什么样。生成报告的岗位只需要读取前两步的JSON结果不接触原始评论。这么做有两个直接收益。第一每个Agent的提示词短短提示词意味着更少的注意力分散和更低的token成本。第二每个岗位可以单独替换我今天用A模型做提取明天换B模型做报告互不影响。2.2 黑板模式所有Agent共用一张工作台多Agent之间怎么传递信息我试过两种方式。第一种是点对点A把结果直接塞给BB再塞给C好处是链路直白坏处是Agent之间一旦有循环关系或者某个结果需要被多个下游步骤复用时依赖关系会乱成一团乱麻。我最后固定用黑板模式。黑板上不存原始对话只存每个Agent产出的结构化结果。一个Agent执行前告诉调度器“我需要哪些key”执行后把结果写到“blackboard”上。这样Agent之间解耦谁都不需要知道消息是从哪来的只需要知道该去黑板上拿什么。对比维度点对点直连黑板模式依赖关系显式但密集网状时很难维护统一由key管理逻辑清晰新增Agent需要改上游下游只需要声明读写key调试链路每个链接都需要单独盯所有结果集中查看最擅长的场景简单线性流水线有并行、分支、多个消费方的场景2.3 消息契约让每个Agent的输出能被下一个Agent消费黑板模式能不能工作全看写入黑板的“消息”是否遵守契约。我给每个Agent规定的输出格式不是自由文本而是JSON对象并且规定字段名。契约里最常见的字段有四组id表示当前任务编号status表示这次产出是否正常data放结构化结果meta放模型自己给出的置信度、来源引用和备注。其中meta很容易被忽略但后边排查幻觉接力时它救了我的命。契约还有一个隐含作用逼着Agent做出明确决定。如果模型拿不准它必须给低置信度说明而不是用模棱两可的话糊弄过去。比起让模型“尽可能完整地输出”这种强制结构更能提前发现异常。2.4 调度编排不只串行还要能并行和分支架构里的第三个决定是编排器怎么做。一开始我写的是硬编码流水线先执行A再执行B再执行C。后来发现很多任务并不是严格的线性关系。比如在反馈归因中先提取完反馈才能做归因但拆解报告和情绪分析可以并行跑。所以调度器不应该是一串if-else而应该围绕“依赖就绪”运行。每个任务声明自己的依赖key调度器持续扫描黑板谁的前置key齐了谁就进入可执行队列。执行完马上把结果写回黑板继续触发下一批。这套模型很轻但能表达串行、并行、条件分支。条件分支怎么做也简单前一步在黑板上写一个route字段调度器根据route的值决定下一步把哪个任务放进队列。3. 骨架代码怎么写得轻又能应付大多数任务3.1 三个核心抽象Task、Worker、Blackboard我写这套骨架时定了个原则不引入重型依赖核心抽象只有三个。明确的Task表示待办任务Worker表示会干活的AgentBlackboard表示公用的成果黑板。下面是我实际用下来的简化版代码去掉了一些日志和容错细节from dataclasses import dataclass, field from typing import Callable, Dict, List, Any dataclass class Task: task_id: str task_type: str inputs: List[str] children: List[str] field(default_factorylist) class Worker: def __init__(self, name: str, system_prompt: str, call_fn: Callable[[str, Dict[str, Any]], Dict[str, Any]]): self.name name self.system_prompt system_prompt self.call_fn call_fn def execute(self, payload: Dict[str, Any]) - Dict[str, Any]: # payload 是来自黑板的所有 input key 的聚合 prompt self.build_prompt(payload) result self.call_fn(self.system_prompt, prompt) return result def build_prompt(self, payload: Dict[str, Any]) - str: lines [] for key, value in payload.items(): lines.append(f[{key}]\n{value}) return \n\n.join(lines) class Blackboard: def __init__(self): self._store: Dict[str, Any] {} def has(self, key: str) - bool: return key in self._store def read(self, key: str) - Any: return self._store.get(key) def write(self, key: str, value: Any) - None: self._store[key] valuecall_fn是一个外部注入的函数实际项目里它封装了对模型API的调用、参数和重试逻辑。把它拆出来是为了让框架不绑定任何具体模型厂商。3.2 调度循环依赖就绪才执行调度器是整个框架的心脏。我用了一个很朴素的队列循环每次拿出一个任务检查它的依赖key在黑板上是不是都存在如果存在就执行如果依赖还没齐就把它放回队尾等下一轮。def run_pipeline(tasks: List[Task], workers: Dict[str, Worker], initial_board: Blackboard None) - Blackboard: board initial_board or Blackboard() pending tasks[:] while pending: task pending.pop(0) missing [key for key in task.inputs if not board.has(key)] if missing: pending.append(task) continue worker workers.get(task.task_type) if not worker: raise ValueError(fno worker for task type {task.task_type}) payload {key: board.read(key) for key in task.inputs} response worker.execute(payload) board.write(task.task_id, response) print(f[ok] {task.task_id} - {response.get(status)}) return board这个循环简单到有点笨但它能处理很多复杂流程。你可能会担心死循环万一一个任务的依赖永远不齐怎么办我在真实版本里加了一个最大轮次判断超过五十轮还执行不完就直接标记失败并把中间状态全部写到日志里。3.3 一个最小团队提取、归因、出报告为了让骨架看得见摸得着我给当时那个反馈分析工具搭建了一个最小Agent团队。总共三个Workerextract_feedback负责从原始评论中提取问题点attribute_issue负责判断问题属于哪个产品模块write_report负责生成给产品团队看的结论。这三个任务之间的依赖是extract_feedback不依赖任何上游可以直接执行attribute_issue依赖extract_feedback的结果write_report依赖前两者的结果。我只需要定义四个对象三个Worker和一个调度器启动入口剩下的全部由调度循环自动完成。def fake_llm_call(system_prompt: str, user_prompt: str) - Dict[str, Any]: # 真实项目里这里会替换为模型调用并且带超时和重试 return { status: ok, data: {summary: f{system_prompt[:20]} processed}, meta: {confidence: 0.85, source: raw input} } extract Worker( nameextract_feedback, system_prompt你只负责从反馈中提取问题点输出JSON不写分析。, call_fnfake_llm_call ) attribute Worker( nameattribute_issue, system_prompt你只负责把问题归因到具体模块输出JSON。, call_fnfake_llm_call ) report Worker( namewrite_report, system_prompt你只负责把多个模块的归因汇总成一段有重点的报告。, call_fnfake_llm_call )跑起来之后黑板上会依次出现三个key正好对应三个任务的产出。因为每个任务都声明了明确的输入key我可以随时调整attribute_issue的执行时机甚至可以增加一个并行执行的sentiment_analysis任务完全不需要改动已有代码。4. 上线一周我踩过的四个坑以及对应的补救手段4.1 死循环的根本原因不是模型是任务设计第一个坑说来有点搞笑。两个Agent在互相传消息时一个Agent说“你提供的材料缺少用户使用场景”另一个Agent马上补充了用户场景结果第一个Agent又说“这版材料缺少时间维度”于是第二个Agent继续补。它们就这样一问一答循环了二十多轮。问题不在模型多嘴而在任务设计给了它们“无边界对话”的空间。我的修复方式有三步。第一给整个流程加最大执行轮数超过阈值直接熔断。第二在调度器里记录每个任务每次输出的文本哈希一旦发现某Agent连续三次输出相似内容就自动终止本轮。第三也是最根本的我在消息契约里明确写了“禁止纯澄清型消息”每次输出必须包含data或者status不允许只提出问题不产出结论。经过这三步Agent之间再也没有出现那种礼貌的无限扯皮。4.2 上下文爆炸我把黑板当成无限存储了第二个坑是黑板越写越大。每个中间结果都被完整保留下游Agent拿到手的数据越来越多。原本每个子步骤只需要一小段数据结果我把整批待处理内容都塞给了生成报告的Agent提示词长度直接爆掉模型输出质量断崖式下降。后来我明确了一个原则黑板存储和Agent上下文是两回事。黑板上可以保存全部结果但每个Agent只能拿到它在输入契约里声明的那几个key。这不是技术上做不到而是之前设计偷懒把所有payload一股脑拼接进提示词。对于确实需要大量历史信息的任务我引入了压缩节点。一个专门的summarizerWorker负责把超长记录压成结构化摘要下游Agent只读摘要不读原始数据。这种设计有点像人看周报不会为了做决定把整个聊天记录翻一遍。4.3 幻觉接力下游Agent把上游编的数据当成事实第三个坑比前两个更隐蔽。我的提取Agent在个别评论上产生了错误归类但它输出的JSON里没有体现任何不确定性。下游的归因AgentA读到这个错误结果后很自然地把它当成事实继续分析最后报告Agent基于这个错误事实写出了一段看着很顺的结论。这就是所谓的幻觉接力源头的小误差沿链路被放大最终交付物和事实发生了偏离。我为此做了一套简单的置信度机制。每个任务输出必须带meta.confidence字段低于0.7的结果默认不允许直接进入下一步。同时在提示词里要求Agent在meta.source里写明依据来自哪些上游key。这套机制不能根除幻觉但它能把异常挡在扩散之前。至少现在出问题我可以顺着日志找到是哪一步的低置信度结果被错误放行了。4.4 日志比架构更容易被忽略第四个坑其实不是技术问题是我一开始太乐观觉得跑通一次就算成功。直到某个结果异常我想复盘整个链路发现日志里只有“任务执行成功”这种毫无信息量的记录。现在我规定每个Agent执行时至少要记录三份信息传入的payload摘要、模型返回的完整结果、以及本次调用消耗的上下文长度。这三份信息存成JSON Lines格式按任务ID组织。排查问题时我能直接从任务ID反推整个链路里每个关键时间点发生了什么。日志的价值在后头当我要优化流程、缩短等待时间或者做单元测试时日志就是我的“黑盒观测仪”。没有它多Agent系统的每一次失败都像是玄学。5. 如果让我重写一遍我会在写代码前先做三件事5.1 用一张表格定义产出物而不是先定义流程我现在接手新需求时第一件做的事不是画架构图而是列一张产出物清单。每个Agent负责什么、输出什么JSON、给谁消费先写在表格里。表格填完后流程自动就浮出来了。这个习惯帮我避免了很多“做着做着发现字段对不上”的问题。比如那个反馈分析工具我现在的表格长这样Agent产出key下游消费方核心数据字段置信度下限extract_resultattribute_issuemodule, quote, sentiment0.7attribute_resultwrite_reportmodule, issue_count, severity0.75report_result最终展示summary, action_items0.8先定契约再写代码Agent想跑偏都难。5.2 先让“最笨链路”跑通再加模型花活我还学到一条很实际的教训第一版尽量用最简单的方案跑通。不需要一上来就接复杂的模型、工具调用和记忆系统。先用模拟返回的fake_llm_call把整个流程跑通确认依赖、调度和输出格式都没问题再把真实模型接进去。这样能把“流程问题”和“模型问题”分开排查起来省一半时间。我见过太多项目一上来就用贵模型加复杂框架最后出了问题根本分不清是哪里坏了。先把链路走通再说优化这句话真的很值钱。5.3 把每个Agent都当作可以单独替换的组件最后一件我会坚持做的事维护好每个Agent的独立边界。任何Agent都只能通过黑板的key来读写数据不允许直接调用另一个Agent的内部函数。听上去是常识但实践中很容易被“临时改一下”打破。一旦破例链路就退化成点对点前期做的松耦合设计等于白做。边界清晰的额外好处是我可以单独升级某个Agent的提示词或者单独给它接更强的模型完全不影响其他岗位。这个特性在多Agent协作里简直像呼吸一样重要你永远不知道哪一个环节会因为业务调整需要重写。时至今日我在处理任何需要多个AI步骤参与的任务时第一反应都是先想岗位划分再想调度逻辑。agency-agents这个项目教会我的不是某段代码而是一种把复杂任务拆给多个自主个体协作的思维方式。每次动手前别看这个Agent不够聪明先看看它的职责是不是给得太宽了。