今年上半年我干了件大事把散落在各处的多个 AI Agent 统一收编到一个叫 OpenRig 的编排层里让它们从“各自为战”变成一套能持续运转的协作系统。项目名字里的 Rig 其实就是“装配台”的意思——一台能把零件固定住、方便你组合出复杂结构的架子。一开始我手上有七八个独立 Agent有做行业情报摘要的有写周报的有跑代码审查的还有半夜执行数据脚本的运维 Agent。单独看每个都挺能打但我很快发现一个尴尬的事实单点 Agent 跑得再好拼不成一个业务闭环。比如“生成一份风险分析报告”这个任务需要情报 Agent 先查资料、分析 Agent 建模型、审查 Agent 复核合规性、最后执行 Agent 渲染出文档。每次转换都靠我手动搬运中间产物稍有延迟或报错整条链路就得从头再来。真正的多智能体编排不是把几个模型串在一条流水线上而是让它们之间产生“可恢复的协作”。这篇文章写的就是我在 OpenRig 项目里的完整实操怎么定义编排原语、怎么设计 Agent 的接入契约、怎么处理并发和死锁、怎么把上下文和状态持久化到系统底层。内容偏工程向但我会尽量把每一步的“为什么”也讲清楚。如果你正在做一个多 Agent 项目或者正被“Agent 之间上下文丢失、状态回滚、重试风暴”这类问题折磨这篇文章应该能帮你省下不少弯路。1. 背景为什么单点 Agent 跑得再好也成不了系统1.1 离散 Agent 的真实处境先说一个比较反直觉的事实多数团队做 AI Agent 产品做完第一个“爆款 Agent”之后往往会陷入一种虚假繁荣。单个 Agent 的效果看起来不错但所有业务逻辑都堆在一个提示词里你要它查资料它有这个能力你要它写代码它也能写你要它做二次复核它硬着头皮也能干。结果就是 Prompt 越来越长、错误率越来越高、迭代越来越难。OpenRig 想解决的问题就是这个与其把一个 Agent 撑成超人不如让多个“窄能力 Agent”像齿轮一样咬合起来。齿轮单独看都很简单但组合起来可以完成复杂传动。所以整个项目的第一阶段不是写编排器而是做“存量 Agent 盘点”把每个 Agent 负责的领域、输入输出、延迟、失败模式全部列出来。这一步看似琐碎却是后面所有设计的基石。我当时盘点出来四类典型 Agent情报型负责搜索、摘要、信息抽取输出结构化文档。分析型负责建模、计算、预测输出JSON结果。执行型负责调API、跑脚本、写数据库输出操作回执。复核型负责审查前面所有Agent的输出质量、合规性、一致性输出审批结论或返工意见。这些 Agent 有的跑在 FastAPI 服务里有的挂在 LangGraph 图上有的是纯函数包成 HTTP 接口。技术栈完全不同但它们的共同点是一致的只要放到一起跑端到端流程就一定会遇到状态和上下文的问题。1.2 协作崩溃的三种现场多 Agent 协作的坑我在 OpenRig 前期踩过不少概括起来基本就是三类状态互相覆盖。两个 Agent 同时往同一个任务里写字段后写的把先写的覆盖了。比如分析 Agent 刚把“预测模型版本”写进任务记录执行 Agent 又把这个字段改成了“脚本路径”最后系统既不知道模型版本也不知道脚本路径。这类问题在分布式系统里很常见但在 Agent 场景更容易被忽略因为大家默认“Agent 很聪明应该能处理”。上下文随进程丢失。Agent 服务一重启内存里的临时上下文全部清空。如果编排器没有把上下文落盘那么中断之后整个任务只能从头再跑。有一次我们的运维 Agent 在执行数据刷新时被杀掉重启后它对“当前刷到第几批、哪些批次已经成功”一无所知结果把整张表重刷了一遍生产数据被搞出脏数据。这就是没有持久化状态导致的典型事故。盲目重试的雪崩。Agent 调用失败后最常见的第一反应就是重试。但如果编排器把“重试”做成对整个子任务的重跑而那个子任务内部又调用了外部付费 API就会造成重复扣费、重复写库。更可怕的是在多个 Agent 并发失败时重试风暴会瞬间把下游服务打满。这三类问题让我意识到多智能体编排的核心不是把 Agent 们“连接起来”而是把协作过程的状态和记忆变成一个可以被持久化、被恢复、被审计的系统。所以 OpenRig 从一开始就定了一个原则——Agent 可以无状态但编排器必须有状态对话可以丢但系统记录不能丢。2. OpenRig 的顶层抽象对话、工作流、持久化系统的三层跃迁很多人看到 LangGraph、Autogen 或者 n8n 这类工具第一反应是“拿它写个工作流编排”。但说实话工作流Workflow和协作系统Collaboration System是两码事。工作流解决的是“一条任务如何按预定义路径跑完”协作系统解决的是“多角色在长时间内如何共享状态、留下记录、不断收敛”。工作流跑完就结束了协作系统跑完还要回答谁在什么时候做了什么现在已经收敛到什么程度如果中断了怎么续上OpenRig 没有另起炉灶再造一个 workflow 引擎而是选择在成熟框架之上加一层“状态与记忆层”。这一层的抽象被刻意保持得非常少只有三个原语。2.1 三个核心原语节点、边、记忆第一个原语是节点Node。节点包装一个实际能干的 Agent 或工具包含三部分agent_id、能力清单、运行时策略。能力清单用来让编排器知道“这件事谁能做”运行时策略定义超时、并发、失败模式。节点不关心内部是 LangChain 还是裸调 OpenAI只要它对外暴露统一的执行接口。第二个原语是边Edge。边不是一段简单的箭头而是一条“受控关系”。它定义了消息从A流向B时哪些字段合并、哪些字段丢弃、在什么条件下才允许流动。为什么我要把边设计得这么重原因很简单如果让 Agent 自己决定“下一步找谁”那协作过程基本不可控也无法预测。把控制权收回到边定义上协作系统才能有确定性。第三个原语是记忆Memory。每个节点可以声明自己要读哪些记忆、写哪些记忆。记忆是分层的、持久化的不会因为 Agent 实例死亡而消失。后面我会专门讲记忆怎么做分层。这三个原语拼在一起就够覆盖绝大多数真实场景。举个例子一个“日报生成任务”由情报 Agent 和复核 Agent 协作完成。情报 Agent 把摘要写入“task:123:raw_material”复核 Agent 读取后把结论写入“task:123:approved_version”。这两个记忆区域绑定在同一个任务 ID 上天然防止了跨任务混淆。2.2 持久化的分层设计记忆系统如果只有一层很快会被长期运行的任务撑爆。OpenRig 把持久化分成了五层每一层的作用域、生命周期、存储介质都不一样层级作用域生命周期典型存储例子工作记忆单次执行步骤步骤结束即清理Redis / 内存当前步骤的输入输出任务状态单个业务任务从创建到关闭PostgreSQL / SQLite任务计划、当前阶段、Checkpoint项目记忆一个项目或租户长期保留PostgreSQL 向量库项目决策、历史方案、领域知识实体记忆某个用户/实体长期保留向量库 关系库用户偏好、实体档案审计日志整个协作系统永久对象存储 / 追加表全部执行记录、错误、回放轨迹为什么“对话”必须和“系统记忆”分开因为对话只是一条线性时间线上的事件流它受限于模型上下文窗口而系统记忆是结构化的、可以被检索和回溯的状态。我记得早期版本里我天真地把每轮 Agent 之间的完整对话存成一份大 JSON放在任务记录里后来任务一复杂这个 JSON 长到几十万字符模型根本读不进去检索也慢。后来才改成对话只保留“关键结论”其余细节进审计日志。只有状态也就是当前最重要的任务进度和结果才放进任务状态层。3. 从 0 到 1 搭出编排骨架注册契约、路由分发与执行循环3.1 Agent 接入协议用契约代替“私有接口”OpenRig 的 Agent 接入靠的是一份显式契约而不是把每个 Agent 的函数直接 import 进来。每个 Agent 在接入时提供一个 manifest声明自己能干什么、不能干什么、超时多久、输出长什么样{ agent_id: code_reviewer, version: 0.4.2, capabilities: [review_code, suggest_patch], runtime: { type: http_service, endpoint: http://agent-review:8000/run, timeout_seconds: 30, concurrency_limit: 4 }, input_schema: { type: object, properties: { code: {type: string}, language: {type: string} } }, output_schema: { type: object, properties: { verdict: {type: string, enum: [approve, reject, amend]}, issues: {type: array} } } }对应的 Python 端点是class AgentWorker(Protocol): agent_id: str capabilities: list[str] async def run(self, payload: dict, context: ExecutionContext) - AgentResult: payload 是编排器下发的任务参数context 里只有该 Agent 被允许读到的记忆切片。为什么要这么设计因为契约能让编排器做动态路由当一个新的调研任务进来编排器知道“信息抽取”这件事有两个 Agent 可以做于是根据当前负载和分数择优选择。如果某个 Agent 挂了能力相同的备用 Agent 可以无缝顶上。这个能力在单 Agent 项目里用不上但在多 Agent 协作项目里就是核心竞争力。3.2 编排器主循环Plan-Dispatch-Checkpoint-ContinueOpenRig 的编排器核心是一段循环逻辑很像游戏的“帧循环”不断检查图状态找出当前能执行的节点执行它们再把结果写回状态。我把它简称为 PDCC 循环async def orchestrate(plan: Plan, graph: StateGraph, store: StateStore): while not graph.is_finished(): # 1. 找出当前不依赖未完成节点、且满足前置条件的节点 batch graph.ready_nodes(limitbudget) # 2. 所有节点共享同一份快照化的任务上下文 for node in batch: snapshot store.load_context(node, scopenode.memory_scope) future executor.run_with_checkpoint(node, snapshot) batch_futures.append(future) # 3. 等待批量结果返回并发执行 results await asyncio.gather(*batch_futures, return_exceptionsTrue) # 4. 每个结果都先写入 checkpoint再更新状态图 for node, result in zip(batch, results): if isinstance(result, Exception): handle_failure(node, result) else: store.save_checkpoint(node, result) graph.apply(node, result) # 5. 根据新状态推进图进入下一轮 graph.continue_from(statestore.current_state()) return store.summarize(plan)这里最容易踩的坑是“快照上下文”和“完整上下文”的区分。早期版本里我把整个任务的所有记忆都塞给每个 Agent结果 Agent 反而不知道该看哪条输出质量大幅下降。后来参考了人的协作方式——开会时只带相关材料其他东西放档案室——所以每个节点只加载它声明需要的那几个记忆片段而不是整个任务。Checkpoint 也不是什么高级东西它就是把“节点执行前状态节点执行后结果”完整记录到数据库里一条记录。但真正让协作系统变可靠的恰恰是这些每步都写的 Checkpoint。后面第 4 节我会展开讲。3.3 并发执行Agent 长耗时下的并行与兜底“AI Agent 怎么扛并发”是最近很火的问题。我的答案可能有点反常识Agent 场景的并发核心不是把 QPS 打满而是把有限的预算花在刀刃上同时不互相拖死。普通 API 请求只要几十毫秒Agent 调用动辄几十秒甚至几分钟。如果编排器同步地“调完 A 再调 B”一个五步链路就是 5 倍的延迟。所以 OpenRig 把没有依赖关系的节点全部并行跑。常用的控制工具是信号量from asyncio import Semaphore sem Semaphore(valueagent_concurrency) async def run_node_with_limit(node): async with sem: return await node.run()此外还要注意“扇出/扇入”fan-out/fan-in模式当编排器不确定哪个 Agent 回答得更好时可以把同一任务分发给两个候选 Agent 并行执行等两者都返回后再让一个聚合器选择或融合结果。这样做确实能显著提升复杂任务的完成度但代价是 Token 费用翻倍。我的策略是只对高影响节点启用扇出比如最后的合规审核、面向客户输出的内容生成其他的普通节点一律单选。并发场景还有一个大坑重复副作用。如果 Agent 在执行中调了外部接口、写了数据库你因为超时重试它就执行了两次。所以给每个执行步骤生成唯一的 execution_id下游所有副作用操作都必须带这个 ID 实现幂等。这一步写在系统设计里很容易但实际落地时大部分 Agent 内部代码根本没考虑幂等我的做法是先在接入契约里强制要求所有有写操作的 Agent必须支持以 execution_id 为条件去重。4. 让协作真正可收敛状态恢复、幂等重试与超时治理4.1 Checkpoint 与幂等执行持久化协作系统最迷人的一点是它可以在错误之后“续跑”而不是“重跑”。要支持续跑Checkpoint 必须回答三个问题跑到哪一步了这些步的结果是什么哪些步骤还没开始我在 OpenRig 里的 checkpoint 表大概长这样字段说明checkpoint_id全局唯一UUIDgraph_id图实例 ID对应一次业务任务node_id节点标识input_hash输入内容的哈希output_json输出结果全文statuspending / running / success / failed / skippedcreated_at记录创建时间恢复逻辑很简单从任务的最新 checkpoint 开始只执行那些 status ! success 的节点。为了防止重试把成功节点再跑一遍每次执行前先算 input_hash如果这张表里已经有同样 input_hash 且 statussuccess 的记录就直接跳过。用 input_hash 做幂等看起来简单但有个细节容易翻车字符串哈希碰撞概率极低但两个语义相同、格式不同的输入比如 JSON 字段顺序不同会被当成不同输入。所以我在存 input_hash 之前会把输入先做一次 canonical serialization把 JSON 的 key 排序、压缩空白再取哈希。这个细节帮我省了不少不必要的重跑。4.2 超时、限流与降级策略多智能体系统的异常处理不能只靠“连续重试三次”这种粗暴逻辑。OpenRig 里我把故障策略分成四类分别应对不同场景故障类型处理策略预期行为单次调用超时超过节点 timeout_seconds允许重试一次重试仍超时则降级到备用 Agent任务不中断连续失败同一节点失败达到 3 次停止该节点进入人工审批通道发出警报上下文完整保留限流 429 / Token 配额耗尽指数退避后入队不硬试队列等待时间可见可查上下文超限触发压缩/摘要节点先整理再继续后续 Agent 拿到精简版而非垃圾长文关键是要把“超时”和“失败”分清楚。超时不一定代表节点死了可能只是模型慢了一点给一次机会是合理的但连续失败则说明这个 Agent 大概率有 bug 或当前输入有问题这时候再重试就是烧钱。OpenRig 里每个 Agent 节点都配了失败计数器和冷却时间进入“冷却态”的节点不会再被调度直到人工介入或下一个版本上线。4.3 一个真实排查案例从互相等待到死锁说一个我印象最深的坑。那是一个“文献调研-撰写建议”双 Agent 协作任务设计思路是调研 Agent 负责找资料并输出初步分析建议 Agent 负责写建议然后调研 Agent 再看一遍建议是否有遗漏。听起来很合理对吧实际跑起来任务会在第三个来回之后卡死。打开监控面板两个 Agent 都显示“running”但没有任何日志输出。CPU 几乎为零数据库连接数却越涨越高。我当时的排查链路是这样的导出状态图把当前图实例的所有节点状态打印出来发现调研 Agent 和建议 Agent 之间出现了一个 A-B-A 的循环依赖而且循环里没有一个“终止条件”可控变量。查看待处理队列建议 Agent 还在等调研 Agent 的新证据而调研 Agent 在等建议 Agent 的反馈两边谁也等不到谁。锁定根因问题出在边的定义上。当时我把“建议 Agent 返回意见”和“调研 Agent 补充材料”放到了同一个边条件里但没有把“任务状态”作为边的守卫。结果两个 Agent 都认为对方应该先动形成了典型的死锁协作。修复方式很朴素在边守卫里增加状态变量 current_round并且限制最大轮数 max_round3当轮数达到上限时强制跳转到“汇总节点”而不是继续循环。同时把两个 Agent 之间那种“互相等待对方结果”的模式改成“写共享记忆区域 订阅通知”的模式谁先把结果写进记忆区谁就先触发下一个节点而不是彼此同步调用。从那以后我总结了一条经验多智能体编排里循环不是天然邪恶的但循环必须有显式的退出条件和防死循环计数。你可以在设计阶段画图觉得很优雅但到了生产环境没有退出条件的协作循环就是一颗定时炸弹。5. 持久化是中台底座上下文窗口管理、存储选型与事件溯源5.1 上下文窗口是编排器最大的敌人做多智能体持久化真正的大 Boss 不是数据库而是模型的上下文窗口。试想一下一次复杂的跨 Agent 协作每个 Agent 都往共享记忆里写内容十几个步骤下来记忆区里可能有几十段资料。如果你在下一次调用时把所有内容全部塞进 Prompt那么Token 费用会指数级上升模型会因为信息过载而忽略真正重要的字段响应延迟也会越来越难控制。OpenRig 的处理办法是“上下文分层供给”步骤内上下文当前节点输入、最近一次上游输出这些直接塞给 Agent。任务级上下文当前的计划、已完成步骤列表、当前状态摘要这些以精简摘要形式注入。项目级上下文领域知识、历史相似案例这些只按需检索每次最多注入 3-5 条。我还专门实现了一个“压缩节点”当某个分支的完整历史超过 50 步时自动调用摘要 Agent 生成每 10 步一节的压缩摘要再用压缩摘要替换原始明细进入后续上下文。原始明细并不是丢弃而是存档到审计日志方便回溯。有人可能会说那直接用长上下文模型不就行了如果预算充足这是一个选项。但长上下文模型在处理几十万 Token 时仍然存在“大海捞针”式的注意力偏移而且费用确实高。所以工程化的做法永远都是给模型最精简的、最相关的信息而不是给它全部信息。5.2 存储选型关系库、队列、向量库、对象存储的分工既然要持久化协作系统存储选型就不能随便拍脑袋。我的分工很明确数据推荐存储原因任务状态、Checkpoint、图状态PostgreSQL / SQLite事务支持强恢复逻辑必须一致待执行任务、消息通知Redis Stream / RabbitMQ天然支持 ack、重投递、队列长度监控实体记忆、语义检索pgvector / FAISS做相似召回不用来作为唯一状态源Agent 产出的文件、日志、审计证据对象存储S3/OSS/MinIO大对象不适合放数据库且需要生命周期管理不可变事件流追加表 / 对象存储事件溯源必须 append-only这里最想强调的一点是不要让向量库承担系统状态职责。我见过不少团队把 agent 的“记忆”全部放向量库然后读取时直接 top-k 相似度检索。问题是向量检索的结果是“可能相关”不是“确定状态”。如果编排器依赖检索结果判断“任务有没有完成”那就会出大乱子。所以我的原则是状态放关系库知识放向量库两边各司其职。状态与知识需要联动时通过显式的引用 ID 去关联而不是把一整段记忆随机塞进嵌入向量里。5.3 事件溯源让协作过程可以被回放持久化协作系统还有一个隐藏需求审计与复盘。当客户说“为什么推荐报告里出现了一条过时数据”时你要能回答是哪个 Agent、在哪个时间点、基于什么上下文、产出了这条数据。OpenRig 的所有关键节点执行都会写一条不可变事件记录{ event_id: evt_123456, graph_id: task_888, node_id: data_fetcher, action: completed, checkpoint_id: ckpt_abc, input_ref: s3://openrig/inputs/abc.json, output_ref: s3://openrig/outputs/abc.json, agent_version: 0.4.2, ts: 2026-01-18T10:22:31Z }事件不存储完整值只存引用加摘要。需要时可以从对象存储取原始输入输出。事件流是 append-only 的不修改、不删除。这样即使线上代码出了 bug你也能把过去任意一个任务的执行轨迹回放出来定位问题到底发生在哪一步。这个习惯现在已经成为我所有 Agent 项目的标配成本不高但对排查问题帮助极大。6. 想复刻这套实践的人我给你的最小起步清单6.1 三步跑通最小多智能体系统如果你也想从 0 到 1 搭一个类似的编排系统我的建议是不要复制整套 OpenRig先跑通最小闭环第一步选三个 Agent 角色规划者把用户需求拆成步骤、执行者真正干活查数据或写代码、复核者检查执行结果并决定是继续、返工还是结束。三个角色足够覆盖 80% 的业务流程又不会让你在项目初期陷入状态图复杂度爆炸。第二步把接口契约写在所有代码之前。先定好 Agent 之间传递消息的 JSON Schema再各自实现。不要“先写 Agent 后补协议”那样很容易出现两个 Agent 互相传私有字段的情况。我在 OpenRig 项目里吃过这个亏后来补协议花了三倍时间。第三步用 SQLite 或 Postgres 直接落状态。不要一开始就上分布式存储、上 K8s。OpenRig 一开始用 Docker Compose 跑三个容器就已经足够复杂了。只要你的状态能持久化、能恢复再考虑扩并发和服务网格。6.2 用指标而不是直觉来优化编排多智能体系统优化如果没有指标基本就是盲人摸象。我建议至少跟踪下面五个数端到端任务完成率失败后恢复成功率每个任务的平均重试次数任务在排队/等待上的耗时占比每次任务的 Token 总消耗和成本。拿这些数做优化决策比凭感觉靠谱得多。比如我发现某个链路完成率只有 82%进一步看数据发现卡在“复核 Agent 认为执行结果不合格”这个节点上于是我不是去优化执行 Agent 的 Prompt而是调整了复核 Agent 的验收标准描述。只改了十几个词完成率就上到 94%。这种收益是“多上两个 Agent”完全比不了的。这套方法论下来我最深的感受是做多智能体编排不要把精力全部花在“让 Agent 更聪明”上而要把更多精力花在“让 Agent 之间的协作更可靠”上。聪明是单个模型的事可靠是系统的事。OpenRig 最大的价值不是那个调度循环而是它让每一次协作都留下了可以回放、可以恢复、可以被审计的持久化痕迹。到最后你会发现所谓“持久化协作系统”核心就是一件事无论中间发生多少次失败系统都清楚地知道自己在哪、要去哪、已经做过什么。想清楚这一点再回头看那些 Agent 框架和编排工具你的思路会清晰很多。