教程目标

本文实现一条完整链路：Agent 调用高风险 Tool 前暂停，保存 RunState；审批服务在另一个进程批准或拒绝；Worker 加载状态继续运行；最后用幂等账本阻止队列重复投递造成的重复副作用。

环境：Python 3.10.2、openai-agents==0.18.3、SQLite。实验使用确定性自定义 Model，不调用外部模型 API。

步骤一：定义需要审批的 Tool

from agents import function_tool

@function_tool(needs_approval=True)
def deploy_release(release: str, idempotency_key: str) -> str:
    return f"deployed:{release}"

needs_approval=True 会让 Runner 在工具函数执行前产生 interruption，因此暂停时副作用应为 0。

步骤二：运行并保存 RunState

result = await Runner.run(agent, "Deploy release 2026.07.24", context=context)
assert result.interruptions
state = result.to_state()

自定义 Context 需要 Serializer：

def context_serializer(context: AppContext):
    return {"tenant_id": context.tenant_id}

state_json = state.to_json(
    context_serializer=context_serializer,
    strict_context=True,
)

不要把 API Key、Access Token、数据库连接和 Client 对象写入持久化 Context。普通 Dict 会直接序列化，严格模式不会自动识别秘密字段。

步骤三：创建审批工单

RunState 只保存 SDK 运行状态，业务系统还需要独立工单：

CREATE TABLE approval_request (
  approval_id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  tool_call_id TEXT NOT NULL,
  tool_name TEXT NOT NULL,
  arguments_digest TEXT NOT NULL,
  app_state_version TEXT NOT NULL,
  sdk_schema_version TEXT NOT NULL,
  state_blob_uri TEXT NOT NULL,
  status TEXT NOT NULL,
  expires_at TIMESTAMP NOT NULL
);

创建工单时，把 Tool 参数规范化后计算 SHA-256 摘要。Worker 恢复前再次计算，防止审批页面看到的参数与最终执行参数不一致。

步骤四：审批或拒绝

state = await RunState.from_json(
    initial_agent,
    state_json,
    context_deserializer=context_deserializer,
    strict_context=True,
)

interruption = state.get_interruptions()[0]
state.approve(interruption)

拒绝使用：

state.reject(
    interruption,
    rejection_message="Deployment was denied by the release manager.",
)

拒绝后仍需恢复 Runner，让 Agent 接收拒绝结果并正常结束，但 Tool 不会执行。

步骤五：Worker 恢复

agent = AGENT_FACTORIES[record.app_state_version](runtime_dependencies)
state = await load_state(agent)
result = await Runner.run(agent, state)

状态 JSON 不包含可执行 Python 函数。必须保留兼容 Agent Factory，并固定 Tool 名称、参数 Schema、关键 Prompt 和 Handoff 拓扑。

常见故障一：同一状态执行两次

实验把同一份已批准 RunState 重放给两个 Worker，结果 effect_count=2。原因是队列通常只能保证至少一次投递，RunState 不提供业务 Exactly-once。

解决方案是在 Tool 真正产生副作用前抢占稳定幂等键：

connection.execute("BEGIN IMMEDIATE")
row = connection.execute(
    "SELECT result FROM idempotency_ledger WHERE idempotency_key=?",
    (key,),
).fetchone()
if row:
    return row[0]
# execute external effect, then store result

幂等键由业务系统生成并绑定 tenant_id + approval_id + operation + resource，不要让模型自由生成。

常见故障二：Context 泄漏 Token

Mapping Context 中的测试 Secret 被原样写入 RunState JSON。应只保存最小标识，恢复时从 Secret Manager 重新注入。State Blob 本身仍可能包含用户输入、Tool 参数和模型输出，因此需要加密、租户 ACL、TTL 和读取审计。

常见故障三：旧状态被新代码恢复

SDK 能反序列化旧状态，不等于旧审批仍然合法。Worker 必须检查应用状态版本、SDK Schema、Agent Revision、参数摘要、过期时间和当前权限。无法兼容时返回“重新生成审批”，不能强行执行。

最终架构

Agent API -> encrypted RunState Blob + approval_request
Approval API -> permission check + CAS decision
Queue -> approval_id only
Worker -> versioned Agent Factory + RunState restore
Idempotency Ledger -> operation ownership + result reuse
Audit -> approver, arguments digest, worker, external result

完整可复现实验、运行结果和生产检查清单：https://www.xbstack.com/ai/openai-agents-sdk-runstate-approval-resume/?utm_source=csdn&utm_medium=community&utm_campaign=openai_agents_runstate_approval_resume_20260724&utm_content=article_body&ref=csdn