Agent 工程体系 · 第 66/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

多 Agent 共享状态：所有权、版本、冲突、锁和事件溯源

多 Agent 系统的难点通常不在于“如何让多个 Agent 协同”，而在于：多个执行者同时观察、修改和解释同一份状态时，系统究竟依据什么判断结果有效。

一个 Agent 可以把数据库记录、任务队列、会话上下文、工具返回值、审批结果或文件系统视为“状态”。当系统只有一个 Agent 时，状态通常可以隐藏在一次运行的上下文中；当系统出现 Supervisor、专用 Agent、异步消费者和重试机制后，状态就变成了一个分布式系统问题：

• 谁可以修改某个字段？
• 一个 Agent 看到的状态是否仍然有效？
• 两个 Agent 同时修改时，哪个结果应该保留？
• 锁失效后，旧 Agent 是否仍然可能写入？
• 事件已经发送但状态尚未更新时，消费者应该相信什么？
• 系统如何解释“为什么最终状态会变成这样”？

因此，共享状态不能简单理解为一个所有 Agent 都能读写的 JSON。更准确的模型是：

共享状态 = 数据本身 + 所有权规则 + 版本条件 + 冲突策略 + 并发控制 + 可追溯变更记录。

---

一、先区分三种“状态”

在多 Agent 系统中，至少存在三种不同含义的状态。把它们混在一起，是后续冲突和恢复困难的根源。

1. 工作状态：任务当前要做什么

工作状态描述任务在业务流程中的位置，例如：

{
  "task_id": "T-1001",
  "status": "awaiting_approval",
  "owner": "approval-agent",
  "revision": 7,
  "attempt": 2
}

其中：

• status 表示业务状态；
• owner 表示当前控制权或写入权；
• revision 表示状态版本；
• attempt 表示某个步骤被执行过几次。

工作状态必须满足业务不变量。例如：

status = awaiting_approval
=> owner 必须是 approval-agent 或 human-review

status = completed
=> 不允许再次进入 running

这里的“不变量”不是提示词，而是系统必须拒绝违反的条件。

2. 上下文状态：Agent 当前知道什么

上下文状态包括：

• 用户最近的输入；
• Supervisor 的路由判断；
• 专用 Agent 的中间结论；
• 工具调用结果；
• 尚未提交的计划；
• 对历史事件的摘要。

上下文是解释材料，不一定是权威事实。

例如，Agent 上一轮上下文中可能写着：

订单金额为 99 元，退款尚未执行。

但另一个 Agent 已经在数据库中完成退款。此时上下文仍然可以作为推理参考，却不能作为执行退款的依据。执行前必须重新读取权威状态。

3. 事实状态：外部系统已经发生了什么

事实状态由数据库、支付系统、工单系统、文件存储或其他外部服务决定。例如：

{
  "payment_id": "P-88",
  "refund_status": "succeeded",
  "refunded_amount": 99
}

事实状态与工作状态可能暂时不一致：

工作状态：refund_requested
支付系统：refund_status = succeeded

这不一定是错误，可能只是事件尚未被消费者处理。真正重要的是系统是否明确规定：

• 哪个字段是权威来源；
• 哪些状态允许短暂不一致；
• 谁负责修复不一致；
• 用户看到的是工作状态还是事实状态。

---

二、共享状态的最小形式化模型

设一个任务状态为：

$$S = (D, O, V, I)$$

其中：

• $D$：业务数据，例如订单金额、审批意见；
• $O$：所有权信息，例如当前 Agent、租约；
• $V$：版本信息；
• $I$：不变量集合。

一次 Agent 操作可以表示为：

$$T = (a, r, p, e)$$

其中：

• $a$：执行者身份；
• $r$：读取到的状态版本；
• $p$：计划执行的变更；
• $e$：可能产生的外部副作用，例如发起退款、发送邮件。

一个安全的状态提交必须满足：

$$\operatorname{commit}(T, S) \iff \operatorname{owner}(a, S) \land \operatorname{version}(r) = \operatorname{version}(S) \land \operatorname{valid}(p, S)$$

直觉上，这三个条件分别回答：

1. 你有没有资格改？
2. 你修改的是否还是你读取的那一版？
3. 修改后是否仍然满足业务规则？

只有“有资格”而没有版本检查，会产生丢失更新；只有版本检查而没有所有权检查，会产生越权写入；只有前两者而没有不变量检查，会产生业务上不合法的状态。

---

三、所有权：谁拥有“下一步控制权”

1. 所有权不是 Agent 名称

“当前 Agent 是 billing-agent”并不自动等于它拥有所有写权限。

至少需要区分四种权限：

权限	含义
读取权	可以查看状态
建议权	可以提出变更，但不能提交
修改权	可以提交指定字段
控制权	可以决定下一步由谁执行

例如：

research-agent:
  可以读取订单
  可以写入 research_result
  不能修改 refund_status
  不能把任务交给自己之外的 Agent

因此，所有权更适合建模为字段级或能力级权限：

{
  "owners": {
    "research_result": "research-agent",
    "refund_request": "billing-agent",
    "final_reply": "supervisor"
  }
}

2. 控制权与最终回复权

多 Agent 编排中，必须明确谁拥有对用户的最终回复权。

常见有两种模式：

• Handoff：控制权移交给专用 Agent，由专用 Agent 继续对话并负责下一次响应；
• Agents as tools：Supervisor 保持控制权，把其他 Agent 当作受限工具调用，最终回复仍由 Supervisor 生成。

OpenAI Agents SDK 当前文档也把这两种模式区分为“专用 Agent 接管会话”和“Manager 保持回复所有权”。(developers.openai.com)

这一区分不仅影响用户体验，也影响状态写入：

handoff:
  supervisor -> billing-agent
  billing-agent 可以拥有当前对话分支的回复权

agents-as-tools:
  supervisor -> 调用 billing-agent
  billing-agent 只返回结果，不拥有外层回复权

如果这两种语义没有区分，容易出现：

1. Supervisor 认为自己仍然负责回复；
2. 专用 Agent 也认为自己已经接管；
3. 两者同时生成用户消息；
4. 一个 Agent 的结果覆盖另一个 Agent 的状态。

3. 所有权转移必须是状态变更

错误做法是只在内存中执行：

current_agent = "billing-agent"

然后假设其他进程会知道所有权已经变化。

正确做法是将所有权转移写入持久状态：

UPDATE tasks
SET owner_agent = 'billing-agent',
    revision = revision + 1,
    updated_at = now()
WHERE task_id = 'T-1001'
  AND owner_agent = 'supervisor'
  AND revision = 6;

如果影响行数为 0，说明至少有一种情况发生：

• Supervisor 已经不是当前所有者；
• 任务版本已变化；
• 任务不存在；
• 其他 Agent 已经抢先完成了转移。

所有权转移本身也需要原子条件，否则两个 Supervisor 可能同时把任务交给不同 Agent。

---

四、版本：避免 Agent 覆盖自己不知道的修改

1. 丢失更新的完整过程

假设数据库中初始状态为：

{
  "revision": 10,
  "priority": "normal",
  "customer_note": ""
}

两个 Agent 几乎同时读取：

research-agent 读取 revision=10
billing-agent 读取 revision=10

随后：

research-agent 修改 priority=high
billing-agent 修改 customer_note="客户要求退款"

如果系统直接保存完整 JSON：

research-agent 保存 revision=11
billing-agent 保存 revision=11

最终结果可能是：

{
  "revision": 11,
  "priority": "normal",
  "customer_note": "客户要求退款"
}

priority=high 被覆盖了。这个问题称为丢失更新。

2. 乐观并发控制

最常见的解决方式是乐观并发控制，也称为 CAS（Compare-And-Swap）式提交。

Agent 保存自己读取的版本：

UPDATE tasks
SET priority = 'high',
    revision = revision + 1
WHERE task_id = 'T-1001'
  AND revision = 10;

如果返回：

UPDATE 1

表示提交成功。

如果返回：

UPDATE 0

表示版本已经变化。Agent 不能直接覆盖，而应该：

1. 重新读取当前状态；
2. 判断自己的变更是否仍然适用；
3. 生成新的变更；
4. 再次提交，或将冲突交给 Supervisor / 人工处理。

完整过程如下：

初始：revision=10

research-agent 读取 revision=10
billing-agent 读取 revision=10

research-agent CAS revision=10 -> 成功，revision=11
billing-agent CAS revision=10 -> 失败

billing-agent 重新读取 revision=11
billing-agent 合并 customer_note
billing-agent CAS revision=11 -> 成功，revision=12

乐观并发控制的前提是：冲突并不频繁，且失败后可以安全重算。它不要求 Agent 长时间占有资源，因此适合长时间运行的 Agent。

3. 字段版本与整体版本

整体版本简单，但粒度较粗：

任何字段发生变化，revision 都递增。

如果两个 Agent 修改完全不相关的字段，它们仍然会发生版本冲突。

可以采用字段级版本：

{
  "versions": {
    "priority": 4,
    "customer_note": 2,
    "refund_status": 8
  }
}

或者把状态拆成不同聚合：

task_core
task_research
task_billing
task_approval

这样可以减少不相关字段之间的冲突，但会增加跨聚合一致性问题。拆分不是免费的性能优化，而是把“同一行上的冲突”改成“多个聚合之间的协调”。

4. 版本号不等于时间戳

错误做法：

WHERE updated_at = '2026-09-01 10:00:00'

时间戳可能存在：

• 精度不足；
• 多台机器时钟偏差；
• 相同时间戳；
• 数据库与应用时区差异。

版本号表达的是逻辑顺序，而不是物理时间：

revision=11 发生在 revision=10 之后

如果需要表达分支关系，可以使用向量时钟或因果父版本：

{
  "event_id": "E-20",
  "parent_revision": 10,
  "revision": 11
}

对于大多数任务状态，单调递增的聚合版本已经足够；对于多主写入、离线合并或跨区域并行写入，单一整数版本可能无法表达两个分支同时产生的事实。

---

五、冲突：不是所有冲突都应该“最后写入者获胜”

1. 三类冲突

写写冲突

两个 Agent 修改同一字段：

A: refund_status = approved
B: refund_status = rejected

这通常需要业务裁决，不能依赖数据库顺序。

读写冲突

Agent 根据旧状态做出决策，但提交时状态已经变化：

A 读取 balance=100
B 扣款 80
A 根据旧值尝试扣款 50

即使 A 修改的是不同字段，也可能违反业务不变量。

语义冲突

两个结果在数据结构上可以同时保存，但业务意义互相矛盾：

{
  "research_result": "供应商支持退款",
  "billing_result": "当前订单不可退款"
}

这不是数据库层面的冲突，而是需要一个裁决步骤。简单地将两个字符串拼接起来，只会把不确定性推迟到最终回复。

2. 冲突解决的四种策略

拒绝并重试

适用于变更可重新计算的情况：

版本冲突 -> 重新读取 -> 重新规划 -> 再提交

字段合并

适用于互不相关的字段：

A 修改 priority
B 修改 customer_note
=> 两者合并

但字段合并必须由程序定义，不能让 Agent 自由判断任意 JSON 是否可合并。

单调状态规则

有些字段只能向前推进：

pending -> approved -> executed -> completed

可以定义偏序：

$$pending \prec approved \prec executed \prec completed$$

若收到旧状态：

当前：executed
事件：approved

系统应丢弃或记录该事件，而不是回退到 approved。

注意：不是所有业务状态都天然单调。例如订单可能从 approved 进入 cancelled，这不是“版本回退”，而是另一条合法路径。状态机必须明确列出允许的边。

业务裁决

当冲突涉及不可自动判断的语义时，交给：

• Supervisor；
• 规则引擎；
• 人工审批；
• 专门的评估 Agent。

裁决输入不能只有两个自然语言结论，还应包含：

{
  "task_id": "T-1001",
  "base_revision": 10,
  "candidate_a": {
    "agent": "research-agent",
    "claim": "可退款",
    "evidence": ["policy-42"]
  },
  "candidate_b": {
    "agent": "billing-agent",
    "claim": "不可退款",
    "evidence": ["order-883"]
  }
}

这样裁决结果能够引用证据，而不是在缺少来源的情况下再次猜测。

---

六、锁：限制同时进入，但不能保证结果正确

1. 悲观锁的语义

悲观锁假设冲突可能发生，因此在修改前先锁住资源：

BEGIN;

SELECT *
FROM tasks
WHERE task_id = 'T-1001'
FOR UPDATE;

UPDATE tasks
SET status = 'approved',
    revision = revision + 1
WHERE task_id = 'T-1001';

COMMIT;

FOR UPDATE 的作用是让同一数据库事务中的其他事务等待或失败。它适合：

• 临界区较短；
• 状态变更必须串行；
• 冲突频率较高；
• 所有修改都经过同一个数据库。

但不能在锁内执行长时间模型调用：

BEGIN
SELECT FOR UPDATE
调用 LLM，耗时 30 秒
调用支付服务，耗时 10 秒
COMMIT

这会导致：

• 数据库连接长期占用；
• 其他 Agent 大量等待；
• 锁竞争放大延迟；
• Agent 超时后事务回滚，但外部服务可能已经执行。

正确的方式是将“推理”和“提交”分开：

1. 读取状态 revision=10
2. 在事务外调用 LLM
3. 得到计划
4. 开启短事务
5. 使用 revision=10 条件提交
6. 版本冲突则重新判断

2. 分布式锁不是自动安全

当资源跨越多个服务时，工程上常使用分布式锁或租约。租约是带过期时间的锁：

{
  "resource": "task:T-1001",
  "holder": "agent-run-abc",
  "expires_at": "2026-09-01T10:05:00Z"
}

租约解决的是“持有者失联后最终释放”，但引入了一个危险情况：

A 获得租约
A 暂停很久，超过租约时间
B 获得新租约
A 恢复运行，继续写入

此时仅检查 holder=A 可能不够，因为 A 的身份仍然可能保存在旧上下文中。

3. 围栏令牌：防止旧持有者写入

为每次租约发放单调递增的 fencing token：

A 获得租约，token=41
租约过期
B 获得租约，token=42
A 恢复

所有写操作都必须携带令牌，存储层只接受不小于当前令牌的写入：

UPDATE tasks
SET status = 'completed',
    lease_token = 42,
    revision = revision + 1
WHERE task_id = 'T-1001'
  AND lease_token < 42;

A 携带 token=41 的写入会被拒绝，即使它仍然认为自己是任务所有者。

围栏令牌的核心不是“锁永远不失效”，而是：

锁失效后，旧执行者即使恢复，也没有资格继续写入。

4. 锁不能覆盖外部副作用

数据库锁只能约束数据库中的并发，不能自动回滚已经发出的 HTTP 请求：

事务获得锁
调用支付退款 API
进程崩溃
数据库事务回滚
支付平台却已经退款成功

因此，外部副作用必须设计为幂等操作，并在状态中记录意图和结果：

{
  "operation_id": "refund:T-1001:99",
  "desired": "refund",
  "provider_request_id": "R-7788",
  "status": "unknown"
}

恢复时不是“再执行一次退款”，而是：

1. 使用同一个 operation_id 查询或重试；
2. 让支付服务按幂等键返回同一结果；
3. 将本地状态推进到已确认结果。

---

七、事件溯源：把状态变化保存为事实序列

1. 当前状态与事件日志的区别

普通 CRUD 保存的是当前结果：

tasks.status = completed

事件溯源保存的是导致结果的事实：

TaskCreated
ResearchCompleted
RefundApproved
RefundRequested
RefundSucceeded

当前状态由事件重放得到：

$$S_n = \operatorname{fold}(S_0, E_1, E_2, \ldots, E_n)$$

其中：

• $S_0$ 是初始状态；
• $E_i$ 是第 $i$ 个事件；
• fold 是按顺序应用事件的状态转换函数；
• $S_n$ 是当前状态。

例如：

S0 = status=pending, amount=99

E1 = TaskCreated
S1 = status=pending, amount=99

E2 = RefundApproved
S2 = status=approved, amount=99

E3 = RefundRequested
S3 = status=requested, amount=99

E4 = RefundSucceeded
S4 = status=completed, amount=99

事件溯源的价值不只是“方便审计”。它还允许系统回答：

• 哪个 Agent 提交了这个决定？
• 它当时看到了什么版本？
• 哪个工具调用产生了证据？
• 状态是正常推进，还是补偿操作后恢复？
• 消费者是否重复处理过事件？

2. 事件必须是事实，不是命令

以下内容是命令：

Please refund order T-1001

以下内容才是事实事件：

{
  "type": "RefundRequested",
  "task_id": "T-1001",
  "amount": 99,
  "actor": "billing-agent",
  "causation_id": "run-abc",
  "correlation_id": "conversation-xyz"
}

命令表达“希望发生什么”，事件表达“已经发生了什么”。

如果支付请求只是发出、尚未确认，就不能记录为 RefundSucceeded，最多记录：

RefundRequested

否则消费者会把“请求成功”误判为“业务成功”。

3. 事件结构

一个可用于生产诊断的事件至少需要：

{
  "event_id": "E-100",
  "event_type": "RefundRequested",
  "aggregate_id": "T-1001",
  "aggregate_type": "task",
  "aggregate_revision": 12,
  "actor_type": "agent",
  "actor_id": "billing-agent",
  "run_id": "run-abc",
  "correlation_id": "conversation-xyz",
  "causation_id": "E-99",
  "occurred_at": "2026-09-01T10:00:00Z",
  "payload": {
    "amount": 99
  }
}

几个 ID 的职责不同：

• event_id：唯一标识这条事件；
• aggregate_id：事件属于哪个任务或业务聚合；
• run_id：哪一次 Agent 运行产生；
• correlation_id：哪条用户请求或业务流程相关；
• causation_id：哪条事件或命令直接导致本事件。

不要用一个 trace_id 替代全部字段。链路追踪、业务关联和因果关系是三个不同维度。

---

八、事件驱动 Agent 中的 Topic、消费者、顺序和最终一致性

1. Topic 与消费者

事件驱动架构通常把事件发布到 Topic：

task-events

不同消费者订阅自己关心的事件：

approval-consumer
billing-consumer
notification-consumer
projection-consumer

数据流可以表示为：

flowchart LR
    U[用户请求] --> S[Supervisor]
    S -->|提交命令| C[(Command Store)]
    C --> A[Aggregate Writer]
    A -->|追加事件| T[(Task Event Topic)]

    T --> B[Billing Consumer]
    T --> P[Projection Consumer]
    T --> N[Notification Consumer]

    B -->|外部退款| Pay[支付系统]
    B -->|发布结果事件| T
    P --> R[(Read Model)]
    N --> M[消息通知]

关键点是：消费者不是直接共享内存，而是通过事件观察状态变化。因此消费者天然面对延迟、重复、乱序和失败重试。

2. 顺序只在有限范围内成立

许多消息系统只能保证同一分区或同一聚合键内的顺序，而不能保证全局顺序。

如果事件按 aggregate_id=T-1001 分区：

T-1001:
  E1 TaskCreated
  E2 RefundRequested
  E3 RefundSucceeded

可以保证该任务内部的顺序。

但以下两个任务之间没有自然顺序：

T-1001: RefundSucceeded
T-1002: RefundRequested

因此，业务逻辑不应依赖全局事件序号。消费者应检查：

{
  "aggregate_id": "T-1001",
  "aggregate_revision": 14
}

若当前投影版本是 12，收到版本 14 时不能直接应用，除非系统允许跳过；更稳妥的处理是等待版本 13 或进入重建队列。

3. 幂等消费者

消费者可能收到同一事件多次：

E-100 第一次处理成功
消费者提交 offset 前崩溃
E-100 再次投递

因此必须使用事件 ID 去重：

CREATE TABLE processed_events (
    consumer_name TEXT NOT NULL,
    event_id TEXT NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer_name, event_id)
);

处理流程：

BEGIN;

INSERT INTO processed_events(consumer_name, event_id)
VALUES ('billing-consumer', 'E-100')
ON CONFLICT DO NOTHING;

如果插入结果为 0 行，说明该消费者已经处理过事件，应直接确认消息，不再重复执行。

如果插入成功，则继续执行状态更新和业务操作。这里仍有一个边界：数据库事务无法自动覆盖外部支付调用。因此，外部操作必须使用幂等键，并将“已请求但结果未知”作为合法状态。

4. 最终一致性不是“最终一定正确”

最终一致性只表示：在没有新的写入、消息最终可达且消费者最终成功的前提下，不同副本可能收敛到同一结果。

它不保证：

• 用户下一秒读取到最新状态；
• 消费者不会永久失败；
• 错误事件会自动被纠正；
• 两个互相矛盾的事实能够自动合并。

例如：

10:00:00 退款请求事件写入
10:00:01 用户查询 read model，仍显示 pending
10:00:05 支付系统返回成功
10:00:06 read model 更新为 completed

这段时间内，系统处于可接受的不一致状态，前提是 API 明确区分：

{
  "status": "processing",
  "source": "read-model",
  "last_applied_revision": 12
}

如果用户要求强一致结果，应直接查询权威聚合或等待指定版本：

GET task/T-1001?min_revision=14

当读模型尚未追上 14 时，服务可以等待、返回处理中，或降级到权威存储查询。

---

九、事件存储与当前状态表可以同时存在

事件溯源并不意味着每次查询都从头重放全部事件。实际系统通常同时保存：

事件日志：完整变更历史
快照：某个版本的聚合状态
读模型：面向查询的投影

重建过程为：

读取 snapshot revision=100
重放 revision=101 之后的事件
生成当前状态

快照必须注明版本：

{
  "aggregate_id": "T-1001",
  "revision": 100,
  "state": {
    "status": "approved"
  }
}

不能把快照当成永远正确的缓存。代码升级后，新的事件解释逻辑可能改变，因此需要考虑：

• 事件版本；
• 事件升级器；
• 快照失效策略；
• 重放测试；
• 部分事件无法重放时的人工修复。

一种常见数据库结构如下：

CREATE TABLE task_events (
    aggregate_id TEXT NOT NULL,
    revision BIGINT NOT NULL,
    event_id TEXT NOT NULL UNIQUE,
    event_type TEXT NOT NULL,
    actor_id TEXT NOT NULL,
    run_id TEXT NOT NULL,
    correlation_id TEXT NOT NULL,
    causation_id TEXT,
    payload JSONB NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (aggregate_id, revision)
);

CREATE TABLE task_snapshots (
    aggregate_id TEXT PRIMARY KEY,
    revision BIGINT NOT NULL,
    state JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

追加事件时，(aggregate_id, revision) 的唯一约束可以防止两个写入者提交同一个聚合版本：

INSERT INTO task_events (
    aggregate_id,
    revision,
    event_id,
    event_type,
    actor_id,
    run_id,
    correlation_id,
    causation_id,
    payload
)
VALUES (
    'T-1001',
    13,
    'E-101',
    'RefundRequested',
    'billing-agent',
    'run-abc',
    'conversation-xyz',
    'E-100',
    '{"amount": 99}'
);

如果插入因主键冲突失败，说明另一个执行者已经占用了版本 13。当前 Agent 必须重新读取，而不能强行使用自己的事件覆盖它。

---

十、事务性 Outbox：避免“状态成功但事件丢失”

考虑以下错误流程：

1. 更新 tasks.status = approved
2. 提交数据库事务
3. 发布 ApprovalGranted 事件
4. 进程在第 3 步崩溃

结果是数据库显示已批准，但下游永远不知道。

反过来也有另一种错误：

1. 发布 ApprovalGranted 事件
2. 数据库事务回滚

下游收到了一条没有对应持久状态的事件。

事务性 Outbox 将状态更新和待发送事件写入同一个事务：

BEGIN;

UPDATE tasks
SET status = 'approved',
    revision = revision + 1
WHERE task_id = 'T-1001'
  AND revision = 12;

INSERT INTO outbox_events (
    event_id,
    topic,
    aggregate_id,
    payload,
    published
)
VALUES (
    'E-101',
    'task-events',
    'T-1001',
    '{"type":"ApprovalGranted","revision":13}',
    false
);

COMMIT;

后台发布器随后读取：

SELECT *
FROM outbox_events
WHERE published = false
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 100;

发布成功后再标记：

UPDATE outbox_events
SET published = true,
    published_at = now()
WHERE event_id = 'E-101';

发布器崩溃可能导致事件重复发布，因此消费者仍然必须幂等。Outbox 解决的是“事件不会因为数据库提交后进程崩溃而永久丢失”，不是“事件只会发布一次”。

---

十一、Supervisor、Handoff 与共享状态的关系

Supervisor 不应被理解为“一个更聪明的 Agent”，而应被理解为一种控制结构：

接收请求
读取任务状态
决定下一步控制权
启动或唤醒专用 Agent
验证结果
推进状态
决定是否结束、重试或转人工

1. Handoff 的状态变化

一个明确的 handoff 可以表示为：

revision=5
owner=supervisor
status=triaging

提交 HandoffIssued:
revision=6
owner=billing-agent
status=billing_pending

Billing Agent 返回后，不应只把自然语言结果交给 Supervisor，而应该提交结构化事件：

{
  "type": "BillingAssessmentCompleted",
  "aggregate_id": "T-1001",
  "base_revision": 6,
  "decision": "eligible",
  "evidence": ["policy-42", "order-883"]
}

Supervisor 根据事件决定：

eligible -> request_refund
ineligible -> explain_rejection
uncertain -> human_review

2. Agents as tools 的状态边界

当专用 Agent 作为工具运行时，它通常不应该直接修改外层任务状态：

Supervisor
  -> 调用 research-agent
  -> 得到 research_result
  -> Supervisor 验证并提交 TaskResearchCompleted

这样做的好处是外层状态只有一个提交者，专用 Agent 的输出是候选结果，而不是未经验证的事实。

但如果专用 Agent 承担长时间异步任务，它也可以拥有自己的子任务聚合：

主任务 T-1001
  子任务 R-1：research-agent
  子任务 B-1：billing-agent

此时不能让多个 Agent 直接写主任务任意字段，而应让各自更新自己的子任务，再由 Supervisor 或聚合器根据子任务事件推进主任务。

---

十二、如何防止共享状态造成死循环

多 Agent 死循环不一定表现为 Agent 之间互相 handoff，也可能表现为事件消费者不断产生新事件：

A 更新状态
-> 触发 B
-> B 重新规划并更新状态
-> 再次触发 A
-> A 认为需要重试
-> ...

必须为流程建立可验证的终止条件。

1. 进度度量

定义一个任务进度函数：

$$P(S) = (r, c, q)$$

例如：

• $r$：聚合版本；
• $c$：已完成检查数量；
• $q$：剩余待处理步骤数量。

每次自动重试至少应满足以下之一：

revision 增加且状态发生合法推进
检查数量增加
剩余问题减少
获得新的外部证据

如果 Agent 连续多次产生等价状态：

status 不变
evidence 集合不变
next_action 不变

则不应继续调用模型，而应进入：

blocked
needs_human_review
retry_exhausted

2. 显式预算

每次运行都应有预算：

{
  "max_model_turns": 12,
  "max_tool_calls": 30,
  "max_retries_per_step": 3,
  "deadline": "2026-09-01T10:30:00Z"
}

预算是运行时安全边界，不是提示词中的一句“不要无限循环”。模型可能无法可靠地维护计数，因此计数必须由外部运行时强制执行。Anthropic 的 Agent 指南也强调，Agent 依赖环境反馈循环执行，并应设置最大迭代次数等停止条件。(anthropic.com)

3. 因果链去环

每个事件带有 causation_id，消费者可以检查：

当前事件 E-20
因果链：E-20 -> E-19 -> E-18 -> ...

如果发现某类事件重复形成环：

HandoffIssued(supervisor -> billing)
BillingBlocked(billing -> supervisor)
HandoffIssued(supervisor -> billing)

可以按：

• 相同 aggregate_id；
• 相同 step_name；
• 相同输入摘要；
• 相同工具参数；

识别等价重入，并停止自动处理。

---

十三、一个最小的可靠提交流程

下面的伪代码展示了 Agent 不直接覆盖状态，而是通过版本和所有权提交变更：

def commit_transition(
    db,
    task_id: str,
    agent_id: str,
    expected_revision: int,
    expected_owner: str,
    next_status: str,
    event_type: str,
    payload: dict,
):
    with db.transaction():
        task = db.fetch_one(
            """
            SELECT status, owner_agent, revision
            FROM tasks
            WHERE task_id = %s
            FOR UPDATE
            """,
            [task_id],
        )

        if task is None:
            raise ValueError("task_not_found")

        if task["owner_agent"] != expected_owner:
            raise Conflict("ownership_changed")

        if task["revision"] != expected_revision:
            raise Conflict("revision_changed")

        validate_transition(
            old_status=task["status"],
            new_status=next_status,
            actor=agent_id,
        )

        new_revision = expected_revision + 1

        db.execute(
            """
            UPDATE tasks
            SET status = %s,
                revision = %s,
                updated_at = now()
            WHERE task_id = %s
            """,
            [next_status, new_revision, task_id],
        )

        db.execute(
            """
            INSERT INTO task_events (
                aggregate_id,
                revision,
                event_id,
                event_type,
                actor_id,
                run_id,
                correlation_id,
                payload
            )
            VALUES (%s, %s, gen_random_uuid(), %s, %s, %s, %s, %s)
            """,
            [
                task_id,
                new_revision,
                event_type,
                agent_id,
                current_run_id(),
                current_correlation_id(),
                payload,
            ],
        )

        db.execute(
            """
            INSERT INTO outbox_events (
                event_id,
                topic,
                aggregate_id,
                payload,
                published
            )
            VALUES (lastval(), 'task-events', %s, %s, false)
            """,
            [task_id, payload],
        )

    return new_revision

这个流程中，每一步都有明确作用：

1. FOR UPDATE 保证当前短事务内不会有另一个事务同时修改同一行；
2. owner_agent 检查防止越权；
3. revision 检查防止基于旧状态提交；
4. validate_transition 保证状态机合法；
5. 当前状态、事件和 Outbox 在同一事务中提交；
6. 事务外的发布器负责把 Outbox 发送到 Topic；
7. 消费者根据 event_id 幂等处理。

这不是某个 Agent 框架的固定 API，而是共享状态层应该提供的语义。框架可以负责运行 Agent、管理 handoff 和工具调用，但不能替业务系统定义订单退款、审批或支付状态的正确性。

---

十四、常见错误与诊断方式

错误一：所有 Agent 共享一个可变 JSON

表现：

Agent A 读 JSON
Agent B 读 JSON
Agent A 改一部分
Agent B 用旧 JSON 整体覆盖

诊断：

• 检查写入是否携带 expected_revision；
• 检查是否保存完整快照而不是结构化变更；
• 对比审计日志中相邻写入的读取版本。

修复：

• 使用 CAS；
• 使用字段级命令；
• 对重要聚合使用事件追加；
• 禁止 Agent 直接提交任意 JSON Patch。

错误二：用最后写入者获胜解决业务冲突

表现：

approved 被 rejected 覆盖
completed 被 pending 覆盖

诊断：

SELECT aggregate_id, revision, event_type, actor_id, occurred_at
FROM task_events
WHERE aggregate_id = 'T-1001'
ORDER BY revision;

如果事件顺序合法但状态回退，说明状态转换函数没有拒绝非法边。

修复：

• 为状态定义显式状态机；
• 为单调字段定义偏序；
• 对语义冲突进入裁决流程。

错误三：锁持有期间调用模型或外部 API

表现：

• 数据库锁等待升高；
• Agent 超时；
• 事务回滚但外部动作已发生；
• 重试导致重复扣款或重复通知。

修复：

• 锁内只做短事务；
• 外部动作采用幂等键；
• 使用 Outbox、操作记录和结果查询；
• 对未知结果设计恢复状态。

错误四：事件只记录“Agent 说了什么”

例如：

{
  "message": "我认为可以退款"
}

这无法区分：

• 建议；
• 已批准；
• 已发起退款；
• 支付系统已成功；
• Agent 是否基于旧版本判断。

修复是把事件拆成事实边界：

RefundEligibilityAssessed
RefundApproved
RefundRequested
RefundSucceeded
RefundFailed

每个事件包含版本、执行者、因果 ID 和证据引用。

错误五：把最终一致性隐藏起来

表现：

用户刚看到“退款成功”
刷新页面却变成“处理中”

这可能不是数据错，而是读取了落后的投影。但如果 API 没有暴露来源版本，用户和运维都无法判断。

至少应记录：

{
  "status": "processing",
  "aggregate_revision": 14,
  "read_model_revision": 13,
  "consistency": "eventually_consistent"
}

当用户需要强一致读取时，服务应等待或读取权威源，而不是假装投影已经最新。

---

十五、规范保证、实现选择和经验建议

需要明确区分三种层次。

规范保证

这些必须由系统契约保证：

• 未持有权限的 Agent 不能修改受保护字段；
• 旧版本不能覆盖新版本；
• 非法状态转换必须被拒绝；
• 事件 ID 在同一消费者内可去重；
• 外部副作用具有明确的幂等语义；
• 每次状态变化可以追溯到执行者和因果链。

常见实现

这些是常用方案，但不是唯一方案：

• PostgreSQL 行锁；
• revision 整数版本；
• Outbox；
• processed_events 去重表；
• Topic 按聚合 ID 分区；
• 快照加事件重放；
• 租约加 fencing token。

经验建议

这些取决于业务边界：

• 低冲突、长任务优先考虑乐观并发控制；
• 高冲突、短临界区可以使用悲观锁；
• 需要完整审计和重建时采用事件溯源；
• 查询量大时使用读模型和最终一致性；
• 语义冲突不要交给“最后写入者”，而应建立裁决步骤；
• 如果拆分 Agent 后没有新增能力隔离、策略隔离或清晰的状态边界，就不应为了形式上的多 Agent 而拆分。

Anthropic 的经验总结也是先采用简单、可组合的工作流，仅在确实需要动态决策和开放式执行时增加 Agent 复杂度；其文章同时提醒，Agent 的自主循环会带来更高成本和错误累积风险。(anthropic.com) OpenAI Agents SDK 的文档则把 Agent 定义、运行时状态、编排与 handoff 分成不同层次，说明“谁负责回复”和“如何持久化共享状态”本身就是两个需要分别设计的问题。(developers.openai.com)

多 Agent 共享状态的最终目标，不是让所有 Agent 看到同一份数据，而是让每一次观察、决策、提交和副作用都具备可验证的边界：

谁在什么版本上做了什么决定
-> 依据什么证据
-> 由哪个状态转换接受
-> 产生了哪些事件
-> 哪些消费者已经处理
-> 如果失败，如何重试或恢复

当这些问题都能由状态模型和事件记录回答时，多 Agent 才真正从“几个模型互相传话”变成了可以诊断、恢复和持续演进的可靠执行系统。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：多 Agent 辩论与 Critic：独立证据、聚合、成本和伪共识
• 下一篇：事件驱动 Agent：Topic、消费者、关联 ID、顺序和最终一致性
• 延伸：多 Agent Supervisor 与 Handoff：控制权、上下文、返回和死循环

官方资料

• OpenAI Agents Guide
• Anthropic Building Effective Agents

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。