DeepAgents的人工介入

介绍

Deep Agents 中基于 LangGraph 的「人工介入（Human-in-the-loop, HITL）」工作流，核心是为高风险 AI 操作（如删文件、发邮件）提供人工审核机制，确保操作安全。

• 背景问题：AI 直接执行某些工具操作（如删除文件、发送邮件）有较高风险，需要人工审核。

• 解决方案：Deep Agents 利用 LangGraph 的「中断机制」，实现完整的 HITL 人工介入流程。

• 控制方式：通过配置 interrupt_on 参数，你可以精确指定：

  • 哪些工具操作需要人工审批

  • 哪些可以直接自动执行，从而在灵活性和安全性之间做平衡。

flowchart LR
    A[Agent] --> B{Interrupt?}
    B -- no --> E[Execute]
    B -- yes --> C{Human}
    C -- approve --> E
    C -- edit --> E
    C -- reject --> D[Cancel]
    E --> A
    D --> A

节点 / 分支	含义
Agent	AI 智能体，负责生成工具调用请求（比如 “删除文件”“发送邮件”）。
Interrupt?	LangGraph 的「中断判断节点」，根据你配置的 interrupt_on 规则，判断当前操作是否需要人工介入。
→ no（不需要中断）	直接进入 Execute 节点，自动执行操作。
→ yes（需要中断）	进入 Human 人工审核节点。
Human（人工审核）	人工对 AI 的操作请求进行处理，有三种选择：
→ approve（批准）	同意执行，直接进入 Execute。
→ edit（修改后执行）	修改 AI 的请求（比如调整邮件内容、修改删除的文件列表），再进入 Execute。
→ reject（拒绝）	取消本次操作，进入 Cancel 节点。
Execute	执行工具操作，完成后流程回到 Agent，继续后续任务。
Cancel	取消本次高风险操作，流程回到 Agent，AI 可以继续处理其他任务或调整方案。

基本配置

interrupt_on 是一个字典，用来给每个工具配置 “是否需要人工审核” 以及 “审核时允许哪些操作”。它有三种配置方式：

配置值	含义
True	启用默认中断行为：人工审核时可以 approve（批准）、edit（修改）、reject（拒绝）
False	禁用此工具的中断：工具操作不需要人工审核，直接自动执行
{"allowed_decisions": [...]}	自定义配置：人工审核时只允许列表里的决策类型（比如只允许批准 / 拒绝，不允许修改）

# 导入依赖
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# 定义三个示例工具（此处省略具体实现）
# delete_file：删除文件（高风险）
# read_file：读取文件（低风险）
# send_email：发送邮件（中高风险）

# 1. 配置状态持久化：必须项
checkpointer = MemorySaver()

# 2. 创建配置了人工介入的智能代理
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",  # 使用的大语言模型
    tools=[delete_file, read_file, send_email],  # 智能代理可用的工具列表
    interrupt_on={
        "delete_file": True,  
        # 删除文件需要审核，使用默认决策类型（approve/edit/reject 都允许）
        "read_file": False,  
        # 读取文件不需要审核，可直接执行
        "send_email": {"allowed_decisions": ["approve", "reject"]},  
        # 发送邮件需要审核，但不允许编辑内容，只能批准或拒绝
    },
    checkpointer=checkpointer  # 必需参数，用于状态持久化
)

• 关键作用：人工介入功能必须搭配 Checkpointer（这里用的是 MemorySaver），用来保存会话状态。

• 为什么需要它？ 当流程被中断（等待人工审核）时，系统需要记住当前的执行状态，人工处理完后才能恢复到正确的位置继续执行。

这里的 interrupt_on 配置，就是核心规则：

1. delete_file: True
   • 场景：删除文件属于高风险操作，需要人工审核。
   • 权限：人工审核时，既可以直接批准、拒绝，也可以修改删除的文件列表后再执行。
2. read_file: False
   • 场景：读取文件是低风险操作，不需要人工审核。
   • 权限：AI 生成读取请求后，会直接执行，不会触发中断。
3. send_email: {"allowed_decisions": ["approve", "reject"]}
   • 场景：发送邮件属于中高风险操作，需要人工审核。
   • 权限：人工只能选择「批准发送」或「拒绝发送」，不能修改邮件内容（避免篡改信息）。

决策类型

当工具操作被中断等待人工审核时，审核人员可以采取以下三种决策：

决策类型	含义	适用场景
approve	批准并执行操作	确认 AI 的请求安全、无误，同意直接执行
reject	拒绝操作并终止执行	认为 AI 的请求有风险或错误，直接取消本次操作
edit	修改操作参数后执行（需显式允许）	保留执行，但修改 AI 的请求参数（比如调整删除的文件、修改邮件内容），再继续执行

注意：edit 必须在 allowed_decisions 里显式声明，否则审核界面不会出现 “编辑” 选项。

interrupt_on = {
    # 高风险操作：允许所有决策类型
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},

    # 中等风险操作：只允许批准或拒绝，不允许修改参数
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # 关键操作：必须执行，只允许批准
    "critical_operation": {"allowed_decisions": ["approve"]},
}

1. delete_file（删除文件，高风险）

   • 配置：["approve", "edit", "reject"]

   • 说明：允许审核人员做任何操作：可以直接批准删除、拒绝删除，也可以修改要删除的文件列表后再执行。

   • 场景：删除文件影响大，需要最大的人工灵活性，既可以纠错，也可以直接阻止。

2. write_file（写入文件，中等风险）

   • 配置：["approve", "reject"]

   • 说明：只能批准或拒绝，不允许修改参数。

   • 场景：写入文件的内容是固定的（比如生成报告），不希望人工篡改内容，只需要确认是否执行。

3. critical_operation（关键操作，如系统初始化）

   • 配置：["approve"]

   • 说明：只允许批准，不能拒绝，也不能修改。

   • 场景：这个操作是任务流程的必要环节，必须执行，人工只能确认，不能干预执行本身。

allowed_decisions 的作用，是让你为不同工具设置不同的「人工干预权限」：

• 高风险操作：给足人工控制权，支持修改、批准、拒绝。
• 中等风险操作：限制人工权限，只能批准 / 拒绝，不能篡改内容。
• 关键操作：只允许人工确认，确保流程必须推进。

这种配置方式，让你可以在安全性、灵活性和流程刚性之间做精准平衡，既不会因为过度审核影响效率，也不会因为权限过大带来新的风险。

处理中断

当智能代理执行到需要人工审核的工具操作时，会自动暂停执行，并返回包含中断信息的结果。你需要完成这几个关键步骤：

1. 检查结果中是否包含 __interrupt__ 字段（判断是否触发了人工审核）
2. 提取待审核的操作信息
3. 获取用户的决策（批准、拒绝或修改）
4. 使用相同的会话配置恢复执行

import uuid
from langgraph.types import Command
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# ---------------------- 示例工具定义（省略具体实现） ----------------------
@tool
def delete_file(file_path: str) -> str:
    """删除指定路径的文件"""
    return f"文件 {file_path} 已删除"

@tool
def read_file(file_path: str) -> str:
    """读取指定路径的文件内容"""
    return "文件内容"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送至 {to}"

# ---------------------- 1. 配置状态持久化 ----------------------
checkpointer = MemorySaver()

# ---------------------- 2. 创建带人工介入的智能代理 ----------------------
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[delete_file, read_file, send_email],
    interrupt_on={
        "delete_file": True,
        "read_file": False,
        "send_email": {"allowed_decisions": ["approve", "reject"]},
    },
    checkpointer=checkpointer
)

# ---------------------- 3. 创建会话配置（必须使用唯一的thread_id） ----------------------
# 创建带有唯一thread_id的配置，用于持久化会话状态
# 这个ID将在中断和恢复之间保持一致
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# ---------------------- 4. 向智能代理发送用户请求（触发中断） ----------------------
result = agent.invoke({
    "messages": [{"role": "user", "content": "删除文件temp.txt"}]
}, config=config)

# ---------------------- 5. 处理中断与人工审核 ----------------------
# 检查智能代理是否因为需要人工审核而中断了执行
if result.get("__interrupt__"):
    # 提取中断信息，包括待审核的操作和允许的决策类型
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]  # 待审核的操作列表
    review_configs = interrupts["review_configs"]    # 各工具的审核配置

    # 创建工具名称到审核配置的映射，方便快速查找
    config_map = {cfg["action_name"]: cfg for cfg in review_configs}

    # 向用户展示所有待审核的操作
    for action in action_requests:
        review_config = config_map[action["name"]]
        print(f"待审核工具: {action['name']}")
        print(f"操作参数: {action['args']}")
        print(f"允许决策: {review_config['allowed_decisions']}")

    # 获取用户的决策（此处为示例，实际应用中应通过界面或终端获取）
    decisions = [
        {"type": "approve"}  # 用户选择批准删除操作
    ]

    # 使用用户决策恢复智能代理的执行
    # 注意：必须使用与之前相同的config配置（包含相同的thread_id）
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

# ---------------------- 6. 处理最终执行结果 ----------------------
print("代理最终响应:", result["messages"][-1].content)

完整展示了人工介入流程的 “闭环”：

1. 触发中断：AI 执行到高风险操作时自动暂停。
2. 展示信息：把待审核的操作、参数和规则展示给用户。
3. 用户决策：用户选择批准、拒绝或修改参数。
4. 恢复执行：用同一个 thread_id 和用户决策，让 AI 继续执行。
5. 得到结果：AI 执行完所有操作后，返回最终响应。

这个流程的关键是：thread_id 必须全程一致，否则 LangGraph 无法恢复会话状态，中断机制就会失效。

代码分步骤详解

1. 初始化依赖与会话配置

   import uuid
   from langgraph.types import Command
   
   # 1. 创建带有唯一thread_id的配置，用于持久化会话状态
   # 这个ID将在中断和恢复之间保持一致
   config = {"configurable": {"thread_id": str(uuid.uuid4())}}

   • 关键作用：创建一个唯一的 thread_id，绑定到会话配置中。
   • 为什么必须这么做？ LangGraph 的 Checkpointer 会用 thread_id 来保存和恢复会话状态，中断和恢复时必须使用同一个 thread_id，否则无法找到之前的执行上下文。

2. 触发 AI 任务，等待中断

   # 向智能代理发送用户请求（删除文件temp.txt）
   result = agent.invoke({
       "messages": [{"role": "user", "content": "删除文件temp.txt"}]
   }, config=config)

   • 这里向代理发送了一个 “删除文件” 的请求。

   • 因为 delete_file 工具配置了 interrupt_on=True，代理执行到这一步时，会自动暂停，返回包含中断信息的 result。

3. 检查是否触发了中断

   # 检查智能代理是否因为需要人工审核而中断了执行
   if result.get("__interrupt__"):

   • result.get("__interrupt__") 是判断中断的关键标志：

     • 如果为 True，说明 AI 触发了需要人工审核的操作，流程暂停。

     • 如果为 False，说明所有操作都自动执行完成，没有需要审核的步骤。

4. 提取中断信息

   # 提取中断信息，包括待审核的操作和允许的决策类型
   interrupts = result["__interrupt__"][0].value
   action_requests = interrupts["action_requests"]  # 待审核的操作列表
   review_configs = interrupts["review_configs"]    # 各工具的审核配置

   • interrupts 里包含了本次中断的所有关键信息：

     • action_requests：AI 发起的所有待审核操作（比如这里的 delete_file(temp.txt)）。

     • review_configs：每个工具对应的审核规则（比如允许 approve/edit/reject）。

5. 构建审核配置映射

   # 创建工具名称到审核配置的映射，方便快速查找
   config_map = {cfg["action_name"]: cfg for cfg in review_configs}

   • 把 review_configs 转换成一个字典，key 是工具名称，方便后续快速查找每个工具允许的决策类型。

6. 向用户展示待审核操作

   # 向用户展示所有待审核的操作
   for action in action_requests:
       review_config = config_map[action["name"]]
       print(f"待审核工具: {action['name']}")
       print(f"操作参数: {action['args']}")
       print(f"允许决策: {review_config['allowed_decisions']}")

   • 把待审核的工具、参数、允许的操作（批准 / 拒绝 / 编辑）展示给用户，让用户了解 AI 要做什么、能做什么。

7. 获取用户的审核决策

   # 获取用户的决策（此处为示例，实际应用中应通过界面或终端获取）
   decisions = [
       {"type": "approve"}  # 用户选择批准删除操作
   ]

   • 这里模拟了用户的决策：approve（批准执行）。

   • 实际场景中，你可以通过前端界面、终端输入等方式让用户选择：

     • {"type": "approve"}：批准执行

     • {"type": "reject"}：拒绝执行

     • {"type": "edit", "args": {"file_path": "new_temp.txt"}}：修改参数后执行

8. 恢复代理执行

   # 使用用户决策恢复智能代理的执行
   # 注意：必须使用与之前相同的config配置（包含相同的thread_id）
   result = agent.invoke(
       Command(resume={"decisions": decisions}),
       config=config
   )

   • 核心是 Command(resume={"decisions": decisions})：告诉 LangGraph “我已经完成了人工审核，请根据用户的决策恢复执行”。

   • 必须传入和第一步相同的 config（包含同一个 thread_id），LangGraph 才能从之前中断的位置继续执行。

9. 处理最终结果

   # 处理智能代理的最终执行结果
   print("代理最终响应:", result["messages"][-1].content)

   • 代理根据用户的决策执行完操作后，会返回最终的响应结果，这里打印了最后一条消息的内容。

多个工具调用

当 AI 代理需要连续执行多个需要人工审核的工具操作（比如同时执行「删除文件」和「发送邮件」）时，LangGraph 会把这些操作批量收集到同一个中断中，而不是每个操作都触发一次中断。

这种设计的关键规则：

• 所有待审核操作会一次性返回在 action_requests 列表里
• 你提供的 decisions 列表，必须和 action_requests 的顺序一一对应，不能乱序

import uuid
from langgraph.types import Command
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# ---------------------- 1. 定义示例工具 ----------------------
@tool
def delete_file(file_path: str) -> str:
    """删除指定路径的文件"""
    return f"文件 {file_path} 已删除"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送至 {to}"

# ---------------------- 2. 配置人工介入与状态持久化 ----------------------
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[delete_file, send_email],
    interrupt_on={
        "delete_file": True,          # 删除文件需要审核（允许approve/edit/reject）
        "send_email": {"allowed_decisions": ["approve", "reject"]}  # 发送邮件只允许批准/拒绝
    },
    checkpointer=checkpointer
)

# ---------------------- 3. 创建会话配置 ----------------------
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# ---------------------- 4. 触发包含多个工具调用的请求 ----------------------
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "删除temp.txt文件并向admin@example.com发送确认邮件"
    }]
}, config=config)

# ---------------------- 5. 处理批量中断 ----------------------
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]

    # 验证待审核操作数量
    assert len(action_requests) == 2
    print(f"共有 {len(action_requests)} 个操作需要审核")

    # 展示每个待审核操作
    for i, action in enumerate(action_requests, 1):
        print(f"[{i}] 工具: {action['name']}，参数: {action['args']}")

    # 按顺序提供决策（必须和action_requests顺序一致）
    decisions = [
        {"type": "approve"},  # 批准第一个操作：删除文件
        {"type": "reject"}    # 拒绝第二个操作：发送邮件
    ]

    # 恢复执行
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

# ---------------------- 6. 输出最终结果 ----------------------
print("代理最终响应:", result["messages"][-1].content)

关键注意事项

1. 顺序必须严格匹配：decisions 列表的顺序，必须和 action_requests 里工具调用的顺序完全一致，否则会出现决策和操作不匹配的错误。
2. 每个操作都要有决策：即使你只想处理其中一个操作，也必须为所有 action_requests 提供决策，不能只给部分。
3. thread_id 必须一致：中断和恢复时，必须使用同一个 config，否则 LangGraph 无法找到会话状态。

编辑工具参数

当工具的 allowed_decisions 配置中包含 "edit" 时，审核人员可以：

• 不直接批准 / 拒绝操作，而是修改操作的参数
• 用修改后的参数恢复执行，相当于 “纠正” AI 的请求

edit 决策的关键是 edited_action 字段，它需要包含：

• name：要修改的工具名称（确保和原操作一致）
• args：修改后的参数（可以只改部分，保留其他参数不变）

import uuid
from langgraph.types import Command
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# 定义发送邮件工具
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送至 {to}，主题：{subject}"

# 配置状态持久化
checkpointer = MemorySaver()

# 创建代理，配置send_email工具允许edit决策
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[send_email],
    interrupt_on={
        "send_email": {"allowed_decisions": ["approve", "reject", "edit"]}
    },
    checkpointer=checkpointer
)

# 创建会话配置
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 触发AI请求（AI可能生成了发给所有人的邮件）
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "发送一封通知邮件"
    }]
}, config=config)

# 处理中断与参数编辑
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_request = interrupts["action_requests"][0]

    print("原始参数:", action_request["args"])

    # 构建edit决策
    decisions = [{
        "type": "edit",
        "edited_action": {
            "name": action_request["name"],
            "args": {
                "to": "team@company.com",
                "subject": "团队通知",
                "body": action_request["args"]["body"]
            }
        }
    }]

    # 恢复执行
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

# 输出最终结果
print("代理最终响应:", result["messages"][-1].content)

关键注意事项

1. edit 必须显式允许：工具的 allowed_decisions 里必须包含 "edit"，否则审核时无法使用该决策类型。
2. edited_action 格式必须正确：
   • 必须包含 name 和 args 字段。
   • args 必须是工具接受的合法参数，不能出现额外字段。
3. 可以修改任意参数：不仅可以修改收件人，还可以修改文件路径、查询条件等所有工具参数，只要符合工具定义即可。

子代理中断

在复杂的多代理系统里，主代理可以创建多个子代理，分别处理不同任务。

• 每个子代理都可以定义自己的 interrupt_on 配置。
• 子代理的配置优先级更高，会覆盖主代理的全局设置。
• 子代理触发中断后的处理方式，和主代理完全一样：检查 __interrupt__ → 获取决策 → 用 Command 恢复执行。

import uuid
from langgraph.types import Command
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# ---------------------- 定义工具 ----------------------
@tool
def delete_file(file_path: str) -> str:
    """删除指定路径的文件"""
    return f"文件 {file_path} 已删除"

@tool
def read_file(file_path: str) -> str:
    """读取指定路径的文件内容"""
    return "文件内容示例"

# ---------------------- 状态持久化 ----------------------
checkpointer = MemorySaver()

# ---------------------- 创建带子代理的智能代理系统 ----------------------
agent = create_deep_agent(
    tools=[delete_file, read_file],  # 主代理可用工具
    interrupt_on={
        "delete_file": True,   # 主代理：删除文件需要审核
        "read_file": False,    # 主代理：读取文件不需要审核
    },
    # 子代理配置
    subagents=[{
        "name": "file-manager",
        "description": "专业的文件管理助手",
        "system_prompt": "您是一个谨慎的文件管理助手，负责处理所有文件操作请求。",
        "tools": [delete_file, read_file],  # 子代理可用工具
        "interrupt_on": {
            "delete_file": True,   # 继承/显式声明需要审核
            "read_file": True,     # 覆盖主代理：子代理读取文件也需要审核
        }
    }],
    checkpointer=checkpointer  # 必须配置 checkpointer
)

# ---------------------- 会话配置 ----------------------
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# ---------------------- 触发任务（会交给子代理处理） ----------------------
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "读取并检查文件report.txt，然后删除临时文件temp.txt"
    }]
}, config=config)

# ---------------------- 处理中断（主/子代理的中断处理逻辑完全相同） ----------------------
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]
    print(f"共有 {len(action_requests)} 个操作需要审核（来自子代理）")

    # 这里模拟用户决策：批准读取文件，批准删除文件
    decisions = [
        {"type": "approve"},
        {"type": "approve"}
    ]

    # 恢复执行
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

# ---------------------- 输出结果 ----------------------
print("代理最终响应:", result["messages"][-1].content)

中断处理规则

当子代理触发中断时，处理方式与主代理相同 —— 检查 __interrupt__ 并使用 Command 恢复执行。

不管是主代理还是子代理触发的中断，你在外部调用时的处理逻辑是完全一样的：

1. 检查 result.get("__interrupt__") 是否存在。
2. 提取 action_requests 等信息。
3. 收集用户决策（approve/reject/edit）。
4. 用同一个 config + Command(resume=...) 恢复执行。

• 你不需要区分中断来自主代理还是子代理，外部接口是统一的。

遵守人工介入规则

1.始终使用checkpointer

人工介入流程中，当 AI 触发需要审核的操作时，会暂停执行，等待人工决策。

• 暂停时，AI 的会话状态（包括对话历史、工具调用上下文、中间结果等）必须被保存下来。
• 人工审核完成后，系统需要从暂停的位置恢复执行，而不是从头再来。
• checkpointer（状态检查点）就是用来持久化保存会话状态的组件，是 HITL 功能的基础。

如果不配置 checkpointer，中断时状态会丢失，人工审核后无法恢复执行，整个流程就会失效

from langgraph.checkpoint.memory import MemorySaver
from langchain.tools import tool
from deepagents import create_deep_agent

# 示例工具
@tool
def delete_file(file_path: str) -> str:
    return f"文件 {file_path} 已删除"

@tool
def read_file(file_path: str) -> str:
    return "文件内容"

# 1. 初始化checkpointer（HITL必需）
checkpointer = MemorySaver()

# 2. 创建带人工介入配置的智能代理
agent = create_deep_agent(
    tools=[delete_file, read_file],
    interrupt_on={
        "delete_file": True,  # 删除文件需要审核
        "read_file": False    # 读取文件无需审核
    },
    checkpointer=checkpointer  # 关键：绑定状态检查点
)

2.保持thread ID的一致性

thread_id 是 LangGraph 中会话状态的唯一标识，中断和恢复时必须使用完全相同的 thread_id，否则系统无法找到之前保存的会话状态，流程会直接出错。

# 第一次调用
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config)

# 恢复（使用相同的配置）
result = agent.invoke(Command(resume={...}), config=config)

• 第一次调用时，创建一个带有固定 thread_id（这里是 my-thread）的配置对象。

• 代理执行时，会把会话状态保存到 checkpointer 中，并和这个 thread_id 绑定。

• 人工审核完成后恢复执行时，必须传入和第一次调用完全相同的 config 对象（包含同一个 thread_id）。
• 系统会根据这个 thread_id 从 checkpointer 中加载之前的状态，从暂停的位置继续执行。

3.确保决策顺序雨操作请求一致

当一次中断里有多个待审核操作时，你提供的 decisions 列表，必须和 action_requests 列表里的操作顺序完全一一对应，不能乱序，也不能少给 / 多给决策。

if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]
    
    # 为每个操作创建一个决策，按顺序
    decisions = []
    for action in action_requests:
        decision = get_user_decision(action)  # 你的逻辑：让用户为每个操作选择approve/reject/edit
        decisions.append(decision)
    
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

• 从中断信息中提取出所有待审核的操作列表 action_requests，它的顺序是 AI 生成工具调用的顺序。
• 遍历 action_requests，按原顺序为每个操作收集用户的决策，存入 decisions 列表。
• 这样可以确保 decisions[i] 对应 action_requests[i]，顺序不会出错。
• 把按顺序收集好的 decisions 传给 Command，恢复执行。
• 系统会根据每个决策，对对应的操作执行批准 / 拒绝 / 修改。

关键注意点

• 顺序必须严格匹配：如果 decisions 的顺序和 action_requests 不一致，会导致决策被错误地应用到其他操作上，引发逻辑错误。
• 每个操作都要有决策：即使你只想处理其中一个操作，也必须为 action_requests 里的所有操作提供决策，不能只给部分。

4.根据风险级别定制审核策略

人工介入不是 “一刀切” 的，不同工具操作的风险等级不同，需要配置不同的审核策略：

• 高风险操作：影响大、不可逆，需要人工完全控制，允许批准、修改、拒绝。
• 中等风险操作：有影响但可追溯，不允许修改，只能批准或拒绝。
• 低风险操作：无实质性影响，不需要审核，直接自动执行。

interrupt_on = {
    # 高风险：完全控制（approve, edit, reject）
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
    "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},

    # 中等风险：不允许编辑
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # 低风险：无中断
    "read_file": False,
    "list_files": False,
}

1. 高风险操作（delete_file / send_email）

• 配置：allowed_decisions: ["approve", "edit", "reject"]
• 含义：人工审核时，拥有完全控制权：
  • approve：直接批准执行
  • edit：修改参数后再执行（比如修改删除的文件、邮件的收件人 / 内容）
  • reject：直接拒绝执行
• 适用场景：
  • delete_file：文件删除是不可逆操作，一旦出错无法恢复，需要人工把关并允许修改错误的删除请求。
  • send_email：邮件发送后无法撤回，可能涉及敏感信息，允许人工修改收件人、内容后再发送。

1. 中等风险操作（write_file）

• 配置：allowed_decisions: ["approve", "reject"]
• 含义：只允许人工批准或拒绝，不允许修改参数。
• 适用场景：
  • write_file：写入文件的内容通常是 AI 生成的报告、日志，人工不应该随意篡改内容，只需要确认是否允许写入即可。
  • 这种配置既保证了安全，又避免了人工修改带来的内容混乱。

1. 低风险操作（read_file / list_files）

• 配置：False
• 含义：禁用中断，不需要人工审核，直接自动执行。
• 适用场景：
  • read_file：读取文件只是查看内容，不会修改数据，没有风险。
  • list_files：列出文件目录，也不会产生任何影响。
  • 直接放行可以大幅提升执行效率，避免无意义的人工审核。