OpenAI Agents SDK #14：Agent 刚要删库，你却毫无感知——Human in the Loop 全解析

2. 核心机制：needs_approval 标注工具

from agents import Agent, Runner, function_tool
from agents.run import RunConfig

# 普通工具——直接执行，无需审批
@function_tool
async def read_file(path: str) -> str:
    """读取文件内容"""
    with open(path) as f:
        return f.read()

# 危险工具——标注 needs_approval=True
# SDK 在执行此工具前会暂停，等待人工决定
@function_tool(needs_approval=True)
async def delete_database(db_name: str) -> str:
    """删除数据库（不可逆操作）"""
    # ... 实际删库逻辑
    return f"数据库 {db_name} 已删除"

agent = Agent(
    name="DBAdmin",
    instructions="你是数据库管理助手，按用户要求操作数据库。",
    tools=[read_file, delete_database],
)

from agents.run import ToolApprovalFunctionContext

def needs_approval(ctx: ToolApprovalFunctionContext) -> bool:
    """只有当操作目标是生产数据库时才需要审批"""
    db_name = ctx.tool_call.arguments.get("db_name", "")
    return "prod" in db_name or "production" in db_name

@function_tool(needs_approval=needs_approval)
async def delete_database(db_name: str) -> str:
    """删除数据库"""
    return f"数据库 {db_name} 已删除"

3. 完整工作流：从捕获到恢复

import asyncio
import json
from agents import Agent, Runner, function_tool

@function_tool(needs_approval=True)
async def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（需要审批）"""
    return f"邮件已发送至 {to}"

@function_tool
async def draft_email(to: str, subject: str, body: str) -> str:
    """起草邮件（无需审批，仅预览）"""
    return f"草稿已生成：收件人={to}, 主题={subject}"

agent = Agent(
    name="EmailAssistant",
    instructions="你是邮件助手，帮助用户起草和发送邮件。",
    tools=[draft_email, send_email],
)

async def main():
    user_input = "给 [email protected] 发一封主题为'Q2 报告'的邮件"

# 第一次运行：Agent 会生成发邮件的工具调用，然后因 needs_approval 暂停
    result = await Runner.run(agent, user_input)

# 处理中断循环：直到所有中断都被处理完毕
    while result.interruptions:
        print(f"\n⏸️  检测到 {len(result.interruptions)} 个待审批项：\n")

# 将当前运行状态转换为可操作的 RunState 对象
        # RunState 包含 SDK 管理的全部运行时元数据（审批决定、工具输入、嵌套 Agent 恢复点等）
        state = result.to_state()

for interruption in result.interruptions:
            print(f"  工具: {interruption.tool_name}")
            print(f"  代理: {interruption.agent_name}")
            print(f"  参数: {json.dumps(interruption.tool_call.arguments, ensure_ascii=False, indent=2)}")

# 请求人工确认
            decision = input("\n  是否批准？(y=批准 / n=拒绝 / e=修改参数): ").strip().lower()

if decision == "y":
                # 批准：允许工具继续执行
                # always_approve=True 表示本次运行中后续相同工具调用自动批准
                state.approve(interruption, always_approve=False)
                print("  ✅ 已批准")

elif decision == "n":
                # 拒绝：工具不执行，Agent 会收到拒绝消息并重新推理
                reason = input("  拒绝原因（可选）: ").strip()
                state.reject(interruption, rejection_message=reason or "用户拒绝")
                print("  ❌ 已拒绝")

else:
                # 修改参数后批准（此处示例直接批准，实际可修改 interruption）
                state.approve(interruption)
                print("  ✅ 已批准（参数未修改）")

# 使用处理后的 RunState 恢复运行
        # Runner.run() 接受 RunState 作为 input 参数，从中断点继续执行
        result = await Runner.run(agent, state)

print(f"\n✅ 运行完成：{result.final_output}")

asyncio.run(main())

• result.interruptions 是一个 ToolApprovalItem 列表，每项包含 agent_name、tool_name、tool_call（含参数）1
• result.to_state() 把 RunResult 转换为 RunState，这是执行审批操作的前提
• state.approve(interruption) 和 state.reject(interruption, rejection_message=...) 记录决定但不立即执行
• Runner.run(agent, state) 接受 RunState 作为 input，从中断点继续 Agent 循环

4. RunState：中断状态的完整结构与持久化

# 序列化：RunState → 字符串（适合写入数据库/文件/Redis）
state_str = state.to_string()
state_json = state.to_json()   # 返回 dict，方便进一步处理

# 反序列化：字符串 → RunState（在另一个进程/机器上恢复）
state = RunState.from_string(state_str)
state = RunState.from_json(state_json)

5. 生产级场景：持久化审批流程完整示例

import asyncio
import json
from pathlib import Path
from agents import Agent, Runner, function_tool
from agents.run import RunState

# ── 工具定义 ────────────────────────────────────────────────────────────────

def needs_database_approval(ctx) -> bool:
    """只有涉及 prod 环境的操作才需要审批"""
    args = ctx.tool_call.arguments
    return args.get("env", "dev") == "prod"

@function_tool(needs_approval=needs_database_approval)
async def run_sql(env: str, sql: str) -> str:
    """执行 SQL 语句
    Args:
        env: 环境（dev/staging/prod）
        sql: SQL 语句
    """
    # 实际执行逻辑...
    return f"[{env}] SQL 执行成功：{sql[:50]}..."

@function_tool
async def list_tables(env: str) -> str:
    """列出数据表（只读，无需审批）"""
    return f"[{env}] 表列表：users, orders, products"

# ── 持久化层（示例用文件，生产用 Redis/DB）──────────────────────────────────

PENDING_FILE = Path("/tmp/pending_approval.json")

async def save_pending_state(state: RunState, interruptions: list):
    """将待审批状态保存到持久化存储"""
    payload = {
        "state": state.to_json(),        # 序列化 RunState
        "interruptions": [
            {
                "tool_name": item.tool_name,
                "agent_name": item.agent_name,
                "arguments": item.tool_call.arguments,
                # interruption 本身不能直接序列化，这里保存 index 用于查找
                "index": i,
            }
            for i, item in enumerate(interruptions)
        ]
    }
    PENDING_FILE.write_text(json.dumps(payload, ensure_ascii=False, indent=2))
    print(f"✉️  审批请求已保存，等待人工处理...")

async def load_and_resume(agent: Agent):
    """从持久化存储加载状态并恢复运行（模拟审批服务）"""
    if not PENDING_FILE.exists():
        print("❌ 无待审批状态")
        return None

payload = json.loads(PENDING_FILE.read_text())

# 从 JSON 反序列化 RunState
    state = RunState.from_json(payload["state"])

print("\n📋 待审批操作列表：")
    for item in payload["interruptions"]:
        print(f"  [{item['index']}] 工具={item['tool_name']}, 环境={item['arguments'].get('env')}")
        print(f"      SQL={item['arguments'].get('sql', '')[:80]}")

# 注意：state.approve/reject 需要原始 interruption 对象
    # 这里需要重新运行一次来获取 interruptions（实际生产中可存储完整对象）
    # 简化示例：直接从状态推断并批准第一个
    print("\n✅ 自动批准所有操作（生产环境应弹出审批 UI）")

PENDING_FILE.unlink()  # 清理待审批文件
    return state

# ── 主流程 ───────────────────────────────────────────────────────────────────

async def main():
    agent = Agent(
        name="DBAAgent",
        instructions="你是数据库管理员助手。按指令执行 SQL 操作，需要时列出表信息。",
        tools=[list_tables, run_sql],
    )

user_input = "在 prod 环境执行：DELETE FROM orders WHERE status='cancelled' AND created_at < '2025-01-01'"

print("▶️  启动 Agent 运行...\n")
    result = await Runner.run(agent, user_input)

while result.interruptions:
        state = result.to_state()

# 将状态持久化（模拟发送审批通知）
        await save_pending_state(state, result.interruptions)

# 模拟等待审批（实际是异步等待人工操作）
        print("\n⏳ 模拟等待审批（实际场景中这里等待人工点击确认）...")
        await asyncio.sleep(1)

# 模拟审批服务加载状态
        state = await load_and_resume(agent)
        if state is None:
            break

# 重新运行以获取原始 interruption 对象并批准
        # 实际生产：审批服务持有 interruption 对象引用，直接调用 state.approve()
        result = await Runner.run(agent, state)

if result.final_output:
        print(f"\n✅ 运行完成：{result.final_output}")

asyncio.run(main())

6. Streaming 模式下的 HITL

async def streaming_with_hitl():
    agent = Agent(
        name="StreamAgent",
        instructions="你是一个流式输出的助手。",
        tools=[send_email],  # send_email 带 needs_approval=True
    )

async with Runner.run_streamed(agent, "给 [email protected] 发一封周报邮件") as result:
        # 消费流式事件，直到中断发生
        async for event in result.stream_events():
            if hasattr(event, 'delta') and event.delta:
                print(event.delta, end="", flush=True)

# 流结束后检查中断（流式模式下中断会终止流）
    if result.interruptions:
        state = result.to_state()
        for interruption in result.interruptions:
            print(f"\n⏸️  需要批准：{interruption.tool_name}({interruption.tool_call.arguments})")
            state.approve(interruption)  # 或 state.reject()

# 用批准后的 RunState 恢复（可以继续用 run_streamed 或 run）
        final = await Runner.run(agent, state)
        print(f"\n结果：{final.final_output}")

7. Sessions + Human in the Loop：上期预告兑现

import asyncio
from agents import Agent, Runner, function_tool
from agents.sessions import SQLiteSession
from agents.run import RunState

@function_tool(needs_approval=True)
async def transfer_money(amount: float, to_account: str) -> str:
    """转账操作（需要审批）"""
    return f"已转账 {amount} 元至 {to_account}"

@function_tool
async def check_balance() -> str:
    """查询余额（无需审批）"""
    return "当前余额：¥12,580.00"

agent = Agent(
    name="BankAgent",
    instructions="你是银行智能助手，帮助用户查询和管理账户。",
    tools=[check_balance, transfer_money],
)

async def main():
    # Session 实例：贯穿整个对话生命周期
    # 中断前后必须复用同一个实例
    session = SQLiteSession("bank_session_001", db_path="/tmp/bank_session.db")

user_messages = [
        "查一下我的余额",
        "把 5000 元转到账户 6217xxxxxx",  # 这条会触发审批
        "转账成功了吗",
    ]

for user_msg in user_messages:
        print(f"\n👤 用户：{user_msg}")

# 每次运行传入同一个 session 实例
        # SDK 自动注入历史，无需手动 to_input_list()
        result = await Runner.run(agent, user_msg, session=session)

# 处理可能的中断
        while result.interruptions:
            state = result.to_state()

for interruption in result.interruptions:
                print(f"\n⚠️  审批请求：{interruption.tool_name}")
                print(f"   参数：{interruption.tool_call.arguments}")

decision = input("   批准？(y/n): ").strip().lower()
                if decision == "y":
                    state.approve(interruption)
                else:
                    state.reject(interruption, rejection_message="用户取消")

# 关键：恢复运行时继续传入同一个 session 实例
            # 这样 Session 才能正确追踪整条对话链
            result = await Runner.run(agent, state, session=session)

print(f"🤖 Agent：{result.final_output}")

asyncio.run(main())

8. 关于 InterruptedError：官方文档说明

1. Runner.run() 正常返回（不抛出异常）
2. 返回的 RunResult 对象上的 result.interruptions 字段非空
3. 通过检查 result.interruptions 来判断是否发生了中断

9. HITL 与 Guardrails 的边界

10. 社区视角：2026 年的行业共识

3 条可直接落地的实践建议