上周跟一个做运营系统的朋友聊天他那边接了个挺头疼的需求客服团队手里同时跑着几个AI助手一个负责查订单、一个负责算价格、还有一个负责审核退款。单个拎出来测效果都不错各说各话也挺流畅。但只要把三个Agent串到同一个业务流程里立刻乱套——订单查完不传给报价的审核的拿着旧数据进行判断流程跑到一半卡住日志里只有一个孤零零的错误谁也不知道刚才执行到哪一步。这个场景我想做多智能体编排的朋友都熟离散的AI Agent个体能力再强没有一套统一的编排和持久化机制兜底协作起来就是灾难。这篇文章就聊一聊我最近实践的一个项目OpenRig。简单说它是一套把“散装”AI Agent组织成“持久化协作系统”的编排方案核心解决三件事Agent之间怎么通信、工作流怎么编排、运行状态怎么持久化。文章会从设计思路、技术选型、核心代码到落地踩坑逐层拆开适合正在做多Agent应用的工程师参考也适合刚接触AI Agent编排、想建立一个全局认知的读者。1. 为什么要做多智能体编排离散Agent的痛点1.1 “能用”和“能用在一起”是两码事单独一个Agent的“能用”通常只依赖两样东西一个还不错的提示词外加几个工具函数。模型负责理解意图、拆解步骤工具负责执行具体动作中间用循环包一层看起来就能自动干活了。但多Agent协作完全是另一个命题。我见过太多团队把单Agent的经验硬套到多Agent上结果踩出一连串问题。最典型的三类上下文割裂A Agent产出的结构化结果B Agent拿不到只能重新解析自然语言状态失忆流程跑到一半某个Agent调用失败重试时已经找不到之前确定的中间结果整条链路从头再来冲突无人裁决多个Agent对同一个数据的判断不一致没有机制拍板输出就随机。这些问题的共性在于单个Agent是“函数式”的调用完就结束没有记忆也不关心上下游而多Agent协作系统是“有状态”的它像一条生产线每个Agent是工位工件数据必须被妥善地在工位间流转任何一步出问题都能定位、能回退、能续跑。1.2 编排的本质从“各自干活”到“共享记忆的流程”所谓编排我理解下来就是两件事流程组织和状态管理。流程组织解决“下一步该让谁干活”状态管理解决“到目前为止干到了哪一步、结果是什么”。对应到工程上就是设计一个图状或者链状的执行结构同时让每个节点都能读写一份共享状态。这里必须单独说“持久化”。在很多多Agent项目里持久化被理解成“把数据存下来”其实这个理解窄了。在多Agent编排语境下持久化至少三层含义状态持久化即每个节点执行完把上下文、中间结果、Token消耗写入外部存储实现任何时刻的可恢复生命周期持久化Agent不是一个一次性调用的临时进程而是长期存在的协作实体可以在多轮会话里持续接收消息、累积记忆审计持久化每次运行从输入到输出的完整路径都能回溯出了问题能查到是哪个Agent哪一步干的。这三点合起来就是我标题里说的“持久化协作系统”。一个不能垮掉重启继续跑的Agent系统和一个断电就丢数据的数据库一样是不能上生产的。2. 系统总体设计与技术选型2.1 OpenRig的整体架构与命名OpenRig这个名字字面意思是“一个开放的、把Agent架起来稳定运转的支架”。Rig在英文里有钻井架的意思钻井架的特点就是结构稳定、层层咬合能在恶劣环境下持续作业我希望这套编排底座也能给Agent提供同样的支撑力。整个系统分四层接入层对外提供API入口接收业务请求返回任务ID。这一层用FastAPI实现异步非阻塞能扛住大量并发请求。编排层核心业务逻辑负责构建Agent执行图、控制节点流转、处理分支并行与条件路由。这一层采用LangGraph因为它天然支持图状态管理和检查点机制正合适。执行层一个个离散的AI Agent实体每个Agent是纯函数式的“输入→处理→输出”不感知全局流程只通过统一协议读写上下文。存储层Redis负责运行时状态、检查点快照和消息队列对象存储或关系库负责长期审计日志这部分按需接入即可。这种分层最大的好处是解耦。编排层不知道Agent内部用了什么模型、什么提示词只认统一协议的输入输出Agent不关心上下游是谁只对上下文对象做读写。两边都能独立演进。2.2 技术栈选型LangGraph、Redis、FastAPI的理由最初设计OpenRig的时候技术选型纠结过好几轮。摆在面前的候选有自己实现一个有限状态机、用LangChain的AgentExecutor、用LangGraph、甚至直接用消息队列把Agent串起来。最终敲定的组合是LangGraph Redis FastAPI理由分别说一下。LangGraph赢在“图编排”这个定位。多Agent流程天然是图不是线性链意图识别可能走A分支也可能走B分支两个Agent可以并行执行最后汇聚到同一个节点。自己写状态机每加一个分支就要改一遍流转逻辑维护成本很高。LangGraph的StateGraph恰好抽象出了这套东西你只需要定义状态结构和节点边和条件路由都作为配置项声明代码结构跟业务流程一一对应。Redis在这个系统里的位置也很特殊。它同时承担三份工作状态检查点存储LangGraph的checkpointer可以直接落到Redis轻量级消息队列用Redis Stream保存待执行任务天然支持消费者组可以做并发消费缓存层存Agent的常用上下文和工具返回结果。一件基础设施解决三个问题运维成本省了很多。FastAPI的价值偏工程侧。它原生支持async配合Redis-py的异步客户端可以支撑大量Agent任务的并发提交。加上Pydantic自动做请求校验接口定义的严谨性显著提升这在给Agent传数据时特别重要——脏数据进到工作流里后面再排查就费劲了。2.3 主流多Agent架构路线对比在动手之前有必要把业界主流的多Agent编排路线放在同一张表里对比一下搞明白每条路线的适用边界选型才不会跑偏。架构路线核心特征适用场景典型问题单Agent工具调用一个Agent反复调多个工具任务边界清晰、步骤固定上下文易膨胀难以分工链式PipelineAgent串行执行前一个Agent输出作为下一个输入稳定的线性流程加分支困难单点故障拖垮全链图状编排Graph显式定义节点、分支、并行与汇聚流程复杂、分支多的业务需要学习框架状态设计有门槛广播/黑board模式Agent共享一块黑板互相读取更新开放式协作、信息共享容易产生脏读无流程约束Human-in-the-loop关键节点等待人工审批高风险决策、合规场景吞吐量受限依赖人工时效OpenRig最终选择了图状编排为主体同时预留Human-in-the-loop节点。核心判断是我们的业务场景分支多、有并行需求但又需要流程可控、每个环节可追溯。图编排的约束力正好补上纯Agent那套“自由发挥”的短板——约束不是坏事对生产系统来说可控比灵活重要得多。3. 核心模块实现AI Agent的单元定义与注册3.1 统一通信协议Agent之间怎么说话多Agent系统最怕什么怕A说话B听不懂。如果每个Agent自定义一套入参出参靠解析自然语言字段互相传递数据那整个系统就是个大型“猜谜现场”。OpenRig的第一步就是定协议所有Agent的输入输出必须走同一个消息结构。我用Pydantic定义了一个AgentMessage作为Agent间通信的标准信封。from pydantic import BaseModel, Field from typing import Any, Optional from datetime import datetime import uuid class AgentMessage(BaseModel): message_id: str Field(default_factorylambda: uuid.uuid4().hex) session_id: str sender: str # 上游 Agent 名称 receiver: str # 下游 Agent 名称* 表示广播 payload_type: str # 数据类型如 order_info / risk_result payload: Any # 实际数据体 created_at: datetime Field(default_factorydatetime.now)这个结构很简单但约束力很强。它强制每个Agent明确声明自己的身份、发给谁、发什么类型的数据。payload_type解决了语义对齐的问题下游Agent不需要“理解”上游的自然语言只需要根据payload_type做结构化消费。这就好比团队协作里大家不靠猜而是靠统一的工单格式传递信息效率完全不同。会话标识session_id是贯穿全流程的线索。一次用户请求对应一个session_id所有Agent消息都挂在这个ID下面既方便追踪链路也方便做持久化和审计——后面查问题、回放、续跑全靠它。3.2 工具与依赖注入让Agent真正能干活Agent光会说话不行还得能干活。干活的手段是工具调用。OpenRig采用一个非常传统的注册表模式把工具注册和Agent逻辑解耦。from typing import Callable, Awaitable from functools import wraps TOOL_REGISTRY: dict[str, Callable] {} def register_tool(name: str): def decorator(func: Callable) - Callable: TOOL_REGISTRY[name] func return func return decorator register_tool(query_inventory) async def query_inventory(sku: str) - dict: # 实际业务中会从这里查真实库存系统 return {sku: sku, available: 42}这里有一个容易被忽略的设计工具注册和Agent注册分开。因为一个工具可能被多个Agent复用比如“查库存”既可以被售前Agent用也可以被审核Agent用。如果工具跟Agent绑死复用就麻烦了。注册表模式的好处是Agent运行时按需从TOOL_REGISTRY里取工具即可新增一个工具不需要改动任何Agent代码。工程上工具函数应当是无状态的只根据传入参数返回结果不保存内部记忆。它的记忆由上层上下文对象统一管理。这一点和函数式编程的理念一致工具的输入输出是确定的出了问题才好复现。3.3 上下文对象与Token计量Agent协作不只是消息传递还需要一份“共享工作区”。OpenRig里我称它为AgentContext本质上是一个只读快照加可写追加的状态。class AgentContext(BaseModel): session_id: str inputs: dict {} # 本次工作的初始输入 messages: list[AgentMessage] [] # 全链路消息记录 shared_memory: dict {} # Agent 间共享的只读结果 step_index: int 0 # 当前执行到第几步 total_tokens: int 0 # 累计 Token 消耗shared_memory是核心设计。它只允许本轮“主写者”写入其他Agent只读从根上避开数据竞争。比如意图识别Agent写出intent字段后续所有Agent只能看这个字段不能改。顺便回应一个高频问题“ai agent token是什么意思”。Token是大模型处理文本的最小计量单位可以粗略理解为一个单词或半个汉字。在多Agent系统里每个Agent调用大模型都会消耗Token而Token消耗直接等于钱。所以context里专门放total_tokens每次Agent返回时由编排层统一累加。这一步在架构上非常重要你不统计Token就没有成本概念等系统上线跑起来账单会比预期高出一大截。4. 多Agent编排把离散Agent编织成工作流4.1 编排图设计从哪里来到哪里去有了Agent单元和通信协议下一步就是把它们“编织”成一张执行图。图编排的核心元素有三个节点Node、边Edge、状态State。节点承载Agent逻辑边定义流转路径状态是节点之间传递数据的载体。在设计编排图之前我强烈建议先画一张业务流转图别急着写代码。把流程里的每个决策点圈出来什么条件下走哪个分支、哪些环节可以并行、哪些环节必须汇聚等待。这一步想不清楚写到代码里就会反复返工。OpenRig使用LangGraph组织编排层。LangGraph不是要替代Agent而是给Agent当“导演”。Agent还是那个Agent但下一步该谁上场、演完要给谁递话全部由StateGraph控制。这样设计的好处是业务调整流程时只需要改图定义不需要改Agent实现。4.2 实战示例客服工单自动处理工作流拿一个具体的例子把整个编排串起来客服工单自动处理。业务流程是这样的用户提交工单系统先做意图识别识别出是“查库存”还是“申请售后”如果是查库存路由到库存查询Agent如果是售后则走风险审核Agent通过后进入退款执行Agent。定义状态结构from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class TicketState(TypedDict): session_id: str user_input: str intent: str sku: str | None inventory_result: dict | None risk_level: str | None approved: bool | None messages: Annotated[list, operator.add] # 消息用 add 归约器收集这里有个关键点messages字段用了Annotated operator.add。LangGraph的StateGraph在节点执行完会更新状态如果多个节点同时往同一个字段写普通赋值会互相覆盖。用operator.add作为归约器就变成追加而不是覆盖天然解决了并行Agent写同一字段的数据竞争问题。接下来定义节点函数每个节点就是调用对应的Agentasync def intent_node(state: TicketState) - dict: intent await intent_agent(state[user_input]) return {intent: intent, messages: [fintent_node - {intent}]} async def inventory_node(state: TicketState) - dict: result await inventory_agent(state[sku]) return {inventory_result: result} async def review_node(state: TicketState) - dict: review await risk_agent(state[sku], state[user_input]) return {risk_level: review[level], approved: review[approved]} async def refund_node(state: TicketState) - dict: result await refund_agent(state[sku]) return {messages: [frefund done: {result}]}最后组装成图from langgraph.graph import StateGraph, END def build_workflow(): g StateGraph(TicketState) g.add_node(intent, intent_node) g.add_node(inventory, inventory_node) g.add_node(review, review_node) g.add_node(refund, refund_node) g.set_entry_point(intent) g.add_conditional_edges(intent, route_by_intent, { inventory: inventory, review: review, }) g.add_edge(inventory, refund) g.add_edge(review, refund) g.add_edge(refund, END) return g.compile()conditional_edges是LangGraph里表达“路由”的入口。route_by_intent是一个纯函数根据state[intent]返回字符串映射到下一个节点。这套写法把路由策略做成显式配置后续加了新意图只需要加一个映射项不需要改动图框架。4.3 并发执行与共享状态的一致性多Agent编排绕不开并发问题。典型场景用户提交工单后需要同时检查黑名单、查库存、查历史订单三个操作互相独立串行执行会很慢。LangGraph支持多条边从同一个节点出发但要注意并发节点的输出归约器必须是“追加型”的。我用一个示例来说明g.add_node(precheck, precheck_node) g.add_node(check_blacklist, check_blacklist_node) g.add_node(check_stock, check_stock_node) g.add_node(check_history, check_history_node) g.add_node(aggregate, aggregate_node) g.set_entry_point(precheck) g.add_edge(precheck, check_blacklist) g.add_edge(precheck, check_stock) g.add_edge(precheck, check_history) g.add_edge(check_blacklist, aggregate) g.add_edge(check_stock, aggregate) g.add_edge(check_history, aggregate)这个拓扑里precheck执行完会同时触发三个检查节点然后汇聚到aggregate节点。这里最关键的是aggregate节点怎么写——它不能等待某个固定字段而要读取全量并发结果并做汇总。我在实际编码时让每个并发节点都向messages字段追加消息aggregate从messages里按前缀区分来源一份状态设计成“共享区域可读不可写”之后并发写冲突就基本绝迹了。还有一层并发在系统级多个用户同时提交工单工作流实例之间不能互相干扰。这就是session_id存在的另一个价值——每个session_id对应一个独立的状态实例Redis里按session_id做键隔离天然满足多租户隔离。5. 持久化机制让Agent拥有记忆和恢复力5.1 Redis持久化机制详解RDB、AOF与混合模式在多Agent系统里选Redis做持久化层就必须把Redis自己的持久化机制说清楚否则重启丢数据前面所有设计都白搭。这里结合几个热词里老被问到的点把“Redis持久化机制”完整拆一遍。RDB模式是定期把内存里的全量数据拍一张快照存到磁盘触发条件由save配置决定。我常用的配置是save 900 1 save 300 10 save 60 10000 appendonly yes appendfsync everysec三行save规则分别表示900秒内有1次写入就拍快照300秒内有10次写入就拍快照60秒内有10000次写入就拍快照。可以这样理解写入频率越高保存数据的间隔就要越短用来平衡宕机丢失量和恢复效率。默认规则是900秒1次、300秒10次、60秒10000次这个量级在Agent场景基本够用。AOF模式是把每次写命令追加到日志文件重启时回放日志恢复到最新状态。appendfsync everysec表示每秒刷盘最多丢1秒的数据。AOF的数据安全级别比RDB高但日志文件会持续膨胀恢复速度也更慢。混合模式是Redis 4.0之后的折中方案AOF重写时先生成一份当前数据的RDB快照再追加重写期间的增量日志。这样既能快速加载快照又不会丢失增量数据。在OpenRig的部署里我直接启用混合模式快照用于崩溃后的快速恢复AOF增量日志用于保证数据少丢。5.2 检查点与断点续跑崩溃了也能接着干对多Agent编排来说系统崩溃不可怕可怕的是崩溃之后整个流程要重新跑一遍。OpenRig用LangGraph的Checkpointer机制解决这个问题在每个节点执行完成后把当前状态打成一个检查点写入Redis。这样工作流宕机恢复后可以从最后一个检查点继续执行而不是从头开始。from langgraph.checkpoint.redis import RedisSaver checkpointer RedisSaver(redis_urlredis://localhost:6379) app build_workflow().compile(checkpointercheckpointer) config {configurable: {thread_id: session_id}} result await app.ainvoke(initial_input, configconfig)这一段里thread_id就是我们的session_id。LangGraph会根据这个ID自动定位到已存在的状态快照继续往下执行。当时我测试“断点续跑”功能时直接把Redis进程杀掉再拉起来重新提交同一个session_id的任务工作流从崩溃点继续跑完了剩余节点而不是回到起点。这个能力在超长任务上尤其值钱——每个节点都要调用大模型重跑一遍的成本可不只是时间还有白花花的Token。5.3 会话存储、审计与回放持久化还有一个容易忽略的价值审计与回放。我在OpenRig里专门加了一个WorkflowRun记录表每个session_id对应一条主记录持有流程名称、创建时间、结束时间、最终状态。class WorkflowRun(models.Model): session_id models.CharField(max_length64, uniqueTrue) flow_name models.CharField(max_length64) status models.CharField(max_length16, defaultrunning) total_tokens models.IntegerField(default0) created_at models.DateTimeField(auto_now_addTrue)每条AgentMessage单独存一份挂到session_id下面。这样出了问题我可以完整回放某个工单为什么被退款看消息流就能还原当时的决策链条。这个能力在生产环境的价值怎么强调都不过分。很多多Agent项目上线后出现问题最尴尬的就是拿不出证据复盘而持久化的消息流天然就是证据链。6. API接入与任务编排FastAPI和Django各司其职6.1 FastAPI异步接入层OpenRig的对外入口用FastAPI实现。接口不做任何业务逻辑只负责接收请求、组装初始状态、丢进编排层、立即返回任务ID。from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app FastAPI(titleOpenRig) class RunRequest(BaseModel): session_id: str flow_name: str payload: dict app.post(/v1/workflow/run) async def run_workflow(req: RunRequest): task_id await workflow_manager.submit( session_idreq.session_id, flow_namereq.flow_name, payloadreq.payload, ) return {task_id: task_id, status: accepted}这里需要留意一个设计原则不要把大模型调用放在HTTP请求里同步执行。Agent调用大模型的耗时动辄几秒甚至几十秒如果同步做并发一上来HTTP连接池就被占满了接口被拖死。异步提交、任务ID轮询是更合理的模式。我还给每个请求加了超时和重试参数。模型调用偶尔会出现长等待或者超时重试时要特别注意只有幂等操作可以安全重试。比如“查库存”是幂等的重复查没问题“发起退款”就必须加幂等键防止同一个退款请求被重试执行两次。6.2 扛并发任务队列、限流与重试“ai agent怎么扛并发”是我见得太多的搜索词这里专门展开。多Agent系统的瓶颈几乎都在大模型API调用上所以扛并发的核心不是让Agent跑得更多而是控制同时打到模型API上的请求数。OpenRig的实践是三步第一步把任务做成消息。用Redis Stream作为任务队列提交任务时推一条队列消息后台Worker消费。XADD openrig:queue * flow_name ticket_workflow session_id abc123 payload {sku: A1001}第二步用消费者组控制消费速度。Redis Stream消费者组可以设定多个消费者争抢消息但我们可以通过限制Worker数量把并发的模型调用控制在合理范围。实测下来单个Worker进程开8个协程并发模型接口依然稳定延迟没有明显劣化。第三步加应用层限流。用Redis的INCREXPIRE做一个简单的令牌桶控制每分钟最多调用多少次大模型防止突发流量打爆模型API配额。一套组合拳下来OpenRig在本地环境跑了压测120个并发工单请求进入系统队列堆积在200条以内任务平均完成时间比串行执行快了近4倍。这个结果说明了编排层和队列层的价值——Agent本身没变变的是它们被组织的方式。6.3 Django管理后台人工审核与兜底FastAPI负责性能敏感的高频接口管理后台我用Django。注意这里不是用Django去跑Agent任务两个框架的位置完全不同FastAPI是“执行通道”Django是“管理窗口”各司其职。在OpenRig里Django主要承担三类工作流功能第一展示WorkflowRun列表运营人员可以按状态筛任务、点进去看消息流历史第二标记风险工单需要人工介入的节点审核人在Django后台直接改状态、写明结论编排层会轮询到状态变更后继续执行第三重跑失败任务找到出错的任务点击重试系统会基于持久化状态从断点续跑。from django.contrib import admin admin.register(WorkflowRun) class WorkflowRunAdmin(admin.ModelAdmin): list_display (session_id, flow_name, status, total_tokens, created_at) list_filter (flow_name, status)这套组合在实际使用中非常顺手线上问题用FastAPI跑人工决策用Django审。热词里“用ai agent开发django”大概率是想把Agent揉进Django业务里我的建议恰恰相反——不要在Django里跑Agent主流程把它定位成管理与审计后台性能耐力和职责边界都更清晰。7. 实战踩坑与排查实录7.1 并发写状态导致数据覆盖现象工单流程里风险审核和库存查询两个节点并行执行审核节点先完成往后走了结果库存节点晚一步返回把整条工作流的库存字段覆盖成空值后续退款节点看到库存为空直接报错。排查过程先从WorkflowRun的消息流回溯发现两个节点确实都往同一个顶层字段写了值。LangGraph状态更新时后执行完的节点覆盖了先完成的节点不是谁逻辑错了是状态归约器没设计对。解决办法把这种并发节点产生的字段全部改成追加型归约器或者让它们写进不同的独立字段最后汇聚节点再做合并。记住一条原则并行节点不要写同一个顶层字段除非那个字段是专门设计成可追加的。7.2 上下文越塞越多Token爆炸现象某个Agent任务跑了30多步到后面模型响应明显变慢而且开始“答非所问”查看total_tokens发现已经爆到很高。排查过程把messages列表打出来发现每个Agent都把上游消息完整带上了链路越长上下文越臃肿。模型能处理的Token量有限上下文一长注意力就被稀释回答质量自然下降。解决办法彻底抛弃“全量历史”的做法改为三层策略。底层短期记忆保留最近N轮消息上层摘要记忆由专门的摘要Agent定期压缩历史需求字段直接写进共享状态不塞自然语言。这套方案上线后单任务Token消耗降低了约35%回答稳定性还提升了。7.3 Agent陷入死循环任务永远跑不完现象某个意图识别Agent在特定输入下不断触发工具调用来回横跳任务队列里同一session_id堆积了大量消息。排查过程定位到工作流图里有一条条件路由配置错误意图识别结果没有匹配到任何分支按照LangGraph默认行为节点原地重试形成死循环。解决办法给所有工作流加两道保险。第一道图编译时设置最大递归深度超过就强制判失败第二道编排层给每个session_id设置整体超时时间超时直接终止并生成告警。这两道保险成本极低但能在很多看不见的边界场景里救命。7.4 问题速查表问题现象可能原因定位手段推荐解法下游Agent拿到空数据并发节点覆盖同名字段查看消息流时间线并行节点写独立字段任务越来越慢、结果变差上下文携带了过多历史查total_tokens趋势压缩历史、提炼摘要流程卡住不结束条件路由没匹配上查看图执行日志加最大步数、整体超时重复退款/重复发消息重试没有幂等保护查同session_id多条成功记录加幂等标识字段重启后状态丢失Redis持久化未开启检查redis.conf开启AOF与混合持久化接口大量超时模型调用占满HTTP连接池查看API吞吐和队列堆积异步任务队列限流写到这里我特别想把一个体会放在最后很多人做多Agent系统一上来就关心模型能力、提示词技巧这些当然重要但真正让系统能扛住生产环境的是编排和持久化这些听起来不那么“AI”的东西。我踩过几次坑之后才真正想明白——多Agent协作的本质不是让Agent更聪明而是给它们一套不会断的“记忆”和不会乱的“秩序”。OpenRig这套方案不一定适合所有人但如果你也在被“Agent单跑很好、协作就翻车”折磨不妨先别动Agent本身去看看你的编排图是否清晰、状态是否持久、审计是否完整。把这三件事做扎实了Agent协作就会从“碰运气”变成“可预期”。