企业官网建设流程全解析

今年我在团队里主要折腾一件事：把零散的AI Agent落成能扛事儿的系统。项目代号OpenRig，目标很简单——做一套多智能体编排实践，把那些互相不认识的离散AI Agent，织成一张持久化协作系统的网。这篇文章不是产品发布会，是我整理出来的设计思路、落地流程和排障记录。做AI应用落地的人，尤其是被多Agent协作、任务恢复、中间状态丢失坑过的人，应该能从里面拿走一些东西。

先说结论：单个Agent的能力再强，解决不了“集成”的问题。OpenRig真正解决的是通信、状态、记忆这三件事。通信解决Agent之间怎么喊话，状态解决任务跑到哪一步，记忆解决Agent换实例之后怎么想起之前聊到哪。下面我会从设计动机、核心模块、完整实操和问题排查四个方面，把整个实践过程摊开讲。

1. 为什么需要把一堆Agent“织”起来

1.1 单Agent在真实业务里撑不过三个来回

我们最开始也是老老实实跑单Agent。流程不复杂：过一个客服工单，先查订单，再算运费，最后生成退款单。demo阶段一切正常，模型调用一次LLM，工具调用几次API，看起来挺聪明。但一放到真实业务里，问题马上暴露。

第一是上下文撑不住。对话超过十轮，早先用户填的地址、商品ID、优惠券信息就开始模糊。模型不是不记得，是它的“注意力”被后面的内容稀释了。你让它在一次会话里又检索又计算又填表，越往后越容易丢关键字段。

第二是会话太脆弱。一次普通网络抖动，LLM接口超时，整个任务直接失败。客户端重试，模型重新开始推理，但前面的工具调用结果已经丢了。用户得从头再描述一遍问题，这在内部流程自动化里几乎是不可接受的。

第三是没法拆人。很多任务天然需要不同角色协作：一个Agent做数据清洗，一个Agent写报告，一个Agent做合规检查。如果全塞在一个Prompt里，Prompt会膨胀到几千行，改一个角色逻辑就要重新调整个系统。

所以我们后来把每个能力拆成独立Agent。拆完之后更痛苦：调研Agent做完了，写作Agent不知道；写作Agent写完了，审核Agent拿不到结果。各自跑得欢，整体是一盘散沙。这也是“离散AI Agent”这个词的由来——每个节点是好的，连不成线。

1.2 编排的隐藏底座：不只是“喊人干活”

我一开始理解的编排，就是让A调用B，B调用C。于是写了各种内部HTTP接口，A处理完直接POST给B。这在两个Agent的时候挺顺，三个Agent也凑合，五个Agent以上就开始失控。

失控的原因在于：直接调用只解决了“数据传递”，没解决“契约管理”。B接口挂了怎么办？B处理到一半超时怎么办？B返回的结果A没收到，重发是幂等的吗？这些全是问题。

我后来想明白一个类比：一个十几人的项目组，不可能所有人都靠一对一喊话协作。必须有工单系统、共享文档、周报、会议纪要。任务派给谁、做到什么程度、卡在哪个环节，这些信息不在任何一个人的脑子里，而是沉淀在系统里。

这就是持久化协作系统的基本要求。OpenRig把“项目组”那套运作机制搬到了Agent世界里。每个Agent不是直接用函数调用去喊别人，而是往公共总线上投递事件，从公共总线上订阅自己关心的事件。谁处理、处理到什么状态、为什么失败，全部有记录。这样系统不再是靠一组脆弱的实时连接维持，而是一个可以回放、恢复、审计的事件流。

1.3 选型：为什么OpenRig用事件驱动而不是静态流程图

在动手前我们也评估了LangGraph这类图编排方案。坦白说，静态图对固定流程是合适的：节点确定，边确定，每个节点做什么一目了然。但真实的多Agent协作很快就会遇到麻烦。

比如审核不通过需要回退，回退之后可能还要人工介入，人工改完可能再触发新的分支。这些动态行为如果全画到图里，图会变成一团乱麻。更麻烦的是，图编排通常把执行状态保存在内存里，进程一重启，整个图的状态就丢了。

OpenRig换了个思路：不把流程定义成一张写死的图，而是用“事件驱动 + 状态机”来描述协作。每个Agent只关心自己订阅的事件，收到事件后做自己的事，再把结果作为新事件发出去。任务当前处于什么阶段，由独立的状态机管理。事件是事实，状态机是规则，Agent是执行者。三者拆开之后，协作系统的可维护性一下子高了很多。

我在后面会详细讲这三个模块。需要先说明的是，这个选型并不是说图编排一无是处。流程极简、场景固定的系统，静态图开发效率确实高。但如果你想做一个能长期演进、能容忍故障、能回溯审计的多Agent系统，事件驱动几乎是必经之路。

2. OpenRig的三个核心模块：总线、状态、记忆

2.1 消息总线：让Agent之间不直接喊话

OpenRig的消息总线默认使用NATS JetStream。为什么选它而不是Kafka或者Redis Stream？因为Kafka有点重，Redis Stream的持久化和消费组机制在丢消息场景下不够干净。NATS JetStream轻量、原生支持持久化、消费组和ack机制，对Agent之间的异步消息已经足够。

在总线上传递的消息格式是统一的。一个标准事件大概长这个样子：

{ "event_id": "evt_01HZ3K8X9XYZ", "task_id": "task_20250601_001", "producer": "researcher", "type": "task.research.completed", "payload": { "topic": "智能客服", "report": "..." }, "timestamp": "2025-06-01T10:30:00Z", "correlation_id": "req_00A1B2" }

event_id是全局唯一的事件标识。task_id用来标记这是哪一次业务任务。producer表示是哪个Agent产出的。type是事件类型，也是Agent订阅时的匹配依据。payload放实际业务数据。correlation_id是链路追踪ID，我要求所有Agent在消费和产出时都必须把它原样透传。

不要小看correlation_id。多Agent协作最怕的就是问题来了不知道怎么串链路。有了这个字段，你可以在日志系统里按correlation_id查询整个任务从创建到结束的所有事件。我们团队内部有个不成文的规定：任何一个Agent日志里没有correlation_id，这条日志就当作没打。

2.2 任务状态机：用乐观锁防住并发碰车

只有消息总线还不够，因为Agent收到消息之后，任务还是那个任务，它处于什么阶段必须有统一凭证。OpenRig里每个任务对应一张状态记录表，结构大概是这样：

状态流转被定义成一个状态机：pending -> assigned -> running -> succeeded，失败则进入failed，允许重试的话进入retrying，retrying之后回到assigned。状态机的好处是非法转移直接拒绝。比如一个已经succeeded的任务，不允许被另一个Agent改成running。

这里有一个特别值得讲的设计：乐观锁。每个Agent要更新任务状态时，必须带上自己读到的version。SQL是这样的：

UPDATE tasks SET current_status = 'succeeded', version = version + 1, updated_at = now() WHERE task_id = ? AND version = ?;

如果更新的行数是0，说明别人已经改过了，这个Agent就不能再覆盖，需要重新读取最新状态再决定下一步。为什么不直接用数据库行锁？因为Agent处理一次任务可能耗时几十秒甚至几分钟，你不可能让一个事务把行锁Hold那么久。悲观锁会浪费连接和数据库资源，乐观锁加版本号，冲突了重试就好。

2.3 事件溯源：把“记忆”从Agent脑子里搬出来

很多Agent框架会把对话历史、中间结果存在LLM调用的上下文里。这在单轮demo里没问题，但一旦Agent实例崩溃、容器重启、Session过期，记忆就没了。OpenRig的做法是事件溯源，所有Agent产生的中间结果，都作为事件追加写入event_store表。

这个表是一个append-only的日志。设计上只有插入操作，没有更新和删除。每次Agent完成一个阶段，就往里面追加一条事件。这样做有三个直接好处。

第一个是可恢复。新启动的Agent实例想了解任务上下文，不需要去翻旧Agent的内存，直接查该task_id的事件列表，按顺序回放一遍，当前状态自然就重建出来了。

第二个是可审计。谁在什么时间、基于什么输入、产出了什么内容，全部有记录。我们接审核类业务流程时，这个能力几乎是被业务方强制要求的。

第三个是可调试。系统出了奇怪问题，可以直接把指定任务的事件日志拉出来，一步一步看状态是怎么转移的。比起看代码推断，效率高太多了。

但事件溯源也有代价。任务跑得越久，事件越多，回放就越慢。OpenRig的解法是快照。系统每隔一段时间或者每积累N条事件，就把任务的最新状态生成一个checkpoint存下来。恢复的时候只需要加载最新快照，再回放快照之后的新事件就行。这个机制和数据库的WAL加checkpoint是同一个道理。

3. 实操：搭一个“调研-写作-审核”持久化协作链路

3.1 三个Agent的分工与协作协议

如果你手边没有具体场景，我建议先照着我这个例子跑一遍，改改就能用到自己的业务里。这个链路包含三个Agent：调研Agent、写作Agent、审核Agent。

需求是这样的：用户给定一个主题，调研Agent先搜集整理背景资料并生成报告；写作Agent看到报告之后，基于报告写一篇初稿；审核Agent检查初稿是否符合合规要求，通过就发布，不通过就要写上修改意见退回给写作Agent重新写。

这个例子覆盖了多Agent协作的三种典型交互：顺序传递、异步并行和回退重试。而且整个过程都是事件驱动，不存在任何一个Agent在自己的代码里写死“调用下一个Agent”。

3.2 环境准备：数据库、消息队列、OpenRig运行时

OpenRig核心是用Rust写的，提供Python SDK和命令行工具。为了演示方便，下面统一用Python SDK。起步前先把基础设施跑起来，我用Docker Compose。

version: "3" services: postgres: image: postgres:16 environment: POSTGRES_DB: openrig POSTGRES_USER: openrig POSTGRES_PASSWORD: openrig ports: - "5432:5432" nats: image: nats:2.10 command: ["-js", "-m", "8222"] ports: - "4222:4222" - "8222:8222"

启动服务之后，安装Python SDK。

pip install openrig-sdk

然后初始化OpenRig运行时。

openrig init --db postgres://openrig:openrig@localhost/openrig --nats nats://localhost:4222

init命令会创建事件表、任务表，以及必要的索引。这一步做完，系统骨架就立起来了。

3.3 配置Agent：订阅什么、产出什么、失败怎么办

OpenRig用一个YAML文件描述所有Agent的协作协议。协议的粒度是事件类型：每个Agent声明自己订阅什么事件、产出什么事件。这样后续新增Agent不需要改旧Agent的代码，改配置就行。

agents: researcher: subscribe: ["task.research.request"] publish: ["result.research"] max_retries: 3 timeout_sec: 600 prompt_template: "你是行业调研专家，围绕{topic}输出结构化报告" writer: subscribe: ["result.research"] publish: ["draft.article"] max_retries: 2 timeout_sec: 900 prompt_template: "基于调研报告撰写文章初稿，语言要专业" reviewer: subscribe: ["draft.article"] publish: ["review.result"] max_retries: 0 timeout_sec: 300 prompt_template: "请审核初稿，输出通过或修改意见" rules: - name: reject-to-writer when: 'review.result.status == "rejected"' action: "rewrite" target_agent: "writer"

这里再说一下rules的作用。审核Agent产出review.result事件，如果是个拒绝，OpenRig会根据规则把这个事件重新路由给writer，而不是让writer自己写代码轮询。也就是说，回退逻辑被收编到系统里了，Agent本身保持无状态。

配置写好之后，执行加载命令：

openrig apply -f agents.yaml

apply命令会把订阅关系和规则下发到运行时里，它会负责监听消息、投递给对应Agent。

3.4 写一个Agent的Python运行主循环

Agent这边不需要理解状态机整套细节，只需要写一个回调函数。以写作Agent为例：

from openrig import OpenRigClient client = OpenRigClient() @client.subscribe("result.research") async def handle_research(msg): task_id = msg["task_id"] topic = msg["payload"]["topic"] report = msg["payload"]["report"] # 恢复这个任务的历史上下文，防止实例重启后丢失 context = await client.load_context(task_id) history = context.events prompt = f""" 基于以下调研报告写一篇初稿。 主题：{topic} 报告：{report} 历史事件数：{len(history)} """ draft = await llm_generate(prompt) # 完成当前阶段，产出新事件 await client.complete( task_id=task_id, event_type="draft.article", payload={"draft": draft}, correlation_id=msg["correlation_id"], ) client.run()

注意几个关键点。load_context不是简单地从数据库捞一段文本，而是把该任务之前的事件列表传进来，让Agent能感知到上下文已经发生过什么。complete是标准动作，它内部会做三件事：追加事件、更新任务状态、把事件发到消息总线。如果Agent处理失败，OpenRig会按配置里的max_retries自动重试，重试次数用完就标记为failed。

所以Agent的代码可以写得很薄。它不需要关心消息可能重复投递，不需要关心状态机怎么转移，那些是运行时和状态机去协调的事情。

3.5 发布任务与观察状态流转

现在把整个链路跑起来。先创建一个任务。

curl -X POST http://localhost:8000/v1/tasks \ -H "Content-Type: application/json" \ -d '{ "type": "task.research.request", "payload": {"topic": "智能客服系统"} }'

返回结果里会有task_id。然后开另一个终端观察任务。

openrig task show task_20250601_001

正常情况下会看到任务状态一步步变化。刚开始是pending，调研Agent消费后变成assigned，再变running，最后succeeded。事件列表会记录每个阶段：调研完成、写作完成、审核通过。

如果审核拒绝，事件列表里能看到review.result的status是rejected，然后状态机会自动把事件重新投递给writer，任务状态进入retrying。整个过程不需要人工介入。

我这里强烈建议你把这个例子原封不动跑下来，然后故意让审核Agent的Prompt改成“永远拒绝”。你会看到系统自动回退三次，最后任务进入failed。这个“失败路径”比“成功路径”更能让你理解持久化协作系统为什么需要状态机和事件溯源。

4. 多智能体持久化排障实录

4.1 消息重复消费：幂等键救场

多Agent系统里，消息至少一次投递是常态。NATS JetStream的ack如果超时，或者消费端处理完后网络抖动，消息就会重投。于是同一个事件可能被同一个Agent消费两次。

我们遇到的实际案例：写作Agent处理调研报告时，因为数据库事务提交慢，触发了超时重投。结果两个进程同时消费了同一份报告，生成了两篇初稿，还都写进了事件表。

排查的时候才发现，问题不是NATS的问题，是我们自己的事件表没有幂等约束。后面在event_store表上加了一个唯一约束。

ALTER TABLE events ADD CONSTRAINT uq_task_event UNIQUE (task_id, event_id);

插入事件时使用ON CONFLICT DO NOTHING，消费时如果插入影响行数为0，说明已经处理过，直接ack并跳过。这样就算消息重复投递，事件也只会落库一次。

这个坑的教训是：不要指望消息队列帮你做到恰好一次，那个成本极高。分布式系统里更现实的方案是“至少一次”加上“业务幂等”。幂等键最好提前设计，到了线上再补，数据洗起来会很痛苦。

4.2 状态冲突和任务死锁：状态机不止画图

第二个高频问题是状态冲突。多个Agent并行跑的时候，两个Agent可能同时认领同一个任务。比如审核Agent和定时补偿Agent同时发现某个任务超时，都尝试把任务状态改成running。没有乐观锁的时候，后提交的会覆盖先提交的，owner_agent就乱了。

加上version乐观锁之后，冲突发生时更新行数为0，Agent就知道自己“抢任务失败”了。正确的处理是等一下随机时间重新读取任务，看是否还需要处理。不要立刻重试，否则两个Agent会一直碰撞。

还有一个比冲突更隐蔽的问题是任务级死锁。A任务在等B任务完成，B任务又在等A任务完成，双方都不进行下一步。纯消息总线系统里，这种死锁不会自动暴露。

OpenRig给每个任务加了一个心跳字段。任何Agent处理任务时都会定期刷新heartbeat。运行时有一条后台扫描任务，如果发现某个任务心跳长时间没更新，就把它标记为stalled。stalled状态会触发告警，业务上我们配置了统一补偿回调，把任务重新投递给空闲Agent。

设计上有一条铁律：禁止Agent在同步等待另一个Agent完成。所有跨Agent等待必须是异步事件驱动。你只要允许Agent写“我先睡一会儿，等对方处理完再继续”，死锁排查会变得无比痛苦。

4.3 并发与性能：怎么扛住一大波任务

很多人问过我“AI Agent怎么扛并发”，这里的瓶颈通常不在模型本身，而在编排层的数据库和消息队列。我们的event_store一开始是全表无脑写入，任务量上来之后，单表查询越来越慢，回放任务上下文时要扫描大量无关记录。

后来做了三件事。第一是按时间分区。event_store按月分区，老数据自动归档到冷分区，热分区保持较小体积。查询任务事件时走task_id加created_at的联合索引，回放速度快了很多。

第二是批量写。Agent的complete操作默认是单条事件落库，在高并发时数据库连接很容易被打满。OpenRig加了本地缓冲，攒够一批事件后统一批量insert。这个改动把数据库写入TPS提升了一个数量级。

第三是定期快照。前面说过事件溯源会积累大量事件，如果一个任务跑了几百个步骤，每次恢复都要回放几百条事件，太慢。快照机制让恢复操作变成“加载快照 + 回放增量”，绝大多数任务的恢复时间可以控制在几十毫秒。

做性能优化的时候要记住一点：先测量再优化。我们一开始凭感觉觉得是数据库慢，结果看了NATS的监控才发现是消费组不平衡导致部分Agent堆积。后来把消费组配置改了，问题就没了。持久化协作系统里，消息队列、数据库、Agent计算三个环节都可能成为瓶颈，别只盯着一个地方。

最后分享两点真实体会

做完OpenRig这套实践，我最大的体会有两点。第一，Agent的上下文不能当持久化存储用，状态和记忆要沉淀在系统里。模型会话是柔软易散的，系统状态必须是坚硬可查的，两者混在一起迟早出事。第二，多Agent协作的能力上限不取决于某个Agent有多聪明，而取决于通信、状态、记忆这些基础设施做得有多扎实。聪明是模型的，可靠是系统的。

最后再分享一个小技巧：调试这类系统时，事件日志一定要打全。一条事件里带上task_id和correlation_id还不够，还要在日志里打清楚“你是谁、你消费了哪个事件、你产出了哪个事件”。多Agent系统跟单机程序不一样，问题常常不在报错的地方，而在某条消息悄悄被丢弃的缝隙里。把这些细节做到位，系统才能从“demo很酷”走向“生产能用”。