OpenRig多智能体编排：实现持久化状态与断点恢复的工程实践

最近聊 AI Agent 的朋友越来越多了但聊来聊去我发现大家都在同一个地方栽跟头——单个 Agent 跑通一个任务没问题两个、三个Agent一起协同时任务一拉长就开始乱套上下文接不上、目标漂移、任务跑到一半进程崩了就直接从头再来。说句实话2026年做 AI Agent 已经不是要不要做的问题而是怎么把它从“Demo”变成“系统”的问题。那些能在生产环境稳定跑起来的 Agent背后一定有一层东西在管状态、管流程、管恢复——这层东西就是多智能体编排。OpenRig 这个名字就是我给自己在实践里沉淀的一套编排方案起的代号它把离散的 AI Agent 编织成一个持久化协作系统重点解决三件事——状态不丢、流程不乱、坏了能恢复。这篇文章我不讲虚的直接把思路、架构、代码和踩过的坑都摊开来讲适合正在做 Agent 中台、Agent 产品化、或者单纯想把多智能体协作落地的朋友参考。1. 多智能体编排到底在解决什么问题1.1 单个 Agent 的天花板能力提上去了可靠性反而降下来先问一个问题你现在手上的 Agent是不是只在“单轮问答”和“低复杂度工具调用”上表现不错一旦你让它把一个长周期任务从上到下完整跑下来比如“分析用户需求 → 查询数据库 → 生成代码 → 自测 → 修复 bug → 输出报告”你就会发现模型本身的能力确实够但可靠性断崖式下跌。原因很简单。大模型本质上是无状态的函数它每次推理只看到当前窗口里的内容。长任务执行到第 5 步时前面第 1 步的中间结果还在不在上下文里工具返回的数据有没有被截断如果模型在第 4 步生成了错误结论能不能准确退回到第 3 步重新走单靠一个 Agent 的“暴力硬扛”这些环节全都不可控。尤其是模型上下文窗口再大也有限任务一长基本就是“记了后面忘了前面”。我自己实测下来单个 Agent 扛 3 步以内的任务成功率能到 90% 以上但跑到 7 步以上成功率会掉到 50% 以下。这个拐点出现得非常快这也是为什么现实项目里几乎所有人都在往“多 Agent 分工 编排”的方向走。1.2 离散 Agent 的三宗罪断档、漂移、无状态当你把任务拆给多个 Agent 之后又会冒出新问题。我把它总结成“离散 Agent 三宗罪”断档Agent A 干完活Agent B 接手时A 的上下文和产出物没有有效传递。B 必须重新理解任务、重新去拉数据效率低不说理解还可能偏差。漂移没有统一的全局状态每个 Agent 都在自己的“局部视野”里做决策。你让三个 Agent 分别研究三个竞品最后汇总时发现它们用了完全不同的分析维度根本合不到一起。无状态Agent 任务中途崩溃进程没了所有运行进度全部消失。更麻烦的是没有运行记录你根本不知道它死在哪一步、当时的输入是什么、下一步该干什么。这三点本质上指向同一个缺陷Agent 之间没有一套共享的、可持久化的协作协议。每个 Agent 都像一个各干各的自由人没有项目经理没有共享白板也没有存档机制。离散 Agent 的天花板就在这里——不是模型能力不够而是协作机制缺失。1.3 OpenRig 的解题目标让 Agent 从“一次性对话”变成“可恢复的长跑选手”OpenRig 这个名字想表达的核心概念就是“刚性骨架”——把柔性的模型能力固定在一套可编排的骨架里。它不追求把单个 Agent 调得更聪明而是把多个 Agent 的关系、状态、流转规则显性化。具体来说OpenRig 围绕三条原则设计状态显性化所有 Agent 读写同一份结构化的状态对象谁都不许“私藏上下文”。流程图化Agent 之间的跳转关系用有向图定义分支、并行、重试都是图上的边而不是藏在代码里的 if-else。恢复自动化系统在关键节点落检查点任务中断后可以从最近的检查点续跑而不是从零开始。在这套思路下Agent 不再是一次性的短命进程而是有记忆、可追踪、能中断恢复的长生命周期实体。这才是“持久化协作系统”的含义——它持久化的是 Agent 协作过程中产生的整个数字足迹。2. 整体架构怎么设计先想清楚“谁来管谁”2.1 三层分工调度层、执行层、持久化层多智能体编排系统听起来很玄乎但拆到最底层就是三层调度、执行、持久化。我先把这三层各自的职责说清楚。调度层Orchestrator是整个系统的“大脑”。它不干活但它决定下一步让谁干活。调度层维护着任务图谱知道当前执行到哪个节点、下一个节点是什么、要不要走分支、要不要做重试。在 OpenRig 里调度核心是一个 while 循环反复执行“查状态 → 找下一步 → 派给执行层 → 等结果 → 更新状态”这个循环直到整个图跑完。执行层Executor是真正干活的部分。每个 Agent 在被调度器点名之后才带着当前状态开始执行。执行层不关心“全局接下来怎么办”它只负责把当前这一步做对做完之后把结果写回共享状态。这个设计非常关键Agent 之间不直接通信只通过状态间接通信。避免了一堆 Agent 互相发消息、聊着聊着就跑偏的混乱场面。持久化层Persistence是容易被人忽视但地位极高的一层。它把共享状态、检查点、任务进度全部落到外部存储里。进程崩溃、机器重启、Agent 超时都靠着持久化层兜底。OpenRig 实践里我用了 Redis 作为主存储——不是因为它能存数据而是因为它同时能把“状态存储、缓存、锁、队列”这些能力全包了省掉很多基础设施上的拼装成本。2.2 为什么用有向图而不是线性链很多人第一次做编排时直觉就是“把 Agent 串成一个链”A 做完给 BB 做完给 C。这个方案在演示场景够用但在生产环境会撞上几个硬问题没有分支真实任务几乎都带条件判断。比如“如果审核 Agent 发现代码有 bug就回到开发 Agent 修复否则进入发布 Agent”。线性链没法表达这种回边。没有并行三个独立的分析任务可以同时跑线性链只能串行时间成本直接 X3。没有恢复锚点线性链崩溃后你不知道它走到哪了只能从头开始。所以 OpenRig 采用有向图DAG 受控环作为流程模型。每个节点是一个 Agent或者一个工具操作边是流转条件。图可以表达顺序、并行、条件跳转、循环重试表达能力比链式模型强一个量级。这个选择背后的逻辑其实是把“业务流程”和“Agent能力”解耦。业务流程是相对稳定的——它就是你这个产品的固定逻辑Agent 能力是可替换的。图把流程固定下来之后你换更聪明的模型、换不同的工具只需要替换节点内部的实现编排关系完全不用动。2.3 状态对象设计让每个 Agent“失忆”后能恢复记忆既然 Agent 之间通过状态间接通信那么状态对象怎么设计直接决定系统能不能真正持久化协作。OpenRig 里状态对象是一个结构化字典大概长这样{ task_id: task_20260101_abc, goal: 分析三个竞品的定价策略并输出对比报告, progress: { current_node: writer, completed_nodes: [researcher, analyst], attempts: {researcher: 1, analyst: 2, writer: 0} }, artifacts: { research_results: file://artifacts/research.json, analysis_report: file://artifacts/analysis.md }, context: { researcher_last_summary: 竞品A主打低单价策略竞品B走增值服务路线..., analyst_conclusion: ... }, metadata: { created_at: ..., updated_at: ..., version: 12 } }设计上有几个要点你得注意Agent 的中间产物不要直接塞进状态对象。比如数据分析 Agent 产出了一个 500MB 的 CSV你把它塞进 Redis内存直接扛不住。正确做法是把大文件落到对象存储或本地磁盘状态里只存文件路径和摘要。这就像工地上的白板只写“钢筋已经运到 3 号堆场”而不是把钢筋搬到白板旁边。每个 Agent 只该读它需要的那部分不该读全量状态。否则 Agent 的输入上下文会越滚越大模型很快被无关信息淹没。状态版本号是并发安全的基础。后面讲并发的时候你还会看到它。这个状态对象就是整个持久化协作系统的“共享白板”。Agent 失忆不可怕只要白板上的记录还在随时都能恢复现场。3. 持久化怎么做才不拖垮性能3.1 持久化不是“存数据”而是“存档游戏”很多做 Agent 的朋友一开始没把持久化当回事状态放内存里跑完就完事了。但只要你的 Agent 任务时长超过几分钟进程崩溃、服务重启、网络抖动几乎是必然发生的事。没有持久化一个已经跑了 80% 的任务直接报废这个成本在生产环境根本接受不了。我更愿意把持久化理解成“游戏存档”。你打一个长线游戏不可能一口气通关中途掉了线好的游戏设计一定是让你从最近的存档点继续而不是重新开始。多智能体编排的持久化就是在给任务的执行过程不断“存档”。存档本身不难难在设计“什么时候存、存多少、怎么恢复”。存太频繁开销大写放大严重存太少崩溃后丢的进度太多。OpenRig 的答案是检查点Checkpoint机制——在任务图的“关键节点”落盘。3.2 Redis 在持久化中的角色RDB 与 AOF 的取舍状态存哪我选了 Redis。原因很实在Redis 同时能干状态存储、消息队列、分布式锁的活基础设施成本极低。但在持久化这个场景下Redis 本身也有讲究——你至少得搞清楚 RDB 和 AOF 两种机制的区别不然数据丢了你都不知道怎么丢的。维度RDB 快照AOF 追加日志保存形式周期性生成全量二进制快照追加记录每次写操作数据安全性可能丢失最后一次快照后的所有写入取决于配置可以做到每秒级丢失窗口恢复速度快直接加载快照慢需要重放日志文件体积相对小可能很大需要定期重写适用场景允许少量数据丢失、追求恢复速度对数据完整性要求高的场景Agent 状态这种东西性质是“怕丢、但也要控成本”。所以 OpenRig 用的是组合策略AOF 设置为 everysec每秒刷盘保证基本不丢同时定期执行 BGSAVE 生成 RDB 快照——因为 RDB 恢复快崩了之后优先加载 RDB再用 AOF 补齐最后几秒的写入这是比较稳妥的组合。不过你也要注意Redis 的持久化是针对 Redis 进程自己的它解决的是“Redis 进程挂了不要丢数据”的问题。如果你的状态只存在 Redis 一个副本Redis 所在的机器如果整个掉电数据依然是危险的。所以 OpenRig 在生产环境还会对关键任务的最终产物做一份落盘备份Redis 是“协作工作台”不是“档案库”。3.3 检查点设计实操存什么、什么时候存、怎么恢复光有存储机制还不够你得设计好“存什么”和“什么时候存”。OpenRig 里我按下面的规则来存什么任务级状态目标、当前节点、已完成节点、各节点重试次数。Agent 运行上下文摘要不是把模型完整对话记录全存进去而是让每个 Agent 在结束时输出一段结构化摘要把关键结论、产出物路径写进状态。恢复时靠摘要重建上下文比重放全量对话可靠得多。中间产物的引用文件路径、版本号、批次号。什么时候存每个 Agent 节点执行完成后落一次检查点。工具调用成功返回后落一次检查点。分支切换、任务下发新子任务时落一次检查点。循环体内限制最大尝试次数每次循环开始前也落一次轻量检查点。怎么恢复恢复的主干逻辑其实很短。核心思路是读检查点 → 重建状态对象 → 定位到中断节点 → 从那里继续执行。def resume_task(task_id): checkpoint load_checkpoint(redis_client, task_id) if not checkpoint: return start_new_task(task_id) state rebuild_state(checkpoint) current_node state[progress][current_node] return orchestrator.run(task_id, start_nodecurrent_node, statestate)这套设计跑下来最直观的收益是任务中断恢复时间从“重跑全量任务”的 20 分钟压缩到“从断点续跑”的 2 分钟。这对生产环境的体验是质的改善。4. 并发一来Agent 编排怎么扛4.1 先想清楚Agent 是 IO 密集不是计算密集聊到并发很多人的第一反应是“加 CPU、加线程池、上异步框架”。但你先得想清楚一件事Agent 到底在等什么一个典型 Agent 的执行时间分布是——大部分时间花在等待 LLM 接口返回网络 IO和等待工具/数据库响应上真正用 CPU 做计算的占比极小。所以Agent 编排系统的瓶颈几乎是 IO 和状态一致性不是算力。你不需要拼命加核你需要的是把并发模型做对。OpenRig 的实现里调度器本身用的是事件循环 异步任务分发每个 Agent 节点作为异步任务提交到执行池。因为编排的每一步都依赖上一步的状态实际上真正并行发生在“分支节点”也就是一张图里有多个互不依赖的节点同时跑。比如“同时让三个研究员分别研究三个竞品”这三个 Agent 就可以并行执行。4.2 并发调度的三个基本动作排队、幂等、聚合并发一来挑战不是“跑得快”而是“不跑乱”。OpenRig 里我拆成三个基本动作第一个动作排队。所有新任务先进入任务队列不直接打到执行层。队列的好处是给系统一个“蓄水池”流量再大也不会直接把后端的模型 API 打爆。我在 Redis 里用一个 Stream 或 List 做任务队列调度器从队列里拉任务分发到对应节点。排队也方便做优先级——高优任务插队、低优任务排队在任务系统中属于标配能力。第二个动作幂等。并发场景下最怕的是一件事被干了两遍。比如任务超时了调度器重试结果原来的 Agent 其实已经完成了于是产出重复、甚至状态冲突。解法是task_id 节点名作为幂等键。每次执行节点前检查这个键是否已经在“已完成”集合里如果是直接跳过不重复执行。def run_once(task_id, node_name, fn): done_key fnode_done:{task_id}:{node_name} if redis_client.exists(done_key): return load_node_result(task_id, node_name) result await fn() redis_client.setex(done_key, TTL, 1) save_node_result(task_id, node_name, result) return result第三个动作聚合。多个并行 Agent 各自跑完后结果怎么汇合OpenRig 的做法是定义一个汇合节点它等待所有上游分支都标记为已完成然后从状态里把每个分支的产物取出来做统一整理再往下传。这对应到图上其实就是一个“扇入”结构天然适合下游接汇总 Agent 或者校验 Agent。4.3 多 Agent 抢同一个目标共享资源的竞争问题编排里的并发不只是“多个任务同时跑”还包括“多个 Agent 想同时改一份共享数据”。最典型的就是两个 Agent 同时分析完都想往同一份报告里追加自己的章节结果互相覆盖。这个问题的本质是共享状态需要并发控制。OpenRig 偏好用乐观锁而不是悲观锁。具体说状态对象里带一个 version 字段Agent 写回结果时带上自己读取时的版本号。写入时检查版本号是否变化如果变了说明别人改过了当前写入要重新读取、合并、再写。def update_state(task_id, expected_version, new_state): # 用 Lua 脚本保证原子性版本号匹配才更新 lua local cur redis.call(GET, KEYS[1]) if cur ARGV[1] then redis.call(SET, KEYS[1], ARGV[2]) return 1 end return 0 ok redis_client.eval(lua, 1, fstate:{task_id}, str(expected_version), json.dumps(new_state)) if ok 0: raise StateConflictError(状态已被其他Agent更新需要重新合并)这个方案在大多数场景下够用而且要的额外开销很小。如果你遇到极端高冲突的场景多个 Agent 频繁写同一份文档再考虑把写入拆细、用分布式锁保护关键区段——但通常情况下乐观锁已经能解决 90% 的“抢白板”问题。5. 手把手搭一个最小可用的 OpenRig5.1 技术选型LangGraph FastAPI Redis 够用了下面我从零搭一个最小可用系统。技术栈就三样LangGraph编排、FastAPI对外接口、Redis持久化和队列。为什么选这三样LangGraph原生支持有状态图编排节点、边、条件边、检查点这些概念都是内置的是我用下来和 OpenRig 思路最契合的工具。比起自己手写一个图引擎直接站在它的肩膀上省很多事。FastAPI异步友好和编排器的异步执行模型天然搭配。而且自带 OpenAPI 文档后期接前端、接管理后台都方便。Redis前面说过了一个 Redis 把状态存储、队列、锁全包了。这三样组装起来代码量不大却能跑通“创建任务 → 编排执行 → 持久化 → 查询状态”一整套核心链路。5.2 定义 Agent 和任务图谱我拿一个真实感比较强的场景来演示做一个简单的“竞品分析报告生成器”包含两个 Agent——研究员 Agent负责查竞品资料、产出结构化笔记和写作 Agent负责把笔记整理成完整报告。流程图长这样start → researcher → writer → 检查质量 → 合格则 end不合格则回 researcher 补查。代码里用 LangGraph 定义出来是这样from typing import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): task_id: str goal: str research_notes: str report: str quality_pass: bool attempts: int async def researcher(state: AgentState) - dict: notes await run_agent( researcher, state[goal], max_attempts3 ) return {research_notes: notes} async def writer(state: AgentState) - dict: assert state[research_notes], 研究员还没产出不能进入写作 report await run_agent( writer, f基于以下研究笔记生成报告\n{state[research_notes]} ) return {report: report} def quality_check(state: AgentState) - str: if state[attempts] 2: # 防止死循环 return done if state[report] and len(state[report]) 200: return done return retry g StateGraph(AgentState) g.add_node(researcher, researcher) g.add_node(writer, writer) g.add_edge(start, researcher) g.add_edge(researcher, writer) g.add_conditional_edge( writer, quality_check, {retry: researcher, done: END} ) workflow g.compile()这段代码看着不多但它已经包含了一条完整的分支 循环 限流逻辑质量不合格就打回研究员重查但最多重试 2 次防止挂死。5.3 把检查点和恢复逻辑写进编排循环上面 LangGraph 定义的是“单次执行”的逻辑可 OpenRig 要的是“中断后能恢复”。这里需要把 LangGraph 的执行接到我们自己的检查点读写逻辑里。比较省事的做法是用 LangGraph 自带的持久化接口配合 Redis 存储检查点。大致思路是每次节点执行前后把状态对象保存到 Redis并记录当前节点位置。下面是简化的插入点class RedisCheckpointSaver: def save(self, task_id, state): redis_client.set( fcheckpoint:{task_id}, json.dumps({**state, saved_at: time.time()}) ) def load(self, task_id): raw redis_client.get(fcheckpoint:{task_id}) if not raw: return None return json.loads(raw)在每次执行一个 Agent 节点之前从检查点恢复状态执行完之后立刻保存新状态。这样编排器就获得了“掉电续跑”的能力async def run_task(task_id, goal): cp checkpoint_saver.load(task_id) if cp: state cp # 跳过已完成的节点从断点继续 else: state {task_id: task_id, goal: goal, attempts: 0, research_notes: , report: } async for event in workflow.astream(state): checkpoint_saver.save(task_id, event) return state这代码的价值在于——一旦 Researcher 跑了几分钟才出结果结果 Writing Agent 突然崩了整个任务不需要重跑。系统读取检查点定位到 writer 节点只重做这一个节点就行。5.4 暴露成 APIFastAPI 接入顺带把任务查询做了编排器搭好之后对外必须有一个干净的业务接口。OpenRig 用 FastAPI 暴露三个核心端点创建任务、查询任务状态、手动恢复任务。创建任务的逻辑是“接单立刻入库再丢队列异步执行”查询则直接读 Redis 里的状态。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uuid, json app FastAPI() class TaskCreate(BaseModel): goal: str app.post(/tasks) async def create_task(req: TaskCreate): task_id uuid.uuid4().hex # 先把任务和初始状态落库再入队异步执行 initial_state {task_id: task_id, goal: req.goal, attempts: 0} redis_client.set(fcheckpoint:{task_id}, json.dumps(initial_state)) await redis_client.rpush(agent_queue, task_id) return {task_id: task_id, status: queued} app.get(/tasks/{task_id}) async def get_task(task_id: str): raw redis_client.get(fcheckpoint:{task_id}) if not raw: raise HTTPException(status_code404, detail任务不存在) state json.loads(raw) return {task_id: task_id, status: state.get(progress, running), state: state} app.post(/tasks/{task_id}/resume) async def resume_task(task_id: str): raw redis_client.get(fcheckpoint:{task_id}) if not raw: raise HTTPException(status_code404, detail任务不存在) # 重新入队编排器会从检查点继续执行 await redis_client.rpush(agent_recovery_queue, task_id) return {task_id: task_id, status: resumed}到这一步你已经有了一个 API 化的多智能体编排服务创建任务、异步执行、状态可查、崩溃可续。这就是一个 Agent 中台最核心的那几个接口。真实产品要加的无非是鉴权、审计、监控面板底层骨架就是这套东西。6. 实弹问题我做多智能体编排时踩过的 6 个坑6.1 问题速查表实践做得多了踩坑清单自然就长。我把最典型的 6 个问题整理成速查表方便你直接对号入座问题现象根因解决方案任务中断后 Agent 失忆崩溃重启后Agent 忘了之前所有分析从零开始状态只在内存里没有检查点关键节点落检查点恢复时读状态重建上下文Agent 输出格式飘忽下游解析 JSON 失败任务卡住LLM 偶尔输出多余字段或截断JSON mode schema 校验 解析失败自动重试Redis 连接耗尽并发一高,调度器整体卡住没用连接池连接反复建立/释放配置连接池加上超时回收策略多个 Agent 互相覆盖产出两个 Agent 同时写一份报告后面的覆盖前面的共享状态没有版本控制乐观锁版本号 Lua 原子更新AOF 文件膨胀,恢复很慢Redis 启动恢复要几分钟状态写入太频繁,日志无限增长定期 BGSAVE 生成 RDB AOF 重写 只落关键检查点图循环跑不完同一节点反复重试,任务永不结束条件边判断失真没有最大步数限制每个节点限制最大尝试次数,超限强制结束6.2 坑 1把“大对象”硬塞进状态对象我的第一个版本让数据分析 Agent 直接把它的中间结果 DataFrame 序列化以后塞进状态对象存 Redis。结果任务跑到第 3 个节点Redis 内存直接报警后续写入开始超时整条流水线全堵死了。后面学乖了状态对象里只存“摘要 引用”不存“全量数据”。DataFrame 落到磁盘或者对象存储状态里记文件路径、schema、行数、几个关键统计量。Agent 在后续节点真的需要原始数据时通过路径去加载而不是通过状态硬扛。这个改动让内存占用直接降了 90% 以上。教训就一句话白板上写的是“线索”不是“原材料”。6.3 坑 2对模型输出格式的过度信任多 Agent 系统里最脆弱的环节就是节点间的数据交接格式。一开始我觉得“模型输出 JSON 有什么难的”结果实践发现模型偶尔会在合法 JSON 后面多一段解释性文字偶尔会漏一个逗号甚至偶尔直接把 JSON 截断一半。下游解析一旦失败整个编排流就卡在某个节点而且报错信息晦涩难懂。现在的做法是每个 Agent 的返回都要过一层“格式校验 纠错重试”。解析失败不直接抛异常而是把错误信息反馈给 Agent让它重新修正输出最多回头重试 2 次。这招让整体成功率大幅提升。做编排时一定要默认“模型输出不可靠”在边界上做保护。6.4 坑 3检查点写得太密集Redis 扛不住最早我为了求稳每一步工具调用都写一次全量检查点。看起来安全实际把 Redis 写放大搞得非常严重——任务一多Redis 的 CPU 和带宽都成了瓶颈甚至反过来拖慢了任务执行本身。现在 OpenRig 的策略是分档关键节点写完整检查点节点完成、分支切换、循环开始普通节点只在内存里更新状态等走到关键节点再统一落盘。同时注意把检查点键名加上过期时间已经完成的任务不会被无限期占着存储。持久化不是越频繁越好而是要在“恢复粒度”和“写放大的成本”之间找平衡。最后的一点实际操作体会我在这个项目里最大的体会是做多智能体编排真正的难点从来不是模型能力不足而是工程化基建缺位。模型再聪明没有一个显性化的状态层没有一套可恢复的执行流程它在长任务场景下依然是个不可依赖的组件。OpenRig 给它加上“状态显性化、流程图化、恢复自动化”这三根骨架之后Agent 才真正从“能聊天的模型接口”变成了“能扛任务的系统组件”。还有一个小技巧想分享给正在动手的朋友一开始别追求大而全的编排平台先把一条最核心的业务流程做成“图 状态 检查点”的最小闭环跑通之后再逐步加分支、加并行、加并发控制。我自己就是因为一上来想得太复杂绕了不少弯路。多智能体编排这套东西越早动手踩坑越早能感受到它和单 Agent 在工程体验上的差距。