写操作、人工确认与幂等

先制造两张重复工单

模型调用 create_ticket 后，客户端正好断网。客户端重试，服务端看不到第一次响应，于是再次执行：

TICKET-000042  VPN 连接失败
TICKET-000043  VPN 连接失败

“模型只调用了一次”不是防线。重试、超时和消息重复投递都会让服务端收到两次请求。任何有副作用的工具都必须幂等。

工具元数据决定是否需要确认

@tool(name='create_ticket', side_effect='write', approval='required')
async def create_ticket(*, user: User, title: str, description: str, idempotency_key: str):
    existing = await audits.find_success(user.id, idempotency_key)
    if existing:
        return existing.result
    ticket = await ticket_service.create(title, description, user.id)
    await audits.record_success(user.id, idempotency_key, ticket.to_public_dict())
    return ticket.to_public_dict()

读工具可以自动执行；写工具先返回一个可展示给用户的计划：

{
  "code": "approval_required",
  "conversation_id": "c_123",
  "approval_id": "a_456",
  "summary": "创建工单：VPN 连接失败",
  "arguments": {"title":"VPN 连接失败","description":"办公网无法访问"}
}

update_ticket：状态迁移不由模型裁决

第二个写工具是 update_ticket。它比 create_ticket 更危险，因为它作用在已存在的数据上：

@tool(name='update_ticket', side_effect='write', approval='required')
async def update_ticket(*, user: User, ticket_id: str, status: str, idempotency_key: str):
    ticket = await ticket_service.get_for_user(ticket_id, user)   # 权限校验在这里
    if ticket is None:
        raise ToolError('ticket_not_found')
    try:
        updated = ticket.transition(TicketStatus(status))          # 第 1 篇的状态机
    except InvalidTransition as exc:
        raise ToolError('invalid_transition', detail=str(exc))     # 回注给模型，不是 500
    return (await ticket_service.save(updated)).to_public_dict()

这里能看到前几篇的设计在同一个函数里汇合：权限来自第 6 篇的 current_user，合法迁移来自第 1 篇的 ALLOWED 表，错误结构化回注来自第 3 篇。模型能提议把工单从 closed 改回 open，但裁决它的是那张表，不是提示词。

status 是枚举，必须写进 JSON Schema 的 enum 里。不写的话模型会发明 "完成"、"done"、"CLOSED" 这类值，每一个都要在运行时才失败。

挂起、确认、恢复

审批状态不能只放进 Python 内存，否则进程重启后按钮会变成“确认成功但没有工单”。存一条待审批记录：

create table approvals (
  id uuid primary key,
  conversation_id text not null,
  user_id uuid not null references users(id),
  tool_name text not null,
  arguments jsonb not null,
  status text not null check (status in ('pending','approved','rejected','expired')),
  expires_at timestamptz not null,
  created_at timestamptz not null default now()
);

POST /approve/{approval_id} 必须重新读取用户身份并比较 approval.user_id，不能相信浏览器传来的用户 ID。确认后恢复原会话，工具再次经过 schema、权限和幂等校验。

Preview

提出、挂起、等待、确认、执行的完整时序

图中间那段虚线框是这套设计的全部意义所在：等待期的长度不可控，可能几秒，也可能用户吃完午饭才回来点确认。只要状态在数据库里，这期间服务重启、滚动发布、请求落到另一个副本，都不影响这次审批继续走完。

async def approve(approval_id: UUID, user: User):
    approval = await approvals.lock_for_update(approval_id)
    if not approval or approval.user_id != user.id:
        raise HTTPException(404, 'approval_not_found')
    if approval.status != 'pending' or approval.expires_at < now():
        raise HTTPException(409, 'approval_expired')
    await approvals.mark(approval.id, 'approved')
    # 执行该写操作并落审计表；第 13 篇会把这里的恢复统一接入 LangGraph 状态机
    return await execute_approved_tool(approval, user=user)

幂等键的来源和约束

客户端为一次用户意图生成 UUID；Agent 重试必须复用原键，不能每次 uuid4()：

create_ticket_key = f'{conversation_id}:{tool_call_id}'

把有键和无键两条路径并排看，这个设计要解决的问题就很直观：

Preview

同样是超时重试，有无幂等键的两种结局

数据库唯一索引保证并发请求只有一个成功：

create unique index audits_idempotency_idx
on tool_call_audits(user_id, idempotency_key)
where status = 'succeeded';

参数变化时不能返回旧结果。审计表保存参数 hash，发现同键不同 hash 返回 idempotency_conflict，提醒调用方生成新意图。

审计表：每次工具调用都要留痕

上面那个唯一索引建在 tool_call_audits 上，这张表本身值得完整写出来。它不只是为幂等服务——第 15 篇排查「为什么这次请求慢」和「这个月成本花在哪」，靠的都是它：

create table tool_call_audits (
  id uuid primary key,
  conversation_id uuid not null,
  user_id uuid not null references users(id),
  tool_name text not null,
  arguments jsonb not null,          -- 请求参数
  arguments_hash text not null,      -- 用于同键不同参的冲突检测
  result jsonb,                      -- 响应，失败时为 null
  status text not null check (status in ('succeeded','failed','rejected')),
  error_code text,                   -- 结构化错误码，不存堆栈
  duration_ms integer not null,      -- 耗时
  idempotency_key text,
  approval_id uuid references approvals(id),   -- 高风险调用关联到审批记录
  created_at timestamptz not null default now()
);
create index audits_conversation_idx on tool_call_audits(conversation_id, created_at);

几个字段的取舍：

arguments 用 jsonb 而不是 text，是为了能直接按参数查询——排查「谁在批量创建工单」时很有用。
error_code 只存错误码。堆栈应该进日志，不进业务表：它会让这张表迅速膨胀，而且包含内部路径。
approval_id 是可空外键。它把「这次调用执行了」和「谁批准了它」连起来，是事后审计唯一能追责的链路。没有这一列，你只知道工单被创建了，不知道是谁点的确认。

超时、重试和结构化错误

只对明确可重试的错误重试两次：连接重置、503、限流；参数错误、权限拒绝、审批拒绝不重试。

RETRYABLE = {'upstream_timeout', 'temporarily_unavailable', 'rate_limited'}
for attempt in range(3):
    try:
        return await asyncio.wait_for(tool(**args), timeout=8)
    except ToolError as exc:
        if exc.code not in RETRYABLE or attempt == 2:
            return {'ok': False, 'error': exc.to_dict()}
        await asyncio.sleep(0.2 * 2 ** attempt)

错误返回 {code, message, retryable, request_id}，不返回堆栈。

测试和验收

uv run pytest tests/agent/test_approval.py tests/agent/test_idempotency.py -q
uv run pytest tests/agent/test_idempotency.py -k concurrent -vv

验收标准：点击确认只创建一张工单；重复确认返回同一结果；用户 A 不能批准用户 B 的请求；过期审批不会执行；参数错误不触发重试；高风险工具有完整请求、响应、耗时和审批关联记录。

下一篇把真实文档喂给系统。我们会先让 80 页 PDF 把同步接口拖到超时，再引入异步索引。

#写操作、人工确认与幂等

#先制造两张重复工单

#工具元数据决定是否需要确认

#update_ticket：状态迁移不由模型裁决

#挂起、确认、恢复

#幂等键的来源和约束

#审计表：每次工具调用都要留痕

#超时、重试和结构化错误

#测试和验收