尧图网站建设 
做内容自动生产管线那段时间我遇到了一个特别典型的瓶颈单Agent把写稿-查资料-查代码-审校全流程扛下来任务越长质量越飘经常写到后半段自己前面说了什么都忘了。后来我把这个流程改成了多Agent协作——一个Agent负责写另一个Agent负责挑错再有一个Agent专门做事实核查最后主编Agent汇总定稿。就是这么一拆质量立刻稳了。今天这篇我就把我做多Agent编排时沉淀下来的架构思路、协作机制、可复用代码和踩坑记录完整写出来给正在研究多Agent编排示例的朋友一份能直接上手的参考。多Agent协作这几年已经从概念讨论走到工程落地阶段了。但很多人第一步就卡住了把多个AI Agent拼在一起跟把一个全能Agent调大参数到底有什么区别真正跑起来之后任务分配、上下文共享、结果收敛这些细节但凡有一个没设计好整个编排就会变成三个和尚没水喝。这篇文章不会停在概念层我会从一个真实可跑的多Agent编排示例出发把架构模式、协作机制、代码实现和避坑经验一条条讲透适合正在做Agent落地、或者准备从单Agent转向多Agent的开发者参考。1. 先从为什么要拆说起单Agent的边界到底在哪很多刚从ChatGPT API入门的同学会有个疑问我的一个Prompt写得足够好是不是就不需要多Agent了我的回答是看场景。如果你只是写一段文案或者翻译一篇文章那单Agent完全够用拆反而多此一举。但一旦任务满足下面几个特征单Agent就会迅速逼近能力边界。第一个特征是任务跨度大。比如做一篇技术评测文章你要先调研产品参数再设计实验跑结果数据然后写分析稿最后还要做合规检查。这个流程让同一个Agent从头做到尾它在前面调研时收集的上下文写到一半可能已经被稀释。大语言模型的注意力是有限的上下文越长它对早期信息的记忆越不可靠。我在实际测试里发现超过一定长度的单Agent长任务后期的输出常常会偏离早期设定的约束条件。第二个特征是角色要求互相冲突。写作角色要求生动、有感染力事实核查角色要求严谨、每个数据可溯源安全审校角色要求敏感词零容忍。这三个要求放在同一个上下文里模型很难同时满足经常出现顾此失彼。而拆成多个Agent之后每个Agent只维护一套角色目标行为就稳定得多。第三个特征是反馈闭环。单Agent写完就直接交稿错了也没有人拦着。多Agent协作天然形成生产者-检视者结构写稿Agent产出初稿审校Agent发现问题打回重写这种循环往复的自我校准过程是单Agent结构很难内部实现的。这里我想强调一个关键认知多Agent不是把任务分给多个万能人而是给每个Agent一个极其专注的单一职责让它们像公司部门一样协同。专业化的收益会抵消掉沟通损耗前提是职责边界划得足够清楚。那拆了之后收益到底能有多大我从自己的项目里拿了一个对比数据同一篇内容使用单Agent一次性生成事实性错误大约每千字会出现1-3处使用多Agent协作写作Agent产出、事实核查Agent逐条核验、合规Agent扫描、编辑Agent终审之后错误率降到了每千字0.5处以下。工程里面实体URL、数据来源、代码示例都不需要人工核对。这个提升不是我调了某一个超参得到的而是结构带来的。2. 多Agent编排的四种典型架构以及它们各自适合的舞台多Agent不是只有一个fixed pattern。根据任务依赖方式不同我梳理了四种常用的编排架构实践中90%的项目跑不出这个范围。理解这四种模式做技术选型时才不会拍脑袋。2.1 主编-记者模式Orchestrator-Worker这是最经典也最好理解的一种一个统筹规划的主控Agent下面挂若干个专职执行的Worker Agent。主控负责把大任务拆解成子任务、派发给合适的执行Agent、收集结果、处理异常。Worker之间原则上不直接通信所有沟通都通过主控中转。我那个内容审校管线就是这个模式主编AgentOrchestrator负责整体流程写手Agent、代码核查Agent、事实核查Agent、合规审查Agent各管一段。主控持有任务状态Worker只认自己的KPI。这个模式的好处是结构清晰、出错后容易定位缺点是主控可能成为瓶颈而且主控理解错了任务下级再能干也是白干。2.2 流水线模式Pipeline流水线就是把任务拆成一串有序节点A的输出是B的输入像工厂流水线一样。适合任务步骤天然有先后顺序的场景比如需求分析 - 架构设计 - 代码生成 - 代码评审 - 测试生成。流水线的核心设计点在每个节点只关注它眼前那一段输入上下文窗口压力小每个节点可以把全部能力集中在当前环节。缺点也明显一个节点质量不好错误会向下游持续放大必须给关键节点设计质检门禁。我在实际项目里会在流水线中间加入质量闸门Agent专门检查上个节点的输出是否达标再放行这不属于业务节点但作用非常大。2.3 辩论/评审模式Debate/Review让多个Agent从不同立场审视同一个问题最后经过辩论或者投票收敛出一个结果。典型场景是决策评审、文案质量评估、代码安全审计。比如让三位Agent分别扮演乐观方风险方用户体验方对一份需求文档提出意见再由仲裁Agent汇总。这种模式能够暴露单视角盲区代价是Token消耗大、响应时间变长。控制好轮数和最小集合很重要不然就是无休止的辩论。2.4 层级组织模式Hierarchical当Worker数量太多比如超过10个主编一个人管不过来时就需要中间管理层。顶级主编管几个组长Agent组长Agent各自管一组Worker。这种设计适合大型系统比如一个模拟公司运营的多Agent系统或者覆盖多业务线的客服体系。层级模式我只有在节点数超过15个时才推荐节点少的时候上管理层纯粹是增加延迟。我把这四种模式放在一起做一个对比方便你快速判断架构模式核心特征适合场景主要风险主编-记者中央调度Worker只干不说子任务相对独立、需要统一协调主控单点瓶颈主控理解偏差流水线上家产出喂给下家步骤天然有序、环形依赖少错误向下游放大调试定位难辩论/评审多角色交叉审视收敛结果质量评估、策划评审、安全审计Token消耗大可能不收敛层级组织多层管理分组协作节点数量多、业务线复杂管理成本高链路变长风险变大实际项目里这几种模式经常混用。我自己的管线就是个典型主线是流水线审校阶段临时切换成辩论模式做多Agent交叉审校最后再由主编Agent汇总裁决。3. 协作机制设计上下文互通、任务编排与结果收敛的细节架构定好了接下来是让多个Agent真正协作起来的三个关键机制。这三个机制做不好多Agent就只是多个Agent各干各的跟并行调用API没什么本质区别。3.1 上下文互通从传大段Prompt到共享工作台多Agent协作的第一个坑就是试图把一个Agent的全部输出上下文直接塞给另一个Agent。结果是上下文重复膨胀token消耗爆炸而且无用信息稀释了核心信息。我的做法是引入**共享黑板Shared Blackboard**模式所有Agent共用一个结构化的任务状态空间每个Agent只往黑板上写入自己领域的关键结果读取时也只看自己关心的键值。比如说事实核查Agent它不需要读全文它只需要从黑板上读到待核查句子列表和参考来源列表输出核查结论后写到核查结果键上。分工明确上下文友好信息不但没有丢失反而比大段文本传递更结构化。3.2 任务编排DAG是一切的基础但别做成无限循环任务编排的本质是解决哪个Agent先跑、哪些Agent可以并行、哪个Agent失败了怎么处理。我强烈建议用**有向无环图DAG**来建模任务依赖而不是自己写一堆干巴巴的if-else。把每个Agent看作一个节点依赖关系看作有向边节点入度为零时就可以执行。以我的审校管线为例任务依赖关系是这样的writer无依赖 → code_checker依赖writer → fact_checker依赖writer → compliance_checker依赖code_checker fact_checker → editor_review依赖compliance_checker在这个DAG里code_checker和fact_checker在writer完成之后是天然可以并行执行的如果我用一个串行Pipeline写就会浪费近一半的墙上时间。所以DAG模型不只是为了好看它是实打实的效率优化。关于失败重试我的原则是两类确定性失败不重试比如输入格式错误、Agent返回了空结果直接走降级方案不确定性失败可以重试比如LLM本次生成内容质量不达标、超时重试次数上限2-3次。重试超过上限就必须报错不能无脑循环。3.3 结果收敛多Agent争论完了怎么拍板辩论模式最容易出现永不收敛的问题——每个Agent都有自己的立场谁也不服谁。结果收敛的关键是设计一套仲裁协议。先说最简单的投票协议针对一个候选方案让每个评审Agent打质量分所有人打分平均值超过阈值方案通过否则把差评意见汇总打回重做。这是比较民主的机制适合评估类任务。还有一种是主编裁决协议评审Agent们只负责提意见最终决定权在主控Agent手里。主控给每个评审角色一票但允许最终仲裁人行使一票否决。这种做法快手、可控适合时间敏感的场景。收敛后还有一个容易被忽视的动作——明确关闭任务。我知道很多半吊子的编排示例跑完一轮就结束了但实际上应该把最终结果、评审轨迹、各Agent中间产物一并写回状态空间这样既能供后续复盘也能让编排系统知道这个任务已经终态避免某次重试又把它拉回活状态。4. 一个可直接复现的多Agent编排示例技术文章自动审校管线理论说完了拿点实际的东西出来。下面这个示例就是一个完整可跑的多Agent编排示例我用纯Python实现不依赖任何重型框架保证你能在一个小时内跑通并理解全部细节。它做的事情是模拟一篇技术文章的生成与审校过程。4.1 先定义协作基础结构消息、黑板上墙、Agent基类第一步搭好地基。我定义一个AgentMessage作为Agent间传递的消息结构一个SharedBlackboard作为共享状态空间以及一个BaseAgent基类强制每个子Agent实现process方法。from dataclasses import dataclass, field from typing import Dict, List, Any, Optional import json, time dataclass class AgentMessage: sender: str receiver: str content: str meta: Dict[str, Any] field(default_factorydict) class SharedBlackboard: 共享黑板所有Agent读写公共状态的地方 def __init__(self): self._data: Dict[str, Any] {} def write(self, key: str, value: Any): self._data[key] value def read(self, key: str, defaultNone): return self._data.get(key, default) def append(self, key: str, item: Any): self._data.setdefault(key, []).append(item) def snapshot(self) - Dict[str, Any]: return json.loads(json.dumps(self._data, defaultstr)) class BaseAgent: name: str base def __init__(self, temperature: float 0.3): self.temperature temperature def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: raise NotImplementedError这里有一个实际工程中的细节黑板的write和read非常简单但append方法是我后来补充的。因为在审校过程中检视者会不断把已发现问题追加到列表里而不是覆盖某个键。这种追加语义在多Agent场景下非常常见。4.2 实现四类专职Agent写作、代码校验、事实核查、合规审查接着实现具体的Agent。为了让你专注理解协作机制我把LLM部分用模拟函数替代——每类Agent的输出可以用一段带有固定语义的字符串模拟比如模拟判断函数名正确、变量类型匹配。等你接入真实大模型时只需要替换这一层。class WriterAgent(BaseAgent): 写作Agent产出文章初稿并把待检点写入黑板 name writer def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: # 真实场景: 调用 chat.completions.create(...)把完成草稿写回 draft 本文介绍了异步任务队列入门示例代码使用 Python 和 Redis。 code_snippet import redis\nr redis.Redis(hostlocalhost)\n board.write(draft, draft) board.write(code_snippet, code_snippet) board.write(claims_to_check, [ Redis 默认端口是 6379, 关于异步队列任务积压时应该优先扩容消费者, ]) return AgentMessage(senderself.name, receiverorchestrator, contentf初稿完成共{len(draft)}字, meta{draft_length: len(draft)}) class CodeCheckerAgent(BaseAgent): 代码校验Agent检查代码块是否有明显问题 name code_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: code board.read(code_snippet, ) # 真实场景: 让LLM执行静态审查同时可以接 pylint/mypy 等确定性工具 issue self._fake_scan(code) if issue: board.append(issues, {agent: self.name, issue: issue}) status pass if not issue else fail board.write(code_check_status, status) return AgentMessage(senderself.name, receiverorchestrator, contentf代码校验完成: {status}, meta{status: status}) def _fake_scan(self, code: str) - Optional[str]: if password in code: return 代码中疑似包含明文口令 if len(code.strip()) 0: return 代码块内容为空 return None class FactCheckerAgent(BaseAgent): 事实核查Agent逐条核对文章中的数据与来源 name fact_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: claims board.read(claims_to_check, []) # 真实场景: 使用联网搜索工具检索并比对可信数据库 for claim in claims: if 6379 in claim: board.append(verified_claims, {claim: claim, result: true}) else: board.append(verified_claims, {claim: claim, result: uncertain}) board.append(issues, {agent: self.name, issue: f无法核实: {claim}}) board.write(fact_check_status, done) return AgentMessage(senderself.name, receiverorchestrator, contentf事实核查完成共核查{len(claims)}条) class ComplianceAgent(BaseAgent): 合规审查Agent检查内容是否符合发布规范 name compliance_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: draft board.read(draft, ) sensitive_words [危险词A, 限用词B] # 真实场景: 维护一个专业词库 found [w for w in sensitive_words if w in draft] if found: board.append(issues, {agent: self.name, issue: f敏感词: {found}}) board.write(compliance_status, pass if not found else fail) return AgentMessage(senderself.name, receiverorchestrator, contentf合规审查完成命中敏感词{len(found)}个)这段代码的核心思路大家注意一下每个Agent在process方法里只有三件事——从黑板上取自己要的输入、做自己的领域判断、把结果写回黑板或者在黑板的问题列表里追加记录。Agent之间没有直接的方法调用这就是协作和串行Pipeline的本质区别。4.3 主编Agent与DAG调度让任务按依赖跑起来紧跟着是编排的核心主编Agent和调度器。我的主编Agent职责是发起任务、接收所有执行Agent的状态、决定是否进入终审。而调度器负责解析DAG计算出每个节点何时可以被执行。from collections import deque class OrchestratorAgent(BaseAgent): name orchestrator def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: issues board.read(issues, []) if issues: return AgentMessage(senderself.name, receivereditor_review, content存在待处理问题请结合评审意见终审, meta{issue_count: len(issues)}) return AgentMessage(senderself.name, receivereditor_review, content全部检查通过进入终审, meta{issue_count: 0}) class GraphScheduler: 基于DAG的任务调度器统计每个节点的入度并依次执行 def __init__(self, agents: Dict[str, BaseAgent], graph: Dict[str, List[str]]): self.agents agents self.graph graph self.indegree {node: 0 for node in graph} self.children {node: [] for node in graph} for node in graph: for dep in graph[node]: self.indegree[node] 1 self.children[dep].append(node) def run(self, board: SharedBlackboard): # 按入度为零动态推进为保证可复现性这里先用广度优先执行 # 在生产环境配合 concurrent.futures 就可以实现真正并行 queue deque([node for node, deg in self.indegree.items() if deg 0]) executed [] while queue: node queue.popleft() agent self.agents[node] msg AgentMessage(senderscheduler, receivernode, contentstart) result agent.process(msg, board) print(f[{node}] - {result.content}) executed.append(node) for child in self.children[node]: self.indegree[child] - 1 if self.indegree[child] 0 and all(dep in executed for dep in self.graph[child]): queue.append(child) return executed4.4 组装跑通把全链路一条命令串起来最后把这些零部件组装成一个完整的main函数。调度器按DAG执行完所有业务节点后把最终数据统一交给终审模块生成交付报告。def main(): agents { writer: WriterAgent(temperature0.7), code_checker: CodeCheckerAgent(temperature0.2), fact_checker: FactCheckerAgent(temperature0.2), compliance_checker: ComplianceAgent(temperature0.2), orchestrator: OrchestratorAgent(temperature0.3), } graph { writer: [], code_checker: [writer], fact_checker: [writer], compliance_checker: [code_checker, fact_checker], orchestrator: [compliance_checker], } board SharedBlackboard() scheduler GraphScheduler(agents, graph) executed scheduler.run(board) # 终审输出审校报告 report { executed_agents: executed, issues: board.read(issues, []), verified_claims: board.read(verified_claims, []), final_decision: rework if board.read(issues, []) else publish, } print(json.dumps(report, ensure_asciiFalse, indent2)) if __name__ __main__: main()跑这个main()你会看到类似这样的输出[writer] - 初稿完成共40字 [code_checker] - 代码校验完成: pass [fact_checker] - 事实核查完成共核查2条 [compliance_checker] - 合规审查完成命中敏感词0个 [orchestrator] - 全部检查通过进入终审然后看终审报告你会发现issues列表为空但verified_claims里有两条核验记录final_decision是publish。这就完整走通了一个多Agent协作闭环。如果你要接入真实大模型有几个位置替换一下即可WriterAgent的draft赋值处改成调用LLMAPI并解析结构化输出FactChecker的claimed核对改成工具调用比如联网检索Compliance的词库改成加载你自己的敏感词表。其余协作逻辑一行都不用动。5. 别急着上框架自研轻量编排与成熟框架的选型判断标准拿到上面的示例之后很多人会问我现在直接上LangGraph、AutoGen或者CrewAI行不行我的观点是如果你的场景里Agent数量不超过10个、流程相对固定自研这个轻量编排反而更香。原因有三点。第一成熟框架的抽象层很厚。LangGraph这类框架为了方便你快速搭图设计了专门的State、Node、Edge概念学习成本不低。你真去读了它的文档把它掰开来用会发现顶层API已经在悄悄替你做很多决策。而你做的多Agent协作项目如果自定义逻辑很强框架提供的默认行为往往变成阻碍你花在摆脱框架上的时间比你自己写还多。第二可控性很关键。自研版每行代码你都知道在干什么出问题直接断点调。框架版本里一旦出现回环重试失控状态更新被静默覆盖排查成本会让你怀疑人生。特别在做生产级管线时可观测性和可控性对稳定性的贡献往往比框架的开箱即用重要得多。第三轻量编排迭代速度更快。框架的发布周期、API变动都不由你控制。我自己早期用AutoGen跑过一个小Demo一堆魔法函数和隐式上下文跑通了翩然自喜等到要往里加企业安全审查逻辑的时候几乎等于重写。反而是自研这套几十行DAG调度器改起来得心应手。那什么情况下该用框架呢我的判断标准是单Agent能力已经不够用而你也暂时没有想清楚协作细节先用框架的High-Level API快速验证多Agent的可行性与收益点再决定是否自研。比如你只是想在一天之内体验一下多Agent辩论是什么感觉那AutoGen的Group Chat几乎三五行代码就能搞定。Tool-Use类任务用LangGraph因为它有完整的Tool调用生态。还有一个常见误区是用了框架就自然实现了多Agent协作。框架只是管道的骨架它不替你解决上下文互通怎么设计、结果怎么收敛、失败怎么降级这些真正的工程问题。你以为换框架能解决协作难题结果只是换了个地方踩同一个坑。自研轻量编排的正确打开方式用到一定程度你需要的是从里面长出自己的通用能力——消息总线、状态快照、重试策略、审计日志。这些在自研代码里就是四个函数的事在框架里则要跟框架的抽象做斗争。所以我的建议是双轨并行小规模探索可以用开源框架快速验证一旦确认方案要进生产就把关键路径改造成自己的轻量调度同时保留框架作为可选的后端。这样既有框架带来的开发速度又有自研带来的确定性。6. 多Agent落地时我踩过的坑以及对应的补救方案最后这部分全是真金白银的踩坑记录。我把最常让新手栽跟头、也最影响实际效果的问题和解决方案整理成一张对照表然后挑几个典型展开讲。常见问题根本原因补救方案上下文爆炸Token消耗不可控每个Agent都继承全部对话历史改共享黑板模式Agent只读自己关心的键Agent之间格式不统一不同Agent输出markdown、JSON、纯文本混杂统一结构体如AgentMessage附带meta字段与协议版本任务反复横跳不收敛审查Agent发现问题就打回没有终审仲裁设置仲裁协议限定重试上限终审Agent拥有拍板权单个Agent失败拖垮全局没有处理Agent级故障的隔离策略超时熔断降级输出可重跑子任务编排Demo能跑生产一换真实LLM就崩模拟函数掩盖了LLM输出的随机性先垫一层输出解析与校验失败时强制重试6.1 上下文爆炸从全量传到按需读我最早做的版本是把上一个Agent的全量输出拼成一个新Prompt丢给下一个Agent结果测试时token消耗飙升到原来的五倍而且下游Agent经常被无关文本干扰输出质量不升反降。后来我彻底抛弃了大文本传递改成了黑板式按需读写。实践下来token消耗能降到原来的两成左右而且各Agent输出的稳定性明显提高。这个改动是整个项目里性价比最高的一次重构。6.2 格式混乱问题的解法多个Agent协作最烦的就是A输出的JSON丢了字段、B输出的markdown里混了代码块C直接把表格用普通文本画了。我的做法是三层加固第一层AgentMessage的meta字段强制要求sender、receiver、version等信息第二层每个Agent的输出必须有明确的schema声明比如返回JSON包含status与detail两个字段第三层在调度器里加一个轻量的格式校验函数解析不通过直接判失败并触发重试。这样能在第一时间发现问题而不是等到终审才发现一堆脏数据。6.3 任务不收敛和无休止循环有次做评审型多Agent三个评审Agent对一份需求文档的意见完全对立主编Agent又特别民主导致同一个文档被打回重写了五轮时间和token全花在来回扯皮上。之后我加了两条硬规则评审Agent只提两轮意见第三轮起只允许同意或弃权主编终审有一票否决权可以拿着不完全满意但风险可控的结果直接放行。设置了这个收敛规则后类似任务的完成时间从原来的十二分钟压缩到了四分钟。6.4 单点失败的雪崩效应流水线模式下第二个节点崩了后面所有节点都跟着白干。我后来在每个节点外面套了隔离舱节点执行前设置超时超时直接记录失败并走fallback失败节点向主编汇报时携带可重跑标记由主编决定是从失败节点重做还是整条链路重来。加上这个机制之后管线对单个Agent抽风的容忍度大幅提升再也不需要整条重跑。实话实说多Agent项目里商量和争执都不难难的是让这个系统长期稳定地维持产出质量。我现在的习惯是每次跑完一个多Agent任务一定会去把黑板的snapshot、每个Agent的输出轨迹、终审决定存下来做一次对账。就是这个习惯帮我把很多偶发的质量问题提前盯了出来而不是等到用户反馈了才四处排查。内容管线的多Agent编排跑通只是开始真正的功夫在于监控它每一轮是怎么协作、怎么争议、怎么收敛的。如果你正在推动一个单Agent转多Agent的项目先把这些机制想明白、把坑提前避掉能省下的时间一定会让你觉得值。 多Agent编排实战：从单Agent到协作架构的完整落地指南
做内容自动生产管线那段时间我遇到了一个特别典型的瓶颈单Agent把写稿-查资料-查代码-审校全流程扛下来任务越长质量越飘经常写到后半段自己前面说了什么都忘了。后来我把这个流程改成了多Agent协作——一个Agent负责写另一个Agent负责挑错再有一个Agent专门做事实核查最后主编Agent汇总定稿。就是这么一拆质量立刻稳了。今天这篇我就把我做多Agent编排时沉淀下来的架构思路、协作机制、可复用代码和踩坑记录完整写出来给正在研究多Agent编排示例的朋友一份能直接上手的参考。多Agent协作这几年已经从概念讨论走到工程落地阶段了。但很多人第一步就卡住了把多个AI Agent拼在一起跟把一个全能Agent调大参数到底有什么区别真正跑起来之后任务分配、上下文共享、结果收敛这些细节但凡有一个没设计好整个编排就会变成三个和尚没水喝。这篇文章不会停在概念层我会从一个真实可跑的多Agent编排示例出发把架构模式、协作机制、代码实现和避坑经验一条条讲透适合正在做Agent落地、或者准备从单Agent转向多Agent的开发者参考。1. 先从为什么要拆说起单Agent的边界到底在哪很多刚从ChatGPT API入门的同学会有个疑问我的一个Prompt写得足够好是不是就不需要多Agent了我的回答是看场景。如果你只是写一段文案或者翻译一篇文章那单Agent完全够用拆反而多此一举。但一旦任务满足下面几个特征单Agent就会迅速逼近能力边界。第一个特征是任务跨度大。比如做一篇技术评测文章你要先调研产品参数再设计实验跑结果数据然后写分析稿最后还要做合规检查。这个流程让同一个Agent从头做到尾它在前面调研时收集的上下文写到一半可能已经被稀释。大语言模型的注意力是有限的上下文越长它对早期信息的记忆越不可靠。我在实际测试里发现超过一定长度的单Agent长任务后期的输出常常会偏离早期设定的约束条件。第二个特征是角色要求互相冲突。写作角色要求生动、有感染力事实核查角色要求严谨、每个数据可溯源安全审校角色要求敏感词零容忍。这三个要求放在同一个上下文里模型很难同时满足经常出现顾此失彼。而拆成多个Agent之后每个Agent只维护一套角色目标行为就稳定得多。第三个特征是反馈闭环。单Agent写完就直接交稿错了也没有人拦着。多Agent协作天然形成生产者-检视者结构写稿Agent产出初稿审校Agent发现问题打回重写这种循环往复的自我校准过程是单Agent结构很难内部实现的。这里我想强调一个关键认知多Agent不是把任务分给多个万能人而是给每个Agent一个极其专注的单一职责让它们像公司部门一样协同。专业化的收益会抵消掉沟通损耗前提是职责边界划得足够清楚。那拆了之后收益到底能有多大我从自己的项目里拿了一个对比数据同一篇内容使用单Agent一次性生成事实性错误大约每千字会出现1-3处使用多Agent协作写作Agent产出、事实核查Agent逐条核验、合规Agent扫描、编辑Agent终审之后错误率降到了每千字0.5处以下。工程里面实体URL、数据来源、代码示例都不需要人工核对。这个提升不是我调了某一个超参得到的而是结构带来的。2. 多Agent编排的四种典型架构以及它们各自适合的舞台多Agent不是只有一个fixed pattern。根据任务依赖方式不同我梳理了四种常用的编排架构实践中90%的项目跑不出这个范围。理解这四种模式做技术选型时才不会拍脑袋。2.1 主编-记者模式Orchestrator-Worker这是最经典也最好理解的一种一个统筹规划的主控Agent下面挂若干个专职执行的Worker Agent。主控负责把大任务拆解成子任务、派发给合适的执行Agent、收集结果、处理异常。Worker之间原则上不直接通信所有沟通都通过主控中转。我那个内容审校管线就是这个模式主编AgentOrchestrator负责整体流程写手Agent、代码核查Agent、事实核查Agent、合规审查Agent各管一段。主控持有任务状态Worker只认自己的KPI。这个模式的好处是结构清晰、出错后容易定位缺点是主控可能成为瓶颈而且主控理解错了任务下级再能干也是白干。2.2 流水线模式Pipeline流水线就是把任务拆成一串有序节点A的输出是B的输入像工厂流水线一样。适合任务步骤天然有先后顺序的场景比如需求分析 - 架构设计 - 代码生成 - 代码评审 - 测试生成。流水线的核心设计点在每个节点只关注它眼前那一段输入上下文窗口压力小每个节点可以把全部能力集中在当前环节。缺点也明显一个节点质量不好错误会向下游持续放大必须给关键节点设计质检门禁。我在实际项目里会在流水线中间加入质量闸门Agent专门检查上个节点的输出是否达标再放行这不属于业务节点但作用非常大。2.3 辩论/评审模式Debate/Review让多个Agent从不同立场审视同一个问题最后经过辩论或者投票收敛出一个结果。典型场景是决策评审、文案质量评估、代码安全审计。比如让三位Agent分别扮演乐观方风险方用户体验方对一份需求文档提出意见再由仲裁Agent汇总。这种模式能够暴露单视角盲区代价是Token消耗大、响应时间变长。控制好轮数和最小集合很重要不然就是无休止的辩论。2.4 层级组织模式Hierarchical当Worker数量太多比如超过10个主编一个人管不过来时就需要中间管理层。顶级主编管几个组长Agent组长Agent各自管一组Worker。这种设计适合大型系统比如一个模拟公司运营的多Agent系统或者覆盖多业务线的客服体系。层级模式我只有在节点数超过15个时才推荐节点少的时候上管理层纯粹是增加延迟。我把这四种模式放在一起做一个对比方便你快速判断架构模式核心特征适合场景主要风险主编-记者中央调度Worker只干不说子任务相对独立、需要统一协调主控单点瓶颈主控理解偏差流水线上家产出喂给下家步骤天然有序、环形依赖少错误向下游放大调试定位难辩论/评审多角色交叉审视收敛结果质量评估、策划评审、安全审计Token消耗大可能不收敛层级组织多层管理分组协作节点数量多、业务线复杂管理成本高链路变长风险变大实际项目里这几种模式经常混用。我自己的管线就是个典型主线是流水线审校阶段临时切换成辩论模式做多Agent交叉审校最后再由主编Agent汇总裁决。3. 协作机制设计上下文互通、任务编排与结果收敛的细节架构定好了接下来是让多个Agent真正协作起来的三个关键机制。这三个机制做不好多Agent就只是多个Agent各干各的跟并行调用API没什么本质区别。3.1 上下文互通从传大段Prompt到共享工作台多Agent协作的第一个坑就是试图把一个Agent的全部输出上下文直接塞给另一个Agent。结果是上下文重复膨胀token消耗爆炸而且无用信息稀释了核心信息。我的做法是引入**共享黑板Shared Blackboard**模式所有Agent共用一个结构化的任务状态空间每个Agent只往黑板上写入自己领域的关键结果读取时也只看自己关心的键值。比如说事实核查Agent它不需要读全文它只需要从黑板上读到待核查句子列表和参考来源列表输出核查结论后写到核查结果键上。分工明确上下文友好信息不但没有丢失反而比大段文本传递更结构化。3.2 任务编排DAG是一切的基础但别做成无限循环任务编排的本质是解决哪个Agent先跑、哪些Agent可以并行、哪个Agent失败了怎么处理。我强烈建议用**有向无环图DAG**来建模任务依赖而不是自己写一堆干巴巴的if-else。把每个Agent看作一个节点依赖关系看作有向边节点入度为零时就可以执行。以我的审校管线为例任务依赖关系是这样的writer无依赖 → code_checker依赖writer → fact_checker依赖writer → compliance_checker依赖code_checker fact_checker → editor_review依赖compliance_checker在这个DAG里code_checker和fact_checker在writer完成之后是天然可以并行执行的如果我用一个串行Pipeline写就会浪费近一半的墙上时间。所以DAG模型不只是为了好看它是实打实的效率优化。关于失败重试我的原则是两类确定性失败不重试比如输入格式错误、Agent返回了空结果直接走降级方案不确定性失败可以重试比如LLM本次生成内容质量不达标、超时重试次数上限2-3次。重试超过上限就必须报错不能无脑循环。3.3 结果收敛多Agent争论完了怎么拍板辩论模式最容易出现永不收敛的问题——每个Agent都有自己的立场谁也不服谁。结果收敛的关键是设计一套仲裁协议。先说最简单的投票协议针对一个候选方案让每个评审Agent打质量分所有人打分平均值超过阈值方案通过否则把差评意见汇总打回重做。这是比较民主的机制适合评估类任务。还有一种是主编裁决协议评审Agent们只负责提意见最终决定权在主控Agent手里。主控给每个评审角色一票但允许最终仲裁人行使一票否决。这种做法快手、可控适合时间敏感的场景。收敛后还有一个容易被忽视的动作——明确关闭任务。我知道很多半吊子的编排示例跑完一轮就结束了但实际上应该把最终结果、评审轨迹、各Agent中间产物一并写回状态空间这样既能供后续复盘也能让编排系统知道这个任务已经终态避免某次重试又把它拉回活状态。4. 一个可直接复现的多Agent编排示例技术文章自动审校管线理论说完了拿点实际的东西出来。下面这个示例就是一个完整可跑的多Agent编排示例我用纯Python实现不依赖任何重型框架保证你能在一个小时内跑通并理解全部细节。它做的事情是模拟一篇技术文章的生成与审校过程。4.1 先定义协作基础结构消息、黑板上墙、Agent基类第一步搭好地基。我定义一个AgentMessage作为Agent间传递的消息结构一个SharedBlackboard作为共享状态空间以及一个BaseAgent基类强制每个子Agent实现process方法。from dataclasses import dataclass, field from typing import Dict, List, Any, Optional import json, time dataclass class AgentMessage: sender: str receiver: str content: str meta: Dict[str, Any] field(default_factorydict) class SharedBlackboard: 共享黑板所有Agent读写公共状态的地方 def __init__(self): self._data: Dict[str, Any] {} def write(self, key: str, value: Any): self._data[key] value def read(self, key: str, defaultNone): return self._data.get(key, default) def append(self, key: str, item: Any): self._data.setdefault(key, []).append(item) def snapshot(self) - Dict[str, Any]: return json.loads(json.dumps(self._data, defaultstr)) class BaseAgent: name: str base def __init__(self, temperature: float 0.3): self.temperature temperature def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: raise NotImplementedError这里有一个实际工程中的细节黑板的write和read非常简单但append方法是我后来补充的。因为在审校过程中检视者会不断把已发现问题追加到列表里而不是覆盖某个键。这种追加语义在多Agent场景下非常常见。4.2 实现四类专职Agent写作、代码校验、事实核查、合规审查接着实现具体的Agent。为了让你专注理解协作机制我把LLM部分用模拟函数替代——每类Agent的输出可以用一段带有固定语义的字符串模拟比如模拟判断函数名正确、变量类型匹配。等你接入真实大模型时只需要替换这一层。class WriterAgent(BaseAgent): 写作Agent产出文章初稿并把待检点写入黑板 name writer def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: # 真实场景: 调用 chat.completions.create(...)把完成草稿写回 draft 本文介绍了异步任务队列入门示例代码使用 Python 和 Redis。 code_snippet import redis\nr redis.Redis(hostlocalhost)\n board.write(draft, draft) board.write(code_snippet, code_snippet) board.write(claims_to_check, [ Redis 默认端口是 6379, 关于异步队列任务积压时应该优先扩容消费者, ]) return AgentMessage(senderself.name, receiverorchestrator, contentf初稿完成共{len(draft)}字, meta{draft_length: len(draft)}) class CodeCheckerAgent(BaseAgent): 代码校验Agent检查代码块是否有明显问题 name code_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: code board.read(code_snippet, ) # 真实场景: 让LLM执行静态审查同时可以接 pylint/mypy 等确定性工具 issue self._fake_scan(code) if issue: board.append(issues, {agent: self.name, issue: issue}) status pass if not issue else fail board.write(code_check_status, status) return AgentMessage(senderself.name, receiverorchestrator, contentf代码校验完成: {status}, meta{status: status}) def _fake_scan(self, code: str) - Optional[str]: if password in code: return 代码中疑似包含明文口令 if len(code.strip()) 0: return 代码块内容为空 return None class FactCheckerAgent(BaseAgent): 事实核查Agent逐条核对文章中的数据与来源 name fact_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: claims board.read(claims_to_check, []) # 真实场景: 使用联网搜索工具检索并比对可信数据库 for claim in claims: if 6379 in claim: board.append(verified_claims, {claim: claim, result: true}) else: board.append(verified_claims, {claim: claim, result: uncertain}) board.append(issues, {agent: self.name, issue: f无法核实: {claim}}) board.write(fact_check_status, done) return AgentMessage(senderself.name, receiverorchestrator, contentf事实核查完成共核查{len(claims)}条) class ComplianceAgent(BaseAgent): 合规审查Agent检查内容是否符合发布规范 name compliance_checker def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: draft board.read(draft, ) sensitive_words [危险词A, 限用词B] # 真实场景: 维护一个专业词库 found [w for w in sensitive_words if w in draft] if found: board.append(issues, {agent: self.name, issue: f敏感词: {found}}) board.write(compliance_status, pass if not found else fail) return AgentMessage(senderself.name, receiverorchestrator, contentf合规审查完成命中敏感词{len(found)}个)这段代码的核心思路大家注意一下每个Agent在process方法里只有三件事——从黑板上取自己要的输入、做自己的领域判断、把结果写回黑板或者在黑板的问题列表里追加记录。Agent之间没有直接的方法调用这就是协作和串行Pipeline的本质区别。4.3 主编Agent与DAG调度让任务按依赖跑起来紧跟着是编排的核心主编Agent和调度器。我的主编Agent职责是发起任务、接收所有执行Agent的状态、决定是否进入终审。而调度器负责解析DAG计算出每个节点何时可以被执行。from collections import deque class OrchestratorAgent(BaseAgent): name orchestrator def process(self, msg: AgentMessage, board: SharedBlackboard) - AgentMessage: issues board.read(issues, []) if issues: return AgentMessage(senderself.name, receivereditor_review, content存在待处理问题请结合评审意见终审, meta{issue_count: len(issues)}) return AgentMessage(senderself.name, receivereditor_review, content全部检查通过进入终审, meta{issue_count: 0}) class GraphScheduler: 基于DAG的任务调度器统计每个节点的入度并依次执行 def __init__(self, agents: Dict[str, BaseAgent], graph: Dict[str, List[str]]): self.agents agents self.graph graph self.indegree {node: 0 for node in graph} self.children {node: [] for node in graph} for node in graph: for dep in graph[node]: self.indegree[node] 1 self.children[dep].append(node) def run(self, board: SharedBlackboard): # 按入度为零动态推进为保证可复现性这里先用广度优先执行 # 在生产环境配合 concurrent.futures 就可以实现真正并行 queue deque([node for node, deg in self.indegree.items() if deg 0]) executed [] while queue: node queue.popleft() agent self.agents[node] msg AgentMessage(senderscheduler, receivernode, contentstart) result agent.process(msg, board) print(f[{node}] - {result.content}) executed.append(node) for child in self.children[node]: self.indegree[child] - 1 if self.indegree[child] 0 and all(dep in executed for dep in self.graph[child]): queue.append(child) return executed4.4 组装跑通把全链路一条命令串起来最后把这些零部件组装成一个完整的main函数。调度器按DAG执行完所有业务节点后把最终数据统一交给终审模块生成交付报告。def main(): agents { writer: WriterAgent(temperature0.7), code_checker: CodeCheckerAgent(temperature0.2), fact_checker: FactCheckerAgent(temperature0.2), compliance_checker: ComplianceAgent(temperature0.2), orchestrator: OrchestratorAgent(temperature0.3), } graph { writer: [], code_checker: [writer], fact_checker: [writer], compliance_checker: [code_checker, fact_checker], orchestrator: [compliance_checker], } board SharedBlackboard() scheduler GraphScheduler(agents, graph) executed scheduler.run(board) # 终审输出审校报告 report { executed_agents: executed, issues: board.read(issues, []), verified_claims: board.read(verified_claims, []), final_decision: rework if board.read(issues, []) else publish, } print(json.dumps(report, ensure_asciiFalse, indent2)) if __name__ __main__: main()跑这个main()你会看到类似这样的输出[writer] - 初稿完成共40字 [code_checker] - 代码校验完成: pass [fact_checker] - 事实核查完成共核查2条 [compliance_checker] - 合规审查完成命中敏感词0个 [orchestrator] - 全部检查通过进入终审然后看终审报告你会发现issues列表为空但verified_claims里有两条核验记录final_decision是publish。这就完整走通了一个多Agent协作闭环。如果你要接入真实大模型有几个位置替换一下即可WriterAgent的draft赋值处改成调用LLMAPI并解析结构化输出FactChecker的claimed核对改成工具调用比如联网检索Compliance的词库改成加载你自己的敏感词表。其余协作逻辑一行都不用动。5. 别急着上框架自研轻量编排与成熟框架的选型判断标准拿到上面的示例之后很多人会问我现在直接上LangGraph、AutoGen或者CrewAI行不行我的观点是如果你的场景里Agent数量不超过10个、流程相对固定自研这个轻量编排反而更香。原因有三点。第一成熟框架的抽象层很厚。LangGraph这类框架为了方便你快速搭图设计了专门的State、Node、Edge概念学习成本不低。你真去读了它的文档把它掰开来用会发现顶层API已经在悄悄替你做很多决策。而你做的多Agent协作项目如果自定义逻辑很强框架提供的默认行为往往变成阻碍你花在摆脱框架上的时间比你自己写还多。第二可控性很关键。自研版每行代码你都知道在干什么出问题直接断点调。框架版本里一旦出现回环重试失控状态更新被静默覆盖排查成本会让你怀疑人生。特别在做生产级管线时可观测性和可控性对稳定性的贡献往往比框架的开箱即用重要得多。第三轻量编排迭代速度更快。框架的发布周期、API变动都不由你控制。我自己早期用AutoGen跑过一个小Demo一堆魔法函数和隐式上下文跑通了翩然自喜等到要往里加企业安全审查逻辑的时候几乎等于重写。反而是自研这套几十行DAG调度器改起来得心应手。那什么情况下该用框架呢我的判断标准是单Agent能力已经不够用而你也暂时没有想清楚协作细节先用框架的High-Level API快速验证多Agent的可行性与收益点再决定是否自研。比如你只是想在一天之内体验一下多Agent辩论是什么感觉那AutoGen的Group Chat几乎三五行代码就能搞定。Tool-Use类任务用LangGraph因为它有完整的Tool调用生态。还有一个常见误区是用了框架就自然实现了多Agent协作。框架只是管道的骨架它不替你解决上下文互通怎么设计、结果怎么收敛、失败怎么降级这些真正的工程问题。你以为换框架能解决协作难题结果只是换了个地方踩同一个坑。自研轻量编排的正确打开方式用到一定程度你需要的是从里面长出自己的通用能力——消息总线、状态快照、重试策略、审计日志。这些在自研代码里就是四个函数的事在框架里则要跟框架的抽象做斗争。所以我的建议是双轨并行小规模探索可以用开源框架快速验证一旦确认方案要进生产就把关键路径改造成自己的轻量调度同时保留框架作为可选的后端。这样既有框架带来的开发速度又有自研带来的确定性。6. 多Agent落地时我踩过的坑以及对应的补救方案最后这部分全是真金白银的踩坑记录。我把最常让新手栽跟头、也最影响实际效果的问题和解决方案整理成一张对照表然后挑几个典型展开讲。常见问题根本原因补救方案上下文爆炸Token消耗不可控每个Agent都继承全部对话历史改共享黑板模式Agent只读自己关心的键Agent之间格式不统一不同Agent输出markdown、JSON、纯文本混杂统一结构体如AgentMessage附带meta字段与协议版本任务反复横跳不收敛审查Agent发现问题就打回没有终审仲裁设置仲裁协议限定重试上限终审Agent拥有拍板权单个Agent失败拖垮全局没有处理Agent级故障的隔离策略超时熔断降级输出可重跑子任务编排Demo能跑生产一换真实LLM就崩模拟函数掩盖了LLM输出的随机性先垫一层输出解析与校验失败时强制重试6.1 上下文爆炸从全量传到按需读我最早做的版本是把上一个Agent的全量输出拼成一个新Prompt丢给下一个Agent结果测试时token消耗飙升到原来的五倍而且下游Agent经常被无关文本干扰输出质量不升反降。后来我彻底抛弃了大文本传递改成了黑板式按需读写。实践下来token消耗能降到原来的两成左右而且各Agent输出的稳定性明显提高。这个改动是整个项目里性价比最高的一次重构。6.2 格式混乱问题的解法多个Agent协作最烦的就是A输出的JSON丢了字段、B输出的markdown里混了代码块C直接把表格用普通文本画了。我的做法是三层加固第一层AgentMessage的meta字段强制要求sender、receiver、version等信息第二层每个Agent的输出必须有明确的schema声明比如返回JSON包含status与detail两个字段第三层在调度器里加一个轻量的格式校验函数解析不通过直接判失败并触发重试。这样能在第一时间发现问题而不是等到终审才发现一堆脏数据。6.3 任务不收敛和无休止循环有次做评审型多Agent三个评审Agent对一份需求文档的意见完全对立主编Agent又特别民主导致同一个文档被打回重写了五轮时间和token全花在来回扯皮上。之后我加了两条硬规则评审Agent只提两轮意见第三轮起只允许同意或弃权主编终审有一票否决权可以拿着不完全满意但风险可控的结果直接放行。设置了这个收敛规则后类似任务的完成时间从原来的十二分钟压缩到了四分钟。6.4 单点失败的雪崩效应流水线模式下第二个节点崩了后面所有节点都跟着白干。我后来在每个节点外面套了隔离舱节点执行前设置超时超时直接记录失败并走fallback失败节点向主编汇报时携带可重跑标记由主编决定是从失败节点重做还是整条链路重来。加上这个机制之后管线对单个Agent抽风的容忍度大幅提升再也不需要整条重跑。实话实说多Agent项目里商量和争执都不难难的是让这个系统长期稳定地维持产出质量。我现在的习惯是每次跑完一个多Agent任务一定会去把黑板的snapshot、每个Agent的输出轨迹、终审决定存下来做一次对账。就是这个习惯帮我把很多偶发的质量问题提前盯了出来而不是等到用户反馈了才四处排查。内容管线的多Agent编排跑通只是开始真正的功夫在于监控它每一轮是怎么协作、怎么争议、怎么收敛的。如果你正在推动一个单Agent转多Agent的项目先把这些机制想明白、把坑提前避掉能省下的时间一定会让你觉得值。