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 做完给 B，B 做完给 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 两种机制的区别，不然数据丢了你都不知道怎么丢的。

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_node=current_node, state=state)

这套设计跑下来，最直观的收益是：任务中断恢复时间从“重跑全量任务”的 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 = f"node_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, f"state:{task_id}", str(expected_version), json.dumps(new_state)) if ok == 0: raise StateConflictError("状态已被其他Agent更新，需要重新合并")

这个方案在大多数场景下够用，而且要的额外开销很小。如果你遇到极端高冲突的场景（多个 Agent 频繁写同一份文档），再考虑把写入拆细、用分布式锁保护关键区段——但通常情况下，乐观锁已经能解决 90% 的“抢白板”问题。

5. 手把手搭一个最小可用的 OpenRig

5.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_attempts=3 ) 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( f"checkpoint:{task_id}", json.dumps({**state, "saved_at": time.time()}) ) def load(self, task_id): raw = redis_client.get(f"checkpoint:{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 暴露成 API：FastAPI 接入顺带把任务查询做了

编排器搭好之后，对外必须有一个干净的业务接口。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(f"checkpoint:{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(f"checkpoint:{task_id}") if not raw: raise HTTPException(status_code=404, 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(f"checkpoint:{task_id}") if not raw: raise HTTPException(status_code=404, detail="任务不存在") # 重新入队，编排器会从检查点继续执行 await redis_client.rpush("agent_recovery_queue", task_id) return {"task_id": task_id, "status": "resumed"}

到这一步，你已经有了一个 API 化的多智能体编排服务：创建任务、异步执行、状态可查、崩溃可续。这就是一个 Agent 中台最核心的那几个接口。真实产品要加的无非是鉴权、审计、监控面板，底层骨架就是这套东西。

6. 实弹问题：我做多智能体编排时踩过的 6 个坑

6.1 问题速查表

实践做得多了，踩坑清单自然就长。我把最典型的 6 个问题整理成速查表，方便你直接对号入座：

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 在工程体验上的差距。