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

审批 Agent：确定流程、草稿生成、确认点、幂等和审计

审批 Agent 不是“让模型替用户点一下提交”，而是让模型参与一个受约束、可暂停、可恢复、可追责的写操作流程。

它通常要完成四类工作：

从自然语言中识别用户意图和业务对象；
根据确定的业务规则生成审批草稿；
在不可逆或高风险动作前暂停，等待授权人确认；
在网络重试、并发请求、进程崩溃和人工复核后，仍然保证结果可解释、不可重复执行，并且能够还原全过程。

这里的“审批”是业务语义，不是模型输出中的一句“已批准”。真正的批准必须落在一个由服务端校验、持久化和执行的状态机中。

Anthropic 将工作流定义为由预先确定的代码路径编排模型和工具，将 Agent 定义为由模型动态决定过程和工具使用的系统；对于边界清晰、规则固定的任务，工作流通常更可预测，而 Agent 更适合需要灵活判断的任务。审批系统应当采用两者的组合：让 Agent 负责理解和起草，让确定性代码负责授权、状态转移和写入外部系统。(anthropic.com)

一、先界定问题：审批 Agent 到底批准什么

1. 审批对象不是一段文本，而是一条可执行命令

审批 Agent 的核心对象应当是一个操作意图，而不是模型生成的自然语言回复。

例如，用户说：

把订单 ORD-20260901-001 的金额调整为 899 元，并通知客户。

模型可以将其解析为：

{
  "operation": "adjust_order_amount",
  "target": {
    "order_id": "ORD-20260901-001"
  },
  "parameters": {
    "new_amount": 89900,
    "currency": "CNY",
    "notify_customer": true
  },
  "reason": "用户要求调整订单金额"
}

其中：

operation 是操作类型；
target 是业务对象；
parameters 是待写入参数；
reason 是申请理由；
金额使用分为单位的整数，避免浮点误差。

这条结构化对象仍然不是执行命令。它还需要经过：

身份认证；
权限判断；
业务规则校验；
当前数据读取；
审批策略匹配；
草稿冻结；
用户确认；
幂等执行。

因此可以把审批 Agent 抽象为：

\text{Agent}: \text{自然语言输入} \rightarrow \text{结构化意图} \rightarrow \text{审批草稿}

而不是：

\text{Agent}: \text{自然语言输入} \rightarrow \text{直接写操作}

第二种设计把理解、授权和执行混在了一起，无法判断一次错误写入究竟是模型理解错误、权限错误、参数错误，还是重试造成的重复执行。

2. 审批和确认不是同一个概念

本文中需要区分四个概念：

概念	含义
申请	请求执行某个业务操作
草稿	经过解析和校验、但尚未获得执行授权的操作快照
确认	授权人明确同意某个特定草稿
执行	服务端根据已确认草稿向目标系统发起写操作

“确认”不能只绑定到会话，也不能只绑定到操作类型。

以下确认都不充分：

用户确认了

用户确认调整订单

用户昨天确认过类似操作

可靠的确认必须绑定到：

哪个审批单；
哪个草稿版本；
哪些参数；
哪个目标资源；
哪个确认人；
哪个时间窗口；
哪种权限；
是否已经使用过。

可以形式化为：

\text{Approval} = (\text{request\_id}, \text{draft\_version}, \text{actor}, \text{scope}, \text{expires\_at})

只要其中任一关键字段发生变化，原确认就不能继续授权新的执行。

二、确定流程：让 Agent 只负责不确定的部分

1. “确定流程”是什么意思

确定流程是指：状态、允许的状态转移、转移条件和副作用都由程序显式定义，而不是由模型临时决定。

一个最小审批流程可以表示为：

RECEIVED
  -> PARSED
  -> VALIDATED
  -> DRAFTED
  -> WAITING_CONFIRMATION
  -> CONFIRMED
  -> EXECUTING
  -> SUCCEEDED

异常路径包括：

PARSED -> NEEDS_CLARIFICATION
VALIDATED -> REJECTED
WAITING_CONFIRMATION -> EXPIRED
WAITING_CONFIRMATION -> CANCELLED
EXECUTING -> UNKNOWN
EXECUTING -> FAILED

这里的关键是：

Agent 可以决定“需要询问什么”；
Agent 可以生成“申请理由”和“参数说明”；
Agent 可以建议“可能需要哪类审批人”；
Agent 不能自行制造批准结果；
Agent 不能绕过 WAITING_CONFIRMATION；
Agent 不能通过修改上下文伪造已经确认的草稿。

OpenAI Agents SDK 的官方文档也将 Agent 描述为能够规划、调用工具、协作并保留足够状态来完成多步工作的应用；SDK 可以管理 Agent 循环、工具调用、分支和暂停审批，但具体的业务权限、工具实现、状态存储和审批决策仍由应用负责。(developers.openai.com)

2. Agent 与确定性编排器的职责边界

一个生产系统通常包含以下组件：

flowchart LR
    U[用户] --> API[审批 API]
    API --> AG[解析 Agent]
    AG --> VAL[确定性校验器]
    VAL --> POL[审批策略引擎]
    POL --> DRAFT[草稿存储]
    DRAFT --> UI[确认界面]
    UI --> CONF[确认服务]
    CONF --> EXEC[执行器]
    EXEC --> EXT[外部业务系统]
    EXEC --> AUDIT[审计日志]
    API --> AUDIT
    AG --> AUDIT
    VAL --> AUDIT

各组件的责任应当清晰：

Agent

负责：

从自然语言提取字段；
识别歧义；
生成面向人的草稿说明；
将工具查询结果组织成上下文；
在缺少信息时提出问题。

不负责：

判断用户是否拥有最终权限；
直接调用高风险写工具；
解释一个已经过期的确认；
决定是否可以忽略校验失败。

确定性校验器

负责：

JSON Schema 或类型校验；
数值范围；
枚举值；
资源是否存在；
当前状态是否允许该操作；
业务约束；
权限和职责分离。

审批策略引擎

负责：

是否需要审批；
需要几级审批；
哪些角色可以批准；
申请人和批准人是否必须不同；
金额、资源类型、环境、风险级别对应的规则。

执行器

负责：

在确认后重新检查关键条件；
使用幂等键调用下游系统；
记录请求和响应；
处理超时和未知结果；
不将普通网络重试误认为安全重试。

审计服务

负责：

记录谁在何时提出了什么请求；
Agent 生成了什么草稿；
哪些规则通过或拒绝；
谁确认了哪个版本；
实际执行了什么；
下游返回了什么；
发生了哪些重试、失败和人工介入。

三、草稿生成：把模型输出变成可审查的操作快照

1. 草稿不是“给用户看的摘要”

草稿是一个准备执行但尚未执行的不可变快照。

假设用户请求：

给客户 C10086 发放 50 元优惠券，有效期 30 天，原因是物流延迟。

系统不应只生成：

将向客户发放 50 元优惠券，有效期 30 天。

而应生成结构化草稿：

{
  "request_id": "apr_01J...",
  "draft_version": 3,
  "operation": "issue_coupon",
  "target": {
    "customer_id": "C10086"
  },
  "parameters": {
    "amount": 5000,
    "currency": "CNY",
    "validity_days": 30,
    "reason_code": "LOGISTICS_DELAY"
  },
  "risk": {
    "level": "medium",
    "estimated_cost": 5000,
    "reversible": true
  },
  "snapshot": {
    "customer_status": "active",
    "policy_version": "coupon-policy-2026-08-12",
    "source_order_id": "ORD-20260830-002"
  },
  "display": {
    "summary": "向客户 C10086 发放 50 元人民币优惠券",
    "details": [
      "有效期：30 天",
      "原因：物流延迟",
      "预计成本：50 元",
      "可撤销：是"
    ]
  }
}

这里同时保存了：

机器执行字段：operation、target、parameters；
风险字段：金额、可撤销性、风险等级；
判断依据：客户状态、策略版本、关联订单；
人类阅读字段：摘要和明细。

人类看到的文本不能反向成为唯一的执行依据。执行器应当读取结构化字段，并将展示文本作为审计和确认界面的一部分。

2. 草稿生成必须经过“解析—规范化—校验”

推荐将生成过程拆成三个阶段。

第一步：解析

模型从输入中抽取候选字段：

{
  "customer_id": "C10086",
  "amount": "50元",
  "validity": "30天",
  "reason": "物流延迟"
}

此时允许字段不完整，也允许存在歧义。

第二步：规范化

程序或受约束的转换层将值转换为标准表示：

{
  "customer_id": "C10086",
  "amount_minor": 5000,
  "currency": "CNY",
  "validity_days": 30,
  "reason_code": "LOGISTICS_DELAY"
}

模型可以建议 reason_code，但最终映射应由程序维护。例如：

REASON_CODES = {
    "物流延迟": "LOGISTICS_DELAY",
    "配送延误": "LOGISTICS_DELAY",
    "商品破损": "DAMAGED_ITEM",
}

第三步：确定性校验

def validate_coupon_draft(draft: dict) -> list[str]:
    errors = []

    if not draft["customer_id"].startswith("C"):
        errors.append("customer_id 格式无效")

    if draft["amount_minor"] <= 0:
        errors.append("优惠券金额必须大于 0")

    if draft["amount_minor"] > 10000:
        errors.append("单次优惠券金额不能超过 100 元")

    if not 1 <= draft["validity_days"] <= 90:
        errors.append("有效期必须在 1 到 90 天之间")

    if draft["reason_code"] not in {
        "LOGISTICS_DELAY",
        "DAMAGED_ITEM",
    }:
        errors.append("原因代码不在允许范围内")

    return errors

即使模型输出严格符合 JSON，也不能因此认为业务上合法。结构化输出解决的是“格式约束”，不是“事实正确”和“授权正确”。

3. 草稿必须有版本

草稿版本用于处理修改和并发确认。

request_id = apr_1001
draft_version = 1

用户提出修改金额后：

request_id = apr_1001
draft_version = 2

版本 1 的确认不能授权版本 2，因为：

\text{confirmed\_version} \neq \text{current\_version} \Rightarrow \text{拒绝执行}

这条规则可以阻止一种危险情况：

用户看到“发放 50 元”；
用户点击确认；
Agent 或其他操作将草稿改成“发放 500 元”；
系统却沿用原来的确认。

四、确认点：在什么地方必须暂停

1. 确认点的定义

确认点是状态机中一个允许暂停并等待外部授权的稳定状态。

它不是模型提示词中的：

请用户确认后再执行

而是数据库中的：

status = WAITING_CONFIRMATION

系统只有在收到独立的确认请求后，才能把状态转为：

status = CONFIRMED

然后执行器再消费 CONFIRMED 状态。

2. 哪些操作需要确认

不能只按工具名称判断是否需要确认。风险通常由多个因素共同决定：

R = f(O, A, T, C, P)

其中：

$O$ ：操作类型，例如退款、删除、发信；
$A$ ：影响范围，例如单个对象或批量对象；
$T$ ：金额或资源规模；
$C$ ：可逆性；
$P$ ：权限和策略上下文。

例如：

操作	低风险情况	高风险情况
修改订单	修改备注	修改金额、收货地址
发消息	生成草稿	发送给外部客户
删除数据	删除临时文件	删除生产记录
发放优惠	金额小且可撤销	金额高或批量发放
部署	本地测试环境	生产环境

一个可执行的策略可以写成：

def requires_confirmation(draft: dict) -> bool:
    if draft["operation"] in {"delete_production_data", "send_external_message"}:
        return True

    if draft.get("risk", {}).get("estimated_cost", 0) >= 10000:
        return True

    if draft.get("target", {}).get("environment") == "production":
        return True

    return False

这只是示例策略，不是通用规范。真正的阈值、角色和审批级别应当来自业务制度，而不是模型自行推断。

3. 确认内容必须可读且可验证

确认界面至少应显示：

操作类型；
目标对象；
每个关键参数；
预计影响；
关联依据；
风险和可逆性；
草稿版本；
过期时间；
批准人身份；
确认后将发生的具体动作。

对于批量操作，还应显示范围和数量：

将向 1,284 个客户发送通知邮件。

筛选条件：
- 客户标签：trial-expired
- 注册日期：2026-08-01 至 2026-08-31
- 排除已退订客户：是

预计发送量：1,284
预计开始时间：立即
不可撤回：是

“确认全部”不能隐藏实际数量、筛选条件和不可逆性。

五、确认令牌：把“同意”绑定到具体草稿

1. 为什么需要确认令牌

如果确认接口只接收：

POST /approvals/apr_1001/confirm

服务端还需要回答：

确认的是哪个草稿版本？
确认人是谁？
确认时看到的参数是什么？
这个确认是否已经使用过？
是否过期？
是否被撤销？
当前权限是否仍然有效？

因此，确认请求应携带一个不可伪造、范围受限、短期有效的令牌。

令牌可以是随机不可预测值，也可以是签名令牌。常见设计是：

{
  "approval_id": "apv_2001",
  "request_id": "apr_1001",
  "draft_version": 3,
  "actor_id": "u_88",
  "scope": "execute:issue_coupon",
  "issued_at": "2026-09-01T09:00:00Z",
  "expires_at": "2026-09-01T09:15:00Z",
  "nonce": "random-value"
}

服务端不应只依赖客户端提交的 JSON。令牌中的声明必须经过签名验证，且服务端还要查询数据库确认其状态。

2. 确认令牌的验证条件

确认请求可以抽象为：

\text{AcceptConfirm} = S \land V \land A \land P \land E \land N

其中：

$S$ ：签名或随机令牌有效；
$V$ ：令牌绑定的草稿版本仍是当前版本；
$A$ ：批准人身份有效；
$P$ ：批准人拥有对应权限；
$E$ ：当前时间未超过过期时间；
$N$ ：令牌尚未使用、撤销或消费。

只要任一条件不成立，就不能转为 CONFIRMED。

3. 防止重放

防重放不等于“令牌足够随机”。

攻击者或客户端可能重复发送同一个确认请求：

第一次：扣款 100 元，成功
第二次：同一个确认请求，再扣款 100 元

因此至少需要：

每个确认令牌只能消费一次；
执行操作还必须有独立的幂等键；
确认和执行状态必须持久化；
重复请求应返回原操作结果，而不是再次执行。

确认令牌和执行幂等键的职责不同：

机制	防止的问题
确认令牌	未授权、错版本、过期确认、确认重放
幂等键	网络重试、消息重复、执行器重复消费
审计日志	事后追责和过程还原

六、幂等：重试不能造成第二次副作用

1. 幂等的定义

对同一业务操作重复提交多次，最终效果与提交一次相同，称为幂等。

形式上，若操作为 $F$ ，同一幂等键为 $k$ ，则应满足：

F(F(x, k), k) = F(x, k)

在实际系统中，第二次调用不一定返回完全相同的网络响应，但不能再次产生业务副作用。它可以返回第一次执行的结果：

{
  "status": "already_completed",
  "operation_id": "op_9001",
  "result": {
    "coupon_id": "cp_7788"
  }
}

2. 幂等键不能只用请求 ID

如果一次审批单允许用户修改草稿并重新确认，那么：

request_id = apr_1001

不能直接作为所有版本的幂等键。

更合理的键是：

K = \text{Hash}( \text{request\_id}, \text{draft\_version}, \text{operation}, \text{target}, \text{canonical\_parameters} )

例如：

execute:apr_1001:v3:issue_coupon:C10086:sha256(...)

其中 canonical_parameters 必须使用稳定的序列化方式：

JSON 对象键排序；
金额使用整数；
时间统一时区和格式；
数组明确是否有序；
忽略不影响执行的展示字段。

否则下面两个 JSON 可能产生不同字符串，却表达相同操作：

{"amount":5000,"currency":"CNY"}

{"currency":"CNY","amount":5000}

3. 数据库层必须有唯一约束

仅在应用代码中先查询再插入存在竞态：

请求 A：查询，不存在
请求 B：查询，不存在
请求 A：执行
请求 B：执行

应当让数据库建立唯一约束：

CREATE TABLE approval_execution (
    execution_id       TEXT PRIMARY KEY,
    request_id         TEXT NOT NULL,
    draft_version      INTEGER NOT NULL,
    idempotency_key    TEXT NOT NULL,
    status             TEXT NOT NULL,
    result_json        JSONB,
    error_code         TEXT,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    started_at         TIMESTAMPTZ,
    finished_at        TIMESTAMPTZ,
    UNIQUE (idempotency_key)
);

执行器首先尝试插入：

INSERT INTO approval_execution (
    execution_id,
    request_id,
    draft_version,
    idempotency_key,
    status
)
VALUES (
    :execution_id,
    :request_id,
    :draft_version,
    :idempotency_key,
    'PENDING'
)
ON CONFLICT (idempotency_key) DO NOTHING;

结果解释如下：

插入成功：当前请求获得执行权；
插入冲突且已有 SUCCEEDED：直接返回历史结果；
插入冲突且已有 PENDING 或 RUNNING：说明已有执行者；
插入冲突且已有 FAILED：根据错误类型决定是否允许恢复。

4. “查询—执行—记录”不是天然原子操作

一个常见错误是：

if not already_executed(key):
    call_external_system()
    mark_executed(key)

如果 call_external_system() 成功后进程崩溃，mark_executed() 没有执行，恢复任务会再次调用外部系统。

正确处理取决于下游能力：

下游支持幂等键

将同一个幂等键传给下游：

POST /refunds
Idempotency-Key: execute:apr_1001:v3:refund:ORD-1:hash

这是最可靠的方式，因为重试由下游识别。

下游支持查询

执行超时后，不立即重试写操作，而是先查询：

1. 查询下游是否存在该业务操作；
2. 若存在，记录为成功；
3. 若不存在，再用相同幂等键重试；
4. 若无法判断，进入 UNKNOWN，等待人工或补偿流程。

下游既不支持幂等，也不能查询

此时无法严格保证“最多一次”与“最终成功”同时成立。

系统只能在两个不完美选项间取舍：

不重试：避免重复副作用，但可能丢失业务操作；
重试：提高成功率，但存在重复执行风险。

不能用数据库事务伪造外部系统的原子性。跨系统场景必须明确承认这一边界，并通过人工核对、业务对账或补偿接口降低风险。

七、审批状态机：节点、事件、守卫和可恢复执行

1. 状态不是日志，而是当前事实

审批单的状态表示当前允许发生什么，不应把所有信息都塞进一个 status 字段。

推荐区分：

status：审批流程状态；
execution_status：外部执行状态；
draft_version：当前草稿版本；
approval_status：确认令牌状态；
last_error：最近一次可诊断错误；
revision：并发控制版本。

例如：

status = CONFIRMED
execution_status = UNKNOWN

这表示“审批已经确认，但外部执行结果未知”，而不是简单的失败。

2. 状态、事件和守卫

一个状态转移可表示为：

(s, e, g) \rightarrow (s', a)

其中：

$s$ ：当前状态；
$e$ ：输入事件；
$g$ ：守卫条件；
$s'$ ：目标状态；
$a$ ：转移动作。

例如：

WAITING_CONFIRMATION
  + CONFIRM
  + token_valid && version_match && permission_ok
  -> CONFIRMED

另一个例子：

EXECUTING
  + TIMEOUT
  + downstream_result_unknown
  -> UNKNOWN

UNKNOWN 是必要状态。把超时直接标记成 FAILED 会导致恢复任务重新执行；把超时直接标记成 SUCCEEDED 又可能掩盖实际失败。

3. 一个完整的状态图

stateDiagram-v2
    [*] --> RECEIVED
    RECEIVED --> PARSED: PARSE_OK
    RECEIVED --> NEEDS_CLARIFICATION: MISSING_REQUIRED_FIELD

    PARSED --> VALIDATED: VALIDATE
    VALIDATED --> REJECTED: POLICY_DENY
    VALIDATED --> DRAFTED: POLICY_PASS

    DRAFTED --> WAITING_CONFIRMATION: RISK_REQUIRES_CONFIRM
    DRAFTED --> CONFIRMED: LOW_RISK_AUTO_APPROVAL

    WAITING_CONFIRMATION --> CONFIRMED: CONFIRM_VALID
    WAITING_CONFIRMATION --> EXPIRED: TOKEN_EXPIRED
    WAITING_CONFIRMATION --> CANCELLED: CANCEL

    CONFIRMED --> EXECUTING: CLAIM_EXECUTION
    EXECUTING --> SUCCEEDED: DOWNSTREAM_SUCCESS
    EXECUTING --> FAILED: DOWNSTREAM_DEFINITE_FAILURE
    EXECUTING --> UNKNOWN: TIMEOUT_OR_LOST_RESPONSE

    UNKNOWN --> EXECUTING: RECONCILE_RETRY
    UNKNOWN --> SUCCEEDED: RECONCILE_FOUND_SUCCESS
    UNKNOWN --> FAILED: RECONCILE_FOUND_FAILURE

    NEEDS_CLARIFICATION --> PARSED: USER_SUPPLIES_INFO
    EXPIRED --> DRAFTED: RECREATE_DRAFT

关键路径不是“确认后马上调用 API”，而是：

确认
 -> 原子领取执行权
 -> 执行前重新校验
 -> 调用下游
 -> 记录确定结果或未知结果
 -> 可恢复处理

4. 可恢复执行

可恢复执行意味着进程在任意时刻崩溃后，系统可以根据持久化状态继续，而不是依赖内存中的 Agent 对话。

至少应持久化：

{
  "request_id": "apr_1001",
  "state": "EXECUTING",
  "draft_version": 3,
  "idempotency_key": "execute:...",
  "execution_id": "op_9001",
  "attempt": 2,
  "last_event": "DOWNSTREAM_TIMEOUT",
  "retry_after": "2026-09-01T09:30:00Z"
}

恢复过程：

扫描 EXECUTING 超过租约时间的记录；
根据幂等键查询下游；
查到成功则写入 SUCCEEDED；
查到明确失败则写入 FAILED；
仍无法确认则保留 UNKNOWN，避免盲目重试；
将需要人工处理的记录放入异常队列。

八、并发：确认、修改和执行可能同时发生

1. 草稿修改与确认的竞态

设当前草稿版本为 3：

请求 A：读取版本 3，准备确认
请求 B：修改参数，生成版本 4
请求 A：提交确认

如果确认接口不检查版本，A 可能确认已经不存在的版本 3，或者错误地确认当前版本 4。

确认时必须使用乐观锁：

UPDATE approval_request
SET status = 'CONFIRMED',
    confirmed_by = :actor_id,
    confirmed_at = now(),
    revision = revision + 1
WHERE request_id = :request_id
  AND status = 'WAITING_CONFIRMATION'
  AND draft_version = :token_draft_version
  AND revision = :expected_revision;

若更新行数为 0，说明发生了以下至少一种情况：

状态已经改变；
草稿版本已经变化；
令牌不再对应当前修订；
其他请求已经完成确认。

此时必须返回冲突，而不是假装确认成功。

2. 两个确认请求同时到达

两个浏览器标签页可能同时点击确认。数据库必须保证只有一个请求完成状态转移：

WAITING_CONFIRMATION -> CONFIRMED

另一个请求应该得到：

{
  "code": "APPROVAL_ALREADY_CONSUMED",
  "request_id": "apr_1001",
  "execution_id": "op_9001"
}

如果第二个请求的确认人与第一个不同，还需要根据业务规则记录为：

重复确认；
无效确认；
额外审批意见；
越权操作尝试。

不能简单丢弃所有重复请求，因为审计可能需要知道它们发生过。

3. 领取执行权

确认完成后，多个 Worker 可能同时消费同一审批单。可以使用数据库行锁：

BEGIN;

SELECT execution_id
FROM approval_execution
WHERE idempotency_key = :key
FOR UPDATE;

-- 不存在时插入执行记录
-- 已存在时根据状态决定返回或退出

COMMIT;

或者使用带租约的任务队列，但租约失效后仍然需要幂等键和未知结果处理。锁只能防止同时执行，不能解决进程在外部调用后崩溃的问题。

九、审计：记录“为什么、谁、看到了什么、做了什么”

1. 审计不等于应用日志

应用日志主要服务于排障，审计记录服务于责任追溯和合规核查。

应用日志可能写：

approval confirmed

审计需要回答：

谁确认的？
以哪个身份确认的？
确认时看到的草稿是什么？
草稿由哪个模型、哪个提示版本生成？
使用了哪些外部事实？
哪些规则通过了？
实际执行参数是什么？
下游返回了什么？
是否发生过重试？
是否有人修改过草稿？

2. 推荐的审计事件结构

{
  "event_id": "evt_01J...",
  "event_type": "APPROVAL_CONFIRMED",
  "request_id": "apr_1001",
  "draft_version": 3,
  "actor": {
    "type": "human",
    "id": "u_88",
    "roles": ["support_supervisor"]
  },
  "operation": "issue_coupon",
  "target": {
    "customer_id": "C10086"
  },
  "parameters_hash": "sha256:...",
  "policy": {
    "policy_id": "coupon-approval",
    "policy_version": "2026-08-12",
    "decision": "allow"
  },
  "confirmation": {
    "token_id": "tok_3001",
    "issued_at": "2026-09-01T09:00:00Z",
    "confirmed_at": "2026-09-01T09:03:12Z"
  },
  "trace_id": "trace_...",
  "created_at": "2026-09-01T09:03:12Z"
}

3. 事件类型要覆盖完整生命周期

至少包括：

REQUEST_RECEIVED
INTENT_PARSED
CLARIFICATION_REQUESTED
DRAFT_CREATED
DRAFT_UPDATED
VALIDATION_PASSED
VALIDATION_FAILED
POLICY_EVALUATED
CONFIRMATION_ISSUED
CONFIRMATION_REJECTED
CONFIRMATION_EXPIRED
APPROVAL_CONFIRMED
EXECUTION_CLAIMED
DOWNSTREAM_REQUESTED
DOWNSTREAM_SUCCEEDED
DOWNSTREAM_FAILED
DOWNSTREAM_UNKNOWN
EXECUTION_RECONCILED
REQUEST_CANCELLED
MANUAL_OVERRIDE

其中 DRAFT_UPDATED 不能只记录“已更新”，而应记录版本变化和差异：

{
  "event_type": "DRAFT_UPDATED",
  "old_version": 2,
  "new_version": 3,
  "changed_fields": [
    {
      "path": "$.parameters.amount_minor",
      "old": 3000,
      "new": 5000
    }
  ],
  "changed_by": "u_88"
}

4. 敏感数据和可验证性

审计不能无条件记录全部原文。身份证号、银行卡号、客户联系方式等字段应根据合规要求脱敏或加密。

但如果只保存参数哈希，又无法在事后解释当时执行了什么。因此常见做法是同时保存：

脱敏后的可读快照；
完整参数的加密副本；
参数规范化后的哈希；
加密密钥版本；
数据访问审计。

哈希用于检测内容是否被改变，不能替代原始证据。哈希本身无法告诉审计人员金额到底是 50 元还是 500 元。

5. 让审计日志尽量不可篡改

可以采用追加式事件表：

CREATE TABLE approval_event (
    event_id       TEXT PRIMARY KEY,
    request_id     TEXT NOT NULL,
    sequence_no    BIGINT NOT NULL,
    event_type     TEXT NOT NULL,
    payload_json   JSONB NOT NULL,
    prev_hash      TEXT,
    event_hash     TEXT NOT NULL,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (request_id, sequence_no)
);

计算：

H_i = \operatorname{SHA256}(H_{i-1} \parallel \text{canonical}(payload_i))

这样可以检测单条事件被修改或删除。但哈希链只能提高篡改可检测性，不能自动提供法律意义上的不可否认性，也不能替代访问控制、备份和独立存储。

十、工具设计：高风险写工具不能只靠 Prompt 约束

1. 将读工具和写工具分开

推荐把工具分成：

查询类：
- get_order
- get_customer
- check_coupon_policy
- calculate_refund

草稿类：
- create_approval_draft
- update_approval_draft
- request_clarification

执行类：
- execute_approved_coupon
- execute_approved_refund

解析 Agent 默认只能访问查询和草稿工具。执行工具还应在服务端验证：

def execute_approved_coupon(
    request_id: str,
    approval_token: str,
) -> dict:
    request = load_request(request_id)
    token = verify_token(approval_token)

    assert request.status == "CONFIRMED"
    assert request.draft_version == token.draft_version
    assert token.scope == "execute:issue_coupon"
    assert token.actor_id == request.confirmed_by

    return execute_with_idempotency(request)

工具描述能够帮助模型正确选择工具，但不能承担最终安全边界。Anthropic 特别强调，工具定义、参数说明、示例、边界和异常情况需要像人机界面一样认真设计；框架可以降低调用和编排成本，但开发者仍应理解底层请求和响应。(anthropic.com)

2. 不要把“确认”作为普通字符串参数

危险设计：

{
  "operation": "refund",
  "order_id": "ORD-1",
  "confirmed": true
}

原因很直接：

confirmed 可以被模型随意生成；
无法证明是谁确认；
无法证明确认对应哪个版本；
无法防重放；
无法判断是否过期。

应当使用服务端生成的批准记录和令牌：

{
  "request_id": "apr_1001",
  "approval_token": "opaque-server-issued-token"
}

执行器通过数据库和签名验证确定授权，而不是相信布尔值。

十一、一个最小可运行的审批服务示例

下面的示例使用 Python 标准库演示核心语义。它不是生产级存储，而是为了展示：

草稿版本；
确认令牌；
令牌一次性消费；
执行幂等；
重复确认和重复执行的结果。

from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from hashlib import sha256
from secrets import token_urlsafe
from typing import Any


def now() -> datetime:
    return datetime.now(timezone.utc)


def canonical_hash(value: Any) -> str:
    import json
    raw = json.dumps(value, sort_keys=True, separators=(",", ":"))
    return sha256(raw.encode("utf-8")).hexdigest()


@dataclass
class Draft:
    request_id: str
    version: int
    operation: str
    target: dict[str, Any]
    parameters: dict[str, Any]
    status: str = "WAITING_CONFIRMATION"


@dataclass
class ApprovalToken:
    token: str
    request_id: str
    draft_version: int
    actor_id: str
    scope: str
    expires_at: datetime
    used: bool = False


@dataclass
class Execution:
    idempotency_key: str
    status: str
    result: dict[str, Any] | None = None


class ApprovalService:
    def __init__(self):
        self.drafts: dict[str, Draft] = {}
        self.tokens: dict[str, ApprovalToken] = {}
        self.executions: dict[str, Execution] = {}

    def create_draft(self) -> Draft:
        draft = Draft(
            request_id="apr_1001",
            version=1,
            operation="issue_coupon",
            target={"customer_id": "C10086"},
            parameters={
                "amount_minor": 5000,
                "currency": "CNY",
                "validity_days": 30,
                "reason_code": "LOGISTICS_DELAY",
            },
        )
        self.drafts[draft.request_id] = draft
        return draft

    def issue_confirmation_token(self, request_id: str, actor_id: str) -> str:
        draft = self.drafts[request_id]
        token_value = token_urlsafe(32)

        token = ApprovalToken(
            token=token_value,
            request_id=request_id,
            draft_version=draft.version,
            actor_id=actor_id,
            scope=f"execute:{draft.operation}",
            expires_at=now() + timedelta(minutes=15),
        )
        self.tokens[token_value] = token
        return token_value

    def confirm(self, token_value: str, actor_id: str) -> str:
        token = self.tokens.get(token_value)
        if token is None:
            return "CONFIRMATION_TOKEN_INVALID"

        draft = self.drafts.get(token.request_id)
        if draft is None:
            return "REQUEST_NOT_FOUND"

        if token.used:
            return "CONFIRMATION_ALREADY_USED"

        if token.expires_at <= now():
            return "CONFIRMATION_EXPIRED"

        if token.actor_id != actor_id:
            return "ACTOR_MISMATCH"

        if draft.version != token.draft_version:
            return "DRAFT_VERSION_MISMATCH"

        if draft.status != "WAITING_CONFIRMATION":
            return f"INVALID_STATE:{draft.status}"

        token.used = True
        draft.status = "CONFIRMED"
        return "CONFIRMED"

    def execute(self, request_id: str) -> Execution | str:
        draft = self.drafts[request_id]

        if draft.status != "CONFIRMED":
            return f"INVALID_STATE:{draft.status}"

        key_material = {
            "request_id": request_id,
            "draft_version": draft.version,
            "operation": draft.operation,
            "target": draft.target,
            "parameters": draft.parameters,
        }
        key = f"execute:{canonical_hash(key_material)}"

        existing = self.executions.get(key)
        if existing:
            return existing

        execution = Execution(idempotency_key=key, status="SUCCEEDED")
        execution.result = {
            "coupon_id": "cp_7788",
            "customer_id": draft.target["customer_id"],
            "amount_minor": draft.parameters["amount_minor"],
        }
        self.executions[key] = execution
        draft.status = "SUCCEEDED"
        return execution

调用示例：

service = ApprovalService()

draft = service.create_draft()
print(draft.version)
# 预期输出：1

token = service.issue_confirmation_token(
    request_id=draft.request_id,
    actor_id="supervisor-1",
)

print(service.confirm(token, "supervisor-1"))
# 预期输出：CONFIRMED

print(service.confirm(token, "supervisor-1"))
# 预期输出：CONFIRMATION_ALREADY_USED

first = service.execute(draft.request_id)
second = service.execute(draft.request_id)

print(first)
print(second)
# 两次返回同一个执行结果，不会生成第二张优惠券

这个示例有意省略了数据库事务、真实权限、令牌签名和外部系统调用。它只能说明逻辑关系，不能直接承担生产审批。

生产实现至少要补充：

数据库唯一约束；
事务内的状态转移；
真实身份和权限校验；
令牌撤销；
草稿版本更新；
Worker 租约；
下游幂等键；
UNKNOWN 状态；
审计事件；
敏感数据保护；
超时和恢复任务。

十二、失败路径：审批系统最容易错在哪里

1. 模型把“用户想做什么”误认为“用户已经授权”

用户说：

帮我退款。

这表示申请意图，不表示已经确认具体金额、订单和退款渠道。

正确流程是：

识别订单
 -> 查询可退金额
 -> 生成退款草稿
 -> 展示金额和影响
 -> 请求确认
 -> 执行退款

如果 Agent 直接调用退款工具，模型实际上替用户完成了授权决策。

2. 用户确认后，数据已经发生变化

用户确认时订单状态是：

PAYMENT_CAPTURED

执行时订单已经被人工退款，状态变成：

REFUNDED

执行器必须在写操作前重新获取事实并检查守卫：

if order.status != "PAYMENT_CAPTURED":
    raise Conflict("订单状态已变化，原确认不能继续执行")

确认不是永久授权。它授权的是“在有效窗口内、对特定版本、在前置条件仍成立时执行”。

3. 草稿展示和实际执行参数不一致

界面显示：

退款 100 元

但执行器读取了 Agent 后续上下文中的：

{"amount": 1000}

这说明系统没有把确认绑定到冻结的结构化草稿。

正确做法是：

确认时记录 parameters_hash
执行时读取已确认草稿
执行前重新计算 parameters_hash
不一致则拒绝

4. 超时被当成失败并自动重试

下游响应超时可能有三种情况：

请求没有到达；
请求到达但未执行；
请求成功，响应丢失。

系统无法区分时，状态必须是 UNKNOWN。立即重试可能导致重复退款、重复发券或重复发信。

5. 审批人权限在确认后被撤销

如果权限只在生成草稿时检查，用户可能在确认前离职、降权或离开审批组。

至少在两个时间点检查权限：

生成草稿时：决定是否进入审批流程；
确认和执行时：确认当前身份仍有资格完成该动作。

高风险系统还应记录授权策略版本，以便审计时解释当时为什么允许。

6. 用对话历史恢复审批状态

对话历史可能被：

截断；
压缩；
修改；
重新组织；
混入新的用户指令；
在不同设备或会话中分叉。

因此，不能通过读取“用户之前说过确认”来恢复审批状态。状态恢复必须来自持久化的审批记录和事件。

OpenAI Agents SDK 的文档将“Agent run”和可恢复状态作为独立能力，并区分由应用自行控制循环的 Responses API 与由 SDK 管理 Agent 循环的 Agents SDK；无论采用哪种方式，审批状态都不应只存在于模型上下文中。(developers.openai.com)

十三、生产取舍：自动批准不是越多越好

1. 将低风险操作自动化

并非所有写操作都需要人工确认。例如：

创建内部草稿；
更新 Agent 自己生成的临时任务；
添加非敏感标签；
生成待发送但未发送的通知；
重复查询和缓存刷新。

但自动批准仍应经过策略引擎。低风险不是“模型认为低风险”，而是业务规则明确判定低风险。

2. 将高风险操作拆成两段

对于外部发送、资金变动、权限授予、生产删除等操作，最好将：

准备内容

和：

产生外部副作用

拆开。

例如邮件 Agent：

生成邮件草稿
 -> 检查收件人、附件和敏感信息
 -> 用户确认
 -> 发送

而不是让模型直接拥有：

send_email(to, subject, body)

3. 多级审批不要用多次布尔确认表示

错误设计：

{
  "manager_confirmed": true,
  "finance_confirmed": true
}

它无法表达：

两次确认是否针对同一版本；
谁在什么时候确认；
顺序是否重要；
某一级是否撤销；
是否满足职责分离；
版本变化后是否需要重新审批。

应当建模为审批步骤：

{
  "steps": [
    {
      "step": 1,
      "role": "line_manager",
      "status": "APPROVED",
      "actor_id": "u_1",
      "draft_version": 4
    },
    {
      "step": 2,
      "role": "finance",
      "status": "WAITING",
      "draft_version": 4
    }
  ]
}

如果草稿版本从 4 变成 5，所有依赖版本 4 的批准都应根据策略重新评估，而不是默认继承。

十四、测试与诊断：验证的是状态不变量

审批 Agent 的测试不能只测“模型是否生成了正确文案”，还要测状态机不变量。

1. 关键不变量

未确认不能执行

\text{status} \neq \text{CONFIRMED} \Rightarrow \text{no external write}

版本不匹配不能执行

\text{token.version} \neq \text{draft.version} \Rightarrow \text{reject}

同一幂等键最多产生一次业务副作用

\forall k,\quad \text{side\_effect\_count}(k) \leq 1

终态不能无故回退

SUCCEEDED -> WAITING_CONFIRMATION

通常应当被拒绝，除非存在明确的补偿流程。

审计事件顺序可解释

DRAFT_CREATED
 -> VALIDATION_PASSED
 -> CONFIRMATION_ISSUED
 -> APPROVAL_CONFIRMED
 -> EXECUTION_CLAIMED
 -> DOWNSTREAM_SUCCEEDED

不能出现：

DOWNSTREAM_SUCCEEDED
 -> APPROVAL_CONFIRMED

2. 必测故障注入

应模拟：

用户连续点击确认；
两个 Worker 同时执行；
确认令牌过期；
草稿确认前被修改；
权限在确认前被撤销；
下游连接超时；
下游成功但响应丢失；
执行器在写入前崩溃；
执行器在下游成功后崩溃；
审计写入失败；
消息重复投递；
Agent 输出非法参数；
用户试图通过自然语言绕过确认。

Anthropic 建议在 Agent 执行中持续从环境获得真实结果，并在检查点或阻塞时暂停等待人工反馈，同时设置最大迭代次数等停止条件；这些原则对应到审批系统，就是每次写操作前重新读取事实、显式设置确认点，并防止 Agent 在异常时无限循环。(anthropic.com)

3. 诊断一笔“重复退款”

遇到重复副作用时，不应先看模型输出，而应按以下顺序定位：

1. 找到 request_id
2. 找到 draft_version
3. 找到 approval_id 和 token_id
4. 找到 idempotency_key
5. 检查 approval_execution 唯一约束
6. 检查下游是否收到相同幂等键
7. 对比 DOWNSTREAM_REQUESTED 事件数量
8. 对比下游实际业务记录
9. 判断是重复确认、重复领取还是下游不支持幂等

如果同一个幂等键对应两条成功的下游记录，问题通常不在审批状态机，而在下游没有正确实现幂等，或执行器使用了不同的键。

十五、审批 Agent 的最小设计原则

一个足够稳健的审批 Agent，至少应当满足以下结构：

自然语言
  -> Agent 解析
  -> 确定性规范化
  -> 确定性校验
  -> 策略判断
  -> 不可变草稿
  -> 明确确认点
  -> 版本绑定的确认令牌
  -> 原子状态转移
  -> 幂等执行
  -> 未知结果对账
  -> 追加式审计

其中最容易被低估的是三个边界：

确认边界：用户确认的必须是具体版本和具体参数；
执行边界：每次外部写操作都必须有幂等语义；
证据边界：系统必须能够证明当时谁看到了什么、批准了什么、实际执行了什么。

Agent 可以提高审批材料准备和流程交互的效率，但它不应成为最终授权记录本身。把模型放在“理解、查询、起草和解释”的位置，把权限、状态、确认、幂等和审计放在确定性系统中，审批 Agent 才能从一个会话功能变成可恢复、可验证、可追责的业务组件。

系列导航与关联阅读

官方资料

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