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

Agent 状态机设计：节点、事件、守卫、转移和可恢复执行

Agent 不是“调用一次大模型并返回文本”的函数，而是一个持续推进任务的运行过程：它观察环境，决定下一步行动，调用工具，读取结果，更新状态，并在完成、失败、等待人工输入或达到控制边界时停止。OpenAI 对 Agent 的描述也包含规划、工具调用、多专家协作和足够的状态管理；其 Agents SDK 将运行循环、工具调用、交接、守卫和可恢复状态组合成一个运行时模型。(developers.openai.com)

状态机的作用，是把这个运行过程从“散落在 while 循环、异常处理和回调中的隐式逻辑”，提升为可检查、可持久化、可恢复的显式模型。

这里的“状态机”不是要求 Agent 的每一步都由开发者预先写死。相反，状态机负责规定运行边界和安全条件，而模型可以在某个节点内部决定具体的工具选择、参数和子步骤。这正对应了工作流与 Agent 的区别：工作流通过预定义代码路径编排模型和工具；Agent 则允许模型动态决定过程和工具使用。(anthropic.com)

一、先区分两种控制：状态机控制边界，模型控制局部策略

一个常见误解是：

只要使用状态机，Agent 就不再具有自主性。

实际情况是，状态机和模型控制的是不同层次。

状态机控制：

当前任务处于什么生命周期阶段；
哪些事件在当前阶段合法；
哪些条件必须满足才能继续；
哪些动作需要人工批准；
哪些错误可以重试，哪些错误必须终止；
进程崩溃后从哪里恢复；
并发执行时谁拥有推进权。

模型控制：

在“分析问题”节点中采用什么推理路径；
在“选择工具”节点中选择哪个工具；
工具参数具体填什么；
根据观察结果是否需要继续搜索、改写计划或请求澄清；
在允许的工具集合内如何组合多个动作。

因此，一个生产级 Agent 通常不是下面这种完全固定的流程：

开始 → 搜索 → 总结 → 结束

而更接近：

开始
  ↓
理解任务
  ├── 信息不足 → 等待用户
  ├── 可直接回答 → 生成结果
  └── 需要外部事实 → 执行工具
                         ├── 成功 → 观察结果
                         ├── 可重试失败 → 重试
                         ├── 高风险动作 → 等待批准
                         └── 不可恢复失败 → 失败结束

状态机给出的是允许的控制空间，不是替模型写出所有决策。

二、形式化模型：Agent 状态机到底描述什么

2.1 基本定义

一个有限状态机可以表示为：

M = (S, E, A, G, \delta, \lambda, s_0, F)

其中：

$S$ ：状态集合，例如 RUNNING、WAITING_APPROVAL、SUCCEEDED；
$E$ ：事件集合，例如 TASK_RECEIVED、TOOL_SUCCEEDED、TIMEOUT；
$A$ ：动作集合，例如调用模型、执行工具、写入数据库；
$G$ ：守卫条件集合；
$\delta$ ：状态转移函数；
$\lambda$ ：动作函数；
$s_0$ ：初始状态；
$F$ ：终止状态集合。

转移可以写成：

\delta(s, e, g) = s'

含义是：当系统当前处于状态 $s$ ，收到事件 $e$ ，且守卫条件 $g$ 为真时，进入状态 $s'$ 。

但 Agent 的状态不应只有一个枚举值。实际运行需要同时保存：

X = (s, d, v, r, o)

其中：

$s$ ：控制状态；
$d$ ：业务数据，例如用户请求、计划、工具结果；
$v$ ：状态机版本；
$r$ ：运行元数据，例如重试次数、租约、时间戳；
$o$ ：外部副作用的幂等信息，例如已经提交的订单号。

于是一次转移不只是“把状态从 A 改成 B”，而是：

T(X, e) \rightarrow (X', a)

即根据事件生成新的状态数据 $X'$ ，并决定是否执行动作 $a$ 。

2.2 状态不是事件，事件也不是动作

三者必须区分：

状态是“现在处于什么阶段”；
事件是“发生了什么事实”；
动作是“系统接下来要做什么”。

例如：

状态：WAITING_TOOL
事件：TOOL_SUCCEEDED
动作：保存工具结果，并进入 OBSERVING

不能把 call_search_tool 直接当成状态。因为调用工具是一个持续时间不确定、可能失败、可能重复执行的动作；状态应该描述动作生命周期，例如：

PLANNED
→ DISPATCHED
→ RUNNING
→ SUCCEEDED

如果把动作和状态混为一谈，进程崩溃时就无法判断：

工具是否已经发出请求；
请求是否已经在外部系统生效；
结果是否已经写入状态；
恢复时应该重试还是查询原请求。

三、节点：把长流程拆成可观测、可暂停的执行单元

3.1 节点的定义

节点是状态机中承担一个局部职责的执行单元。一个节点至少应明确：

输入状态；
允许读取的数据；
可调用的工具；
输出事件；
可产生的副作用；
超时和取消行为；
是否允许暂停；
成功和失败的判定方式。

例如，一个研究型 Agent 可以定义如下节点：

节点	职责	主要输出
`INTAKE`	接收任务并校验输入	`TASK_ACCEPTED`、`NEED_CLARIFICATION`
`PLAN`	形成或更新任务计划	`PLAN_READY`、`NEED_CLARIFICATION`
`ACT`	调用搜索、数据库或业务工具	`TOOL_SUCCEEDED`、`TOOL_FAILED`
`OBSERVE`	解释工具结果并判断进度	`CONTINUE`、`TASK_READY`、`BLOCKED`
`REVIEW`	风险检查或等待人工批准	`APPROVED`、`REJECTED`
`FINALIZE`	生成最终结果并提交	`TASK_SUCCEEDED`、`TASK_FAILED`

节点不是越细越好。拆分节点的依据应是控制边界：

是否需要独立重试；
是否需要独立超时；
是否需要人工介入；
是否产生不同类型的副作用；
是否需要在节点之间持久化；
是否由不同权限或不同执行器处理。

如果仅仅把每一行代码都建成状态，状态机会变成难以维护的程序计数器；如果把整个任务建成一个节点，恢复、重试和审计又会失去粒度。

3.2 节点的进入、执行和离开

一个节点的生命周期可以写成：

ENTER
  ↓
加载输入快照
  ↓
执行节点动作
  ↓
产生事件
  ↓
持久化事件与状态
  ↓
根据守卫选择下一转移
  ↓
EXIT

关键点在于：节点动作产生事件，事件驱动状态转移，而不是节点执行完后直接修改状态。

不推荐：

state.status = "OBSERVING"
result = search(...)
state.search_result = result

因为在 search() 返回之前进程崩溃时，数据库里的 status 可能已经变成 OBSERVING，但结果并不存在。恢复逻辑无法判断这次搜索是否完成。

更可靠的顺序是：

记录“工具调用已派发”
→ 执行工具
→ 记录“工具调用成功”或“工具调用失败”
→ 根据事件更新状态

这使得状态成为事件的投影，而不是唯一事实来源。

四、事件：让状态变化有原因、有顺序、可重放

4.1 事件是不可变事实

事件描述已经发生的事情，不描述意图。

好的事件：

{
  "type": "TOOL_SUCCEEDED",
  "tool": "search",
  "call_id": "call-17",
  "result_ref": "blob://runs/r1/call-17.json"
}

不好的事件：

{
  "type": "SHOULD_SEARCH"
}

SHOULD_SEARCH 是决策或计划，不是事实。它可能尚未执行，不能作为恢复依据。

事件通常包含：

{
  "event_id": "evt-42",
  "run_id": "run-1",
  "sequence": 42,
  "type": "TOOL_SUCCEEDED",
  "occurred_at": "2026-09-01T10:00:00Z",
  "node": "ACT",
  "payload": {
    "call_id": "call-17",
    "tool": "search",
    "input_hash": "sha256:...",
    "output_ref": "blob://..."
  },
  "schema_version": 1
}

其中 sequence 是同一个运行实例内的逻辑顺序，不应依赖机器本地时间排序。时间戳用于观察，序号用于状态演进。

4.2 事件日志与当前状态

事件日志记录：

任务创建
→ 计划生成
→ 工具调用派发
→ 工具调用成功
→ 观察结果
→ 任务完成

当前状态是对事件日志的物化结果：

X_n = R(X_0, e_1, e_2, \dots, e_n)

其中：

$X_0$ 是初始状态；
$e_i$ 是第 $i$ 个事件；
$R$ 是确定性的状态归约函数；
$X_n$ 是应用前 $n$ 个事件后的状态。

例如：

def reduce_state(state, event):
    state = dict(state)

    if event["type"] == "TASK_RECEIVED":
        state["status"] = "PLANNING"
        state["task"] = event["payload"]["task"]

    elif event["type"] == "PLAN_READY":
        state["status"] = "ACTING"
        state["plan"] = event["payload"]["steps"]

    elif event["type"] == "TOOL_SUCCEEDED":
        state["status"] = "OBSERVING"
        state.setdefault("tool_results", []).append(event["payload"])

    elif event["type"] == "TASK_SUCCEEDED":
        state["status"] = "SUCCEEDED"
        state["answer"] = event["payload"]["answer"]

    return state

只要归约函数没有读取当前时间、随机数、外部数据库或未记录的全局变量，同一组事件就能得到同一个状态。

4.3 为什么不能只保存最后状态

只保存：

{
  "status": "OBSERVING",
  "answer": null
}

无法回答：

哪一次模型调用产生了当前计划；
工具是否已经执行；
工具调用参数是什么；
哪个守卫导致了当前转移；
为什么没有重试；
该状态是旧版本代码产生的，还是新版本代码产生的。

事件日志提供因果链，当前状态提供快速读取。生产系统通常同时保留二者：

Event Log：完整事实
Checkpoint：某个事件序号之后的状态快照
Projection：面向查询的业务视图

五、守卫：转移前必须验证的条件

5.1 守卫不是提示词里的自然语言

守卫是一个可执行的布尔条件：

g(X, e) \in \{true, false\}

例如：

def has_sufficient_evidence(state, event):
    results = state.get("tool_results", [])
    return len(results) >= 2 and all(r.get("trusted") for r in results)

守卫应该基于结构化数据，而不是解析模型的自然语言：

# 不可靠
if "可以完成" in model_output:
    transition_to("FINALIZE")

# 更可靠
if decision["decision"] == "TASK_READY" and decision["confidence"] >= 0.8:
    transition_to("FINALIZE")

模型可以提出候选决策，但最终守卫应由代码验证。

5.2 守卫的三类作用

1. 合法性守卫

判断事件在当前状态是否允许发生：

WAITING_APPROVAL + APPROVED → ACTING
WAITING_APPROVAL + TOOL_SUCCEEDED → 非法

如果一个审批尚未完成，迟到的工具成功事件不能直接推进任务。

2. 业务守卫

判断是否满足继续执行的条件：

OBSERVING + TASK_READY → FINALIZING
OBSERVING + CONTINUE → ACTING
OBSERVING + BLOCKED → FAILED

3. 安全守卫

判断动作是否超过风险边界：

金额 ≤ 100 元 → 自动执行
金额 > 100 元 → 等待人工审批
涉及删除或付款 → 必须审批
工具权限与用户权限不匹配 → 拒绝

守卫的优先级必须明确。一个动作同时满足“模型建议执行”和“业务允许执行”，但不满足“安全允许执行”时，结果必须是阻断，而不是继续。

5.3 守卫顺序与决策冲突

设有三个候选转移：

OBSERVING + TASK_READY  → FINALIZING
OBSERVING + CONTINUE    → ACTING
OBSERVING + BLOCKED     → FAILED

如果模型输出同时包含 TASK_READY 和 BLOCKED，系统不能依赖字典顺序或 if/elif 的偶然行为。应定义优先级：

BLOCKED \succ NEED\_APPROVAL \succ TASK\_READY \succ CONTINUE

例如：

PRIORITY = {
    "BLOCKED": 100,
    "NEED_APPROVAL": 90,
    "TASK_READY": 50,
    "CONTINUE": 10,
}

然后选择优先级最高、且通过代码守卫的事件。否则“模型输出字段顺序变化”就可能导致不同状态转移。

六、转移：状态机的因果骨架

6.1 一个完整转移的五个部分

一条转移至少包含：

源状态
+ 触发事件
+ 守卫条件
+ 状态更新
+ 目标状态

可以写成：

(s, e, g) \xrightarrow{a} (s', \Delta d)

含义是：

当前状态为 $s$ ；
收到事件 $e$ ；
守卫 $g$ 成立；
执行动作 $a$ ；
更新数据 $\Delta d$ ；
进入 $s'$ 。

例如：

OBSERVING
  + TASK_READY
  + evidence_count >= required_count
  + no_pending_tool_call
  → FINALIZING

如果 TASK_READY 到达时仍有一个工具调用处于 RUNNING，这条转移必须拒绝，否则最终答案可能早于最后一个工具结果。

6.2 完整状态图

stateDiagram-v2
    [*] --> INTAKE

    INTAKE --> PLANNING: TASK_ACCEPTED
    INTAKE --> WAITING_USER: NEED_CLARIFICATION
    INTAKE --> FAILED: INVALID_INPUT

    WAITING_USER --> PLANNING: USER_REPLY
    WAITING_USER --> CANCELLED: USER_CANCELLED

    PLANNING --> ACTING: PLAN_READY
    PLANNING --> WAITING_USER: NEED_CLARIFICATION
    PLANNING --> FAILED: PLAN_FAILED

    ACTING --> OBSERVING: TOOL_SUCCEEDED
    ACTING --> RETRY_WAIT: TOOL_RETRYABLE_FAILED
    ACTING --> WAITING_APPROVAL: APPROVAL_REQUIRED
    ACTING --> FAILED: TOOL_FATAL_FAILED

    RETRY_WAIT --> ACTING: RETRY_DUE
    RETRY_WAIT --> FAILED: RETRY_EXHAUSTED

    WAITING_APPROVAL --> ACTING: APPROVED
    WAITING_APPROVAL --> CANCELLED: REJECTED
    WAITING_APPROVAL --> WAITING_APPROVAL: APPROVAL_TIMEOUT

    OBSERVING --> ACTING: CONTINUE
    OBSERVING --> FINALIZING: TASK_READY
    OBSERVING --> WAITING_USER: BLOCKED_BY_USER
    OBSERVING --> FAILED: UNRECOVERABLE

    FINALIZING --> SUCCEEDED: OUTPUT_COMMITTED
    FINALIZING --> FAILED: OUTPUT_COMMIT_FAILED

    SUCCEEDED --> [*]
    FAILED --> [*]
    CANCELLED --> [*]

图中的 ACTING 不代表一个不可分割的大动作。它通常还需要记录工具调用的生命周期；状态机的粒度应足以支持超时、重试和恢复。

6.3 终止状态不是“函数返回”

终止状态意味着该运行实例不会再接受普通推进事件：

SUCCEEDED
FAILED
CANCELLED

但终止状态仍可能接受运维事件，例如：

SUCCEEDED + REPLAY_REQUESTED
FAILED + MANUAL_RETRY_REQUESTED

这类事件不应直接修改原运行实例，而应创建新的运行实例或新的尝试编号。否则原始执行历史会被覆盖，审计边界消失。

七、Agent 运行循环如何嵌入状态机

一个典型 Agent 循环可以表示为：

\text{Observe} \rightarrow \text{Decide} \rightarrow \text{Act} \rightarrow \text{Feedback}

映射到状态机：

OBSERVING
  ↓
模型读取当前状态和环境事实
  ↓
产生结构化决策
  ↓
守卫验证
  ↓
ACTING
  ↓
调用工具
  ↓
产生 TOOL_SUCCEEDED / TOOL_FAILED
  ↓
OBSERVING

Anthropic 将 Agent 概括为模型在工具循环中自主使用工具，并强调每一步都需要从环境获得事实反馈，同时在阻塞点或检查点暂停；为控制运行边界，还应设置最大迭代次数等停止条件。(anthropic.com)

一个最小的结构化决策协议可以是：

{
  "decision": "CALL_TOOL",
  "tool": "search",
  "arguments": {
    "query": "杭州 2026 年 9 月 1 日天气"
  },
  "reason": "需要实时事实"
}

状态机不应接受任意模型文本直接执行，而应经过以下步骤：

解析 JSON；
验证 decision 是否属于允许枚举；
验证工具是否属于当前节点允许集合；
验证参数是否符合 JSON Schema；
验证用户权限和资源范围；
执行幂等检查；
写入派发事件；
执行工具；
写入结果事件；
根据结果事件推进状态。

OpenAI 的 Agents SDK 和 Responses API 都支持工具型 Agent，但控制权不同：Responses API 更适合由应用自己管理模型交互、工具结果、循环和分支；Agents SDK 则提供运行循环、交接、守卫、追踪和可恢复运行状态等更高层能力。(developers.openai.com)

八、一个可执行的最小状态机

下面的 Python 示例不依赖第三方库，演示四个核心机制：

事件驱动转移；
守卫检查；
可重放归约；
工具失败后的重试。

from dataclasses import dataclass, field
from typing import Any


TERMINAL_STATES = {"SUCCEEDED", "FAILED", "CANCELLED"}


@dataclass
class Event:
    sequence: int
    type: str
    payload: dict[str, Any] = field(default_factory=dict)


def reduce_state(state: dict[str, Any], event: Event) -> dict[str, Any]:
    """只根据已有状态和事件生成新状态，不访问外部系统。"""
    new_state = dict(state)
    event_type = event.type
    payload = event.payload

    if event_type == "TASK_RECEIVED":
        new_state.update({
            "status": "PLANNING",
            "task": payload["task"],
            "attempt": 0,
        })

    elif event_type == "PLAN_READY":
        new_state.update({
            "status": "ACTING",
            "plan": payload["plan"],
        })

    elif event_type == "TOOL_STARTED":
        new_state.update({
            "status": "ACTING",
            "pending_call_id": payload["call_id"],
        })

    elif event_type == "TOOL_SUCCEEDED":
        results = list(new_state.get("results", []))
        results.append({
            "call_id": payload["call_id"],
            "value": payload["value"],
        })
        new_state.update({
            "status": "OBSERVING",
            "pending_call_id": None,
            "results": results,
        })

    elif event_type == "TOOL_RETRYABLE_FAILED":
        new_state.update({
            "status": "RETRY_WAIT",
            "pending_call_id": None,
            "last_error": payload["error"],
        })

    elif event_type == "RETRY_DUE":
        new_state.update({
            "status": "ACTING",
            "attempt": new_state.get("attempt", 0) + 1,
        })

    elif event_type == "TASK_READY":
        new_state.update({
            "status": "FINALIZING",
        })

    elif event_type == "OUTPUT_COMMITTED":
        new_state.update({
            "status": "SUCCEEDED",
            "answer": payload["answer"],
        })

    elif event_type == "FAILED":
        new_state.update({
            "status": "FAILED",
            "last_error": payload["error"],
        })

    else:
        raise ValueError(f"unknown event: {event_type}")

    return new_state


def can_apply(state: dict[str, Any], event: Event) -> bool:
    """守卫：判断事件能否在当前状态应用。"""
    status = state.get("status")

    if status in TERMINAL_STATES:
        return False

    if event.type == "TASK_RECEIVED":
        return status is None

    if event.type == "PLAN_READY":
        return status == "PLANNING"

    if event.type == "TOOL_STARTED":
        return status == "ACTING" and not state.get("pending_call_id")

    if event.type in {"TOOL_SUCCEEDED", "TOOL_RETRYABLE_FAILED"}:
        return status == "ACTING" and state.get("pending_call_id") == event.payload["call_id"]

    if event.type == "RETRY_DUE":
        return (
            status == "RETRY_WAIT"
            and state.get("attempt", 0) < 3
        )

    if event.type == "TASK_READY":
        return (
            status == "OBSERVING"
            and len(state.get("results", [])) >= 1
            and not state.get("pending_call_id")
        )

    if event.type == "OUTPUT_COMMITTED":
        return status == "FINALIZING"

    if event.type == "FAILED":
        return status not in TERMINAL_STATES

    return False


def replay(events: list[Event]) -> dict[str, Any]:
    state: dict[str, Any] = {}

    for event in events:
        if not can_apply(state, event):
            raise ValueError(
                f"illegal transition: state={state.get('status')}, "
                f"event={event.type}"
            )
        state = reduce_state(state, event)

    return state


events = [
    Event(1, "TASK_RECEIVED", {"task": "查询一个事实"}),
    Event(2, "PLAN_READY", {"plan": ["search"]}),
    Event(3, "TOOL_STARTED", {"call_id": "call-1"}),
    Event(4, "TOOL_RETRYABLE_FAILED", {
        "call_id": "call-1",
        "error": "temporary timeout",
    }),
    Event(5, "RETRY_DUE"),
    Event(6, "TOOL_STARTED", {"call_id": "call-2"}),
    Event(7, "TOOL_SUCCEEDED", {
        "call_id": "call-2",
        "value": "事实结果",
    }),
    Event(8, "TASK_READY"),
    Event(9, "OUTPUT_COMMITTED", {"answer": "最终答案"}),
]

state = replay(events)
print(state)

预期输出类似：

{
    'status': 'SUCCEEDED',
    'task': '查询一个事实',
    'attempt': 1,
    'plan': ['search'],
    'pending_call_id': None,
    'results': [
        {'call_id': 'call-2', 'value': '事实结果'}
    ],
    'last_error': 'temporary timeout',
    'answer': '最终答案'
}

这个例子中，TOOL_RETRYABLE_FAILED 并没有直接回到 ACTING，而是先进入 RETRY_WAIT。原因是重试通常涉及等待时间、退避策略、调度器和租约，不应在异常处理栈内无限递归。

同样，TOOL_SUCCEEDED 必须携带 call_id。如果只根据工具名判断，旧请求的迟到结果可能覆盖新请求的结果：

call-1 超时
→ 发起 call-2
→ call-2 成功
→ call-1 的迟到结果到达

没有调用 ID 和版本检查时，call-1 可能错误地把任务状态改回旧路径。

九、可恢复执行：恢复的不是“代码位置”，而是逻辑状态

9.1 为什么进程恢复不等于任务恢复

进程崩溃后，不能简单地从 Python 调用栈或线程栈恢复。因为：

模型调用可能已经完成，但响应尚未写入本地内存；
工具请求可能已经到达外部系统，但客户端没有收到响应；
数据库提交可能成功，但网络响应丢失；
进程可能在写事件前或写事件后崩溃；
原来的代码版本可能已经升级。

可恢复执行要保存的是：

任务身份
+ 当前状态
+ 已确认的事件
+ 未完成的动作
+ 动作幂等键
+ 状态机版本
+ 重试信息
+ 外部副作用引用

9.2 事件日志、Checkpoint 和重放

如果每次恢复都从第一条事件开始重放，运行很长时成本会增加。因此可以周期性保存 Checkpoint：

{
  "run_id": "run-123",
  "checkpoint_sequence": 800,
  "state_machine_version": "2026-09",
  "state": {
    "status": "OBSERVING",
    "results": ["..."]
  }
}

恢复步骤为：

读取最新合法 Checkpoint；
获取 checkpoint_sequence 之后的事件；
按序归约；
检查是否存在未完成动作；
重新获取外部结果或按策略重试；
继续状态机推进。

Checkpoint 不是事实来源，而是事件日志的加速索引。如果 Checkpoint 与事件不一致，应以事件日志为准，并将异常标记为数据完整性问题。

9.3 “至少一次”执行与幂等

分布式系统中，最常见的执行语义是至少一次：

执行器可能重复执行同一个动作，但不会故意丢弃已经确认的动作。

这意味着工具必须设计幂等键：

idempotency\_key = hash(run\_id, node\_id, action\_id)

例如：

run-123 / ACT / action-7

调用支付工具时，重试请求应携带同一个幂等键。支付服务看到相同键时，应返回第一次执行的结果，而不是再次扣款。

对于读操作，重复调用通常只增加成本；对于写操作，重复调用可能造成：

重复下单；
重复发邮件；
重复创建工单；
重复扣款；
重复修改配置。

因此，“自动重试工具调用”不是通用安全策略。是否可重试必须由工具契约声明：

{
  "tool": "create_ticket",
  "retry_policy": {
    "retryable": true,
    "requires_idempotency_key": true
  }
}

十、租约：避免多个执行器同时推进同一个 Agent

10.1 租约解决什么问题

当任务存储在队列或数据库中时，可能有多个 Worker 同时尝试恢复同一个 run_id：

Worker A 读取 run-123
Worker B 也读取 run-123
A 调用工具
B 也调用工具
A 写入成功事件
B 写入另一个成功事件

如果工具有副作用，就会产生重复操作。

租约是带过期时间的所有权记录：

run-123
owner = worker-a
lease_until = 2026-09-01T10:05:00Z
fencing_token = 81

Worker 必须在推进状态或提交事件时证明自己仍拥有有效租约。

10.2 仅靠锁不够

数据库锁通常只保护一次事务，不能覆盖一个持续几十秒的模型调用或工具调用。若持锁等待外部服务，会造成连接和锁资源长期占用。

更常见的方式是：

短事务获取租约；
事务外执行模型或工具；
周期性续租；
提交事件时校验 fencing_token；
租约失效后，旧 Worker 的写入必须被拒绝。

fencing_token 用于防止“旧 Worker 虽然已经失去租约，但仍在网络恢复后写入数据”。每次获取新租约时递增令牌：

Worker A：token=81
Worker A 停顿，租约过期
Worker B：token=82
Worker A 恢复并尝试提交
→ 存储层拒绝 token=81

因此，租约不仅是“谁正在处理”的标记，还必须成为写入条件的一部分。

十一、故障路径：超时、迟到、重复和部分成功

11.1 模型调用超时

模型调用超时后，不能直接假设模型没有生成结果。可能的情况有：

客户端超时
├── 请求未到达模型服务
├── 模型服务已开始处理但未完成
├── 模型已完成，响应在网络中丢失
└── 响应已到达，但本地进程在写入前崩溃

因此，模型调用也应有唯一的 request_id。恢复时优先查询供应商或中间层是否存在同一请求的结果；无法查询时，才根据业务风险决定重试。

对于无副作用的模型生成，重试通常是可接受的；但如果模型调用会触发工具执行或外部事务，应该把“模型决策”和“工具执行”分成两个可审计阶段。

11.2 工具超时

工具超时至少要区分：

连接前失败：通常可以重试；
服务明确拒绝：根据错误码决定；
请求已发送但结果未知：必须先查询幂等键；
服务返回业务失败：通常不能通过网络重试解决；
结果已成功但确认丢失：查询操作状态，而不是重新创建。

错误分类比异常类名更重要。TimeoutError 只能说明客户端未在期限内得到结果，不能说明外部动作没有发生。

11.3 事件重复

事件写入应具备唯一约束：

PRIMARY KEY (run_id, sequence)
UNIQUE (run_id, event_id)

消费事件时也要幂等。若同一 event_id 重复到达，应返回已有结果，而不是再次应用副作用。

11.4 事件乱序

事件系统可能提供至少一次投递，但不保证跨分区顺序。状态机应要求：

sequence == last_applied_sequence + 1

如果收到序号 12，而当前只应用到 10：

暂存 12；
等待 11；
超时后报警；
不得直接应用 12。

否则状态归约可能跳过关键事实。

十二、并发状态机：顺序执行、并行执行与共享状态

12.1 顺序执行适合有依赖的任务

如果步骤 B 依赖步骤 A 的结果：

读取订单
→ 判断是否符合退款条件
→ 请求退款

它们不能简单并行。退款动作必须建立在读取结果和策略判断之上。

Anthropic 将并行模式描述为适合相互独立的子任务；当任务需要特定顺序、共享状态协调或明确的冲突解决策略时，不应使用并行模式。(resources.anthropic.com)

12.2 并行执行需要 Join 状态

假设 Agent 同时从三个来源获取证据：

OBSERVING
  ↓ fan-out
SEARCH_A ─┐
SEARCH_B ─┼→ JOIN_EVIDENCE → FINALIZING
SEARCH_C ─┘

状态不能只保存一个 pending 标记，而应保存每个分支：

{
  "status": "WAITING_JOIN",
  "branches": {
    "a": {"status": "SUCCEEDED"},
    "b": {"status": "FAILED", "retryable": true},
    "c": {"status": "SUCCEEDED"}
  }
}

Join 守卫必须明确：

所有分支成功
或
成功分支数量达到最小阈值
或
剩余失败分支不可恢复但已有足够证据

如果没有明确的 Join 条件，并发只会把“什么时候算完成”变成隐式竞态。

12.3 共享状态更新必须可合并

两个并行节点同时写入同一个字段：

worker-a: state.summary = "结论 A"
worker-b: state.summary = "结论 B"

最后写入者覆盖前者，结果取决于调度时序。更安全的做法是让分支写入独立事实：

{
  "branch": "a",
  "finding": "结论 A"
}

由 Join 节点统一聚合。聚合规则可以是：

集合合并；
按来源优先级选择；
版本向量合并；
冲突转人工审核；
由专门的评估节点重新判断。

十三、人工审批：等待不是失败，暂停也不是结束

高风险 Agent 需要把人工审核建模成一等状态：

WAITING_APPROVAL

而不是在代码中同步阻塞：

input("请审批：")

同步阻塞的问题是：

Worker 资源长期占用；
进程重启后审批上下文丢失；
无法设置审批超时；
无法记录谁批准了什么版本的动作；
多次点击可能重复提交。

一个审批请求应保存：

{
  "approval_id": "approval-9",
  "run_id": "run-123",
  "action_hash": "sha256:...",
  "requested_by": "agent",
  "requested_at": "2026-09-01T10:00:00Z",
  "expires_at": "2026-09-01T12:00:00Z",
  "risk": "HIGH"
}

审批事件必须绑定 action_hash。如果 Agent 在等待期间修改了工具参数，原来的批准不能自动适用于新动作：

提交退款 100 元
→ 请求审批
→ Agent 修改为退款 10,000 元
→ 旧批准无效

正确做法是重新生成审批请求。

OpenAI Agents SDK 的文档将守卫、人审和可恢复审批流作为 Agent 运行能力的一部分；当运行暂停等待批准时，恢复应基于明确的运行状态，而不是重新从用户输入猜测上下文。(developers.openai.com)

十四、确定性：哪些部分必须稳定，哪些部分可以随机

Agent 不可能整体确定性，因为模型输出可能变化，外部环境也可能变化。但可恢复执行要求关键控制点尽量确定。

14.1 必须确定的部分

以下逻辑应尽量纯函数化：

事件归约；
守卫判断；
重试次数计算；
状态合法性校验；
终止条件；
Checkpoint 版本检查；
幂等键生成；
事件序号分配。

例如：

def retry_delay(attempt: int) -> int:
    # attempt=0,1,2 时分别为 1、2、4 秒
    return 2 ** attempt

不要在归约函数中直接调用：

random.random()
datetime.now()
database.query(...)

否则同一事件日志重放两次可能得到不同状态。

14.2 可以非确定的部分

模型的自然语言生成、候选工具选择和开放式规划可以具有不确定性，但其输出应被转化为结构化事件：

模型自由生成
→ 解析与验证
→ 记录决策摘要、输入哈希和模型版本
→ 代码守卫裁决
→ 执行工具

“记录模型版本”并不意味着未来一定能复现字面输出。它的作用是解释当时使用了什么运行环境，并帮助定位行为变化。

14.3 环境事实必须保存引用

如果工具返回一个很大的结果，不一定要将完整结果复制进每一条事件，但至少要保存：

结果引用
+ 内容哈希
+ 工具版本
+ 请求参数哈希
+ 获取时间

恢复时可以重新读取原结果并校验哈希。若外部结果会变化，例如实时库存或天气，重放时应使用历史结果，而不是重新查询当前值；否则恢复过程实际上变成了另一条执行路径。

十五、版本演进：状态机代码升级后如何恢复旧任务

状态机版本是运行数据的一部分：

{
  "state_machine_version": "2026-09",
  "checkpoint_sequence": 120
}

当状态结构变化时，必须定义迁移：

v1:
  status = "WAITING_TOOL"

v2:
  status = "ACTING"
  pending_call = {...}

迁移函数应显式存在：

def migrate_v1_to_v2(state):
    if state.get("status") == "WAITING_TOOL":
        state["status"] = "ACTING"
    return state

不能只升级代码然后假设旧 Checkpoint 会自然兼容。常见风险包括：

旧状态没有新字段；
同一个事件在新版本中含义改变；
新守卫拒绝旧流程已经产生的事件；
工具参数 schema 发生不兼容变化；
旧任务恢复后执行了与原意不同的副作用。

对于长时间运行的任务，通常需要同时支持多个状态机版本，直到旧任务完成或被迁移。

十六、诊断：先看哪一层出了问题

当 Agent “卡住”或“做错了”，应按以下顺序诊断。

16.1 先看当前状态

当前状态：WAITING_APPROVAL

这说明它不是模型失控，而是在等待一个审批事件。若状态是 RETRY_WAIT，则应查看调度器；若状态是 ACTING，则要检查未完成动作。

16.2 再看最后一条已应用事件

sequence=42
type=TOOL_STARTED
call_id=call-17

如果之后没有 TOOL_SUCCEEDED、TOOL_FAILED 或 TOOL_TIMED_OUT，说明执行器或事件回写链路存在问题。

16.3 检查事件间的合法性

OBSERVING + TOOL_SUCCEEDED

如果当前状态是 OBSERVING，却收到一个工具成功事件，可能是：

重复投递；
迟到事件；
旧 Worker 越权写入；
状态已被错误推进；
事件序号或运行 ID 绑定错误。

16.4 检查模型决策和代码守卫

如果模型输出了 CALL_TOOL，但状态没有进入 ACTING，需要区分：

JSON 解析失败；
工具不在白名单；
参数 schema 不合法；
用户权限不足；
守卫不满足；
事件持久化失败。

不能只看模型文本判断“模型没有执行”。

16.5 检查副作用状态

如果 Agent 显示 TOOL_FAILED，但外部订单已经创建，说明“业务副作用”和“Agent 事件”之间发生了不一致。此时不能简单重试创建操作，而应通过幂等键或业务查询接口确认外部状态，再决定补写成功事件、执行补偿，或进入人工处理。

十七、常见错误及其反例

错误一：用布尔变量替代状态机

done = False
approved = False
has_result = False

当变量增加到一定数量后，合法组合会急剧增加：

done=True, approved=False, has_result=True

这组组合是否允许？如果没有统一转移规则，系统会进入无法解释的状态。

状态机通过显式状态集合限制组合：

WAITING_APPROVAL
SUCCEEDED

而不是允许任意布尔变量自由组合。

错误二：在工具调用前就写“成功状态”

status = SUCCEEDED
call external API

进程在 API 调用期间崩溃后，任务已经显示成功，但副作用可能未发生。成功状态必须由可验证的成功事件驱动。

错误三：把模型输出当作事实

模型说：“退款已经完成”
→ 状态改为 SUCCEEDED

模型输出只是声明，不是外部事实。必须读取退款 API 的确认结果，或者通过查询接口验证。

错误四：无限重试

while True:
    try:
        call_tool()
        break
    except Exception:
        continue

这会掩盖永久性错误、制造重复副作用，并使任务无法终止。重试必须受以下条件约束：

retry\_allowed = is\_retryable(error) \land attempt < max\_attempts \land elapsed < deadline

错误五：只把人工审批当成异常分支

try:
    execute()
except NeedApproval:
    raise

异常栈不适合表示跨进程、跨小时甚至跨天的等待。人工审批应持久化为状态和审批请求，之后由新事件恢复。

十八、如何划分状态机边界

不是所有 Agent 都需要完整事件溯源状态机。

一次性问答通常只需要：

输入
→ 模型调用
→ 输出

如果任务具有以下特征，状态机的收益会明显增加：

包含多个工具调用；
工具可能超时或重复；
任务需要跨进程或跨时间继续；
存在人工审批；
需要审计每次决策；
具有不可逆副作用；
有多个并行分支；
需要在模型、工具和业务系统之间恢复一致性。

Anthropic 建议从最简单的提示词或单次模型调用开始，仅在简单方案不能满足任务时增加 Agent 化复杂度，因为更高的自主性通常同时带来更高的延迟、成本和错误累积风险。(anthropic.com)

状态机也不应替代领域模型。它负责：

运行生命周期、事件顺序、权限边界、恢复语义

领域模型负责：

订单是否可退款、用户是否满足资格、库存是否充足

把所有业务规则都塞进状态机，会得到一个巨大且无法演进的转移表；把所有运行控制都交给领域服务，又会失去恢复和审计能力。

十九、面向生产的最小契约

一个可恢复的 Agent 运行实例，至少应满足以下契约：

1. 每个运行实例有唯一 run_id。
2. 每个动作有唯一 action_id 或 idempotency_key。
3. 每个事件不可变，并有单调 sequence。
4. 状态只能由合法事件转移产生。
5. 归约函数不依赖未记录的外部变量。
6. 外部副作用必须可查询、可幂等或可补偿。
7. 长时间等待必须进入持久化等待状态。
8. Worker 推进任务必须持有有效租约。
9. 旧 Worker 必须通过 fencing token 被拒绝。
10. Checkpoint 必须绑定事件序号和状态机版本。
11. 所有终止路径都必须显式记录原因。
12. 模型决策必须经过结构化解析和守卫校验。

其中最重要的不是状态名称，而是因果链完整：

事件发生
→ 事件持久化
→ 状态归约
→ 守卫判断
→ 动作派发
→ 外部结果确认
→ 新事件持久化

如果这条链在任意位置被隐式跳过，恢复时就会出现“状态看起来正确，但事实并不确定”的问题。

Agent 状态机的核心不是画出更多方框，而是建立一套可证明的运行语义：节点定义执行边界，事件记录事实，守卫限制合法条件，转移表达因果关系，事件日志和 Checkpoint 保存历史，租约和幂等键处理并发与重复，恢复逻辑则从最后一个可确认事实继续推进。这样，Agent 才不只是一个偶尔成功的工具调用循环，而是一个能够暂停、失败、重试、审计和恢复的工程系统。

系列导航与关联阅读

官方资料

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