🤝 DeepAgents 人工介入

"""
interrupt_on 基础配置
"""
from deepagents import create_deep_agent

# 方式 1：布尔值（推荐用于简单场景）
agent = create_deep_agent(
    model="gpt-4o",
    tools=[delete_file, read_file, send_email],
    interrupt_on={
        "delete_file": True,   # 启用中断
        "read_file": False,    # 禁用中断
        "send_email": True,    # 启用中断
    }
)

# 方式 2：字典形式（高级控制）
agent = create_deep_agent(
    model="gpt-4o",
    tools=[delete_file, send_email, modify_database],
    interrupt_on={
        # 允许所有三种决策
        "delete_file": {
            "allowed_decisions": ["approve", "edit", "reject"]
        },

        # 仅允许批准或拒绝，不允许编辑
        "send_email": {
            "allowed_decisions": ["approve", "reject"]
        },

        # 仅允许批准（强制执行，但需要确认）
        "modify_database": {
            "allowed_decisions": ["approve"]
        },
    }
)

"""
完整的 Human-in-the-Loop 工作流
"""
import os
import uuid
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

# 步骤 1: 定义需要审批的工具
@tool
def delete_file(path: str) -> str:
    """删除文件系统中的文件"""
    # 实际会执行删除操作
    import os
    os.remove(path)
    return f"已删除文件：{path}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件"""
    # 实际会调用邮件服务
    return f"已发送邮件到 {to}"

# 步骤 2: 创建 Checkpointer（必需！）
checkpointer = MemorySaver()

# 步骤 3: 配置 Agent 的 interrupt_on
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[delete_file, send_email],
    interrupt_on={
        "delete_file": True,    # 需要审批
        "send_email": True,     # 需要审批
    },
    checkpointer=checkpointer,  # 必需：状态持久化
    system_prompt="你是文件管理助手，可以删除文件和发送通知邮件。"
)

# 步骤 4: 配置线程 ID（用于状态管理）
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 步骤 5: 调用 Agent
print("=== 步骤 1: 调用 Agent ===")
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "删除临时文件 /tmp/cache.txt 并发邮件通知管理员"
    }]
}, config=config)

# 步骤 6: 检查是否发生中断
if result.get("__interrupt__"):
    print("\n=== 步骤 2: 检测到中断 ===")

    # 提取中断信息
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]
    review_configs = interrupts["review_configs"]

    print(f"需要审批的操作数量: {len(action_requests)}")

    # 步骤 7: 展示待审批操作
    print("\n=== 步骤 3: 待审批操作 ===")
    for i, action in enumerate(action_requests, 1):
        print(f"\n操作 {i}:")
        print(f"  工具名称: {action['name']}")
        print(f"  工具参数: {action['args']}")

    # 步骤 8: 人工决策（这里模拟用户输入）
    print("\n=== 步骤 4: 人工决策 ===")
    decisions = []

    for action in action_requests:
        if action['name'] == 'delete_file':
            # 批准删除操作
            decisions.append({"type": "approve"})
            print(f"✅ 批准执行: {action['name']}")

        elif action['name'] == 'send_email':
            # 编辑邮件地址
            decisions.append({
                "type": "edit",
                "edited_action": {
                    "name": "send_email",
                    "args": {
                        "to": "[email protected]",  # 修改收件人
                        "subject": action['args']['subject'],
                        "body": action['args']['body']
                    }
                }
            })
            print(f"✏️ 编辑后执行: {action['name']}")

    # 步骤 9: 恢复执行（使用相同的 config！）
    print("\n=== 步骤 5: 恢复执行 ===")
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config  # ⚠️ 必须使用相同的 thread_id
    )

    print("\n=== 步骤 6: 执行完成 ===")
    print(f"最终结果:\n{result['messages'][-1].content}")

else:
    print("无需审批，直接执行完成")
    print(f"结果:\n{result['messages'][-1].content}")

# 输出示例：
# === 步骤 1: 调用 Agent ===
# === 步骤 2: 检测到中断 ===
# 需要审批的操作数量: 2
#
# === 步骤 3: 待审批操作 ===
# 操作 1:
#   工具名称: delete_file
#   工具参数: {'path': '/tmp/cache.txt'}
#
# 操作 2:
#   工具名称: send_email
#   工具参数: {'to': '[email protected]', 'subject': '...', 'body': '...'}
#
# === 步骤 4: 人工决策 ===
# ✅ 批准执行: delete_file
# ✏️ 编辑后执行: send_email
#
# === 步骤 5: 恢复执行 ===
# === 步骤 6: 执行完成 ===
# 最终结果:
# 已成功删除文件并发送通知邮件。

"""
决策类型 1: Approve
"""
# 批准执行，使用原始参数
decisions = [{"type": "approve"}]

result = agent.invoke(
    Command(resume={"decisions": decisions}),
    config=config
)

# Agent 会使用原始参数执行工具：
# delete_file(path="/tmp/cache.txt")

"""
决策类型 2: Edit
"""
# 编辑参数后执行
decisions = [{
    "type": "edit",
    "edited_action": {
        "name": "send_email",  # 必须指定工具名称
        "args": {               # 完整的新参数
            "to": "[email protected]",  # 修改后的收件人
            "subject": "System Alert",
            "body": "File deleted successfully."
        }
    }
}]

result = agent.invoke(
    Command(resume={"decisions": decisions}),
    config=config
)

# Agent 会使用修改后的参数执行工具：
# send_email(to="[email protected]", ...)

"""
决策类型 3: Reject
"""
# 拒绝执行，跳过工具调用
decisions = [{"type": "reject"}]

result = agent.invoke(
    Command(resume={"decisions": decisions}),
    config=config
)

# Agent 会跳过工具执行，不调用 delete_file()
# Agent 会知道操作被拒绝，可以继续执行其他任务

"""
多工具中断处理
"""
import uuid
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

@tool
def delete_file(path: str) -> str:
    """删除文件"""
    return f"已删除：{path}"

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

@tool
def backup_database() -> str:
    """备份数据库"""
    return "数据库已备份"

# 配置 Agent
checkpointer = MemorySaver()
agent = create_deep_agent(
    model="gpt-4o",
    tools=[delete_file, send_email, backup_database],
    interrupt_on={
        "delete_file": True,
        "send_email": True,
        "backup_database": True,
    },
    checkpointer=checkpointer
)

config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 调用 Agent（需要多个工具）
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "删除旧日志文件、备份数据库、然后发邮件通知"
    }]
}, config=config)

# 检查中断
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]

    print(f"需要审批 {len(action_requests)} 个操作\n")

    # 展示所有待审批操作
    for i, action in enumerate(action_requests, 1):
        print(f"操作 {i}: {action['name']}")
        print(f"  参数: {action['args']}\n")

    # ⚠️ 关键：决策顺序必须与 action_requests 顺序一致
    decisions = [
        {"type": "approve"},     # 第 1 个工具：delete_file
        {"type": "reject"},      # 第 2 个工具：backup_database（拒绝）
        {"type": "approve"},     # 第 3 个工具：send_email
    ]

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

    print("执行结果:")
    print(result["messages"][-1].content)

# 输出示例：
# 需要审批 3 个操作
#
# 操作 1: delete_file
#   参数: {'path': '/var/log/old.log'}
#
# 操作 2: backup_database
#   参数: {}
#
# 操作 3: send_email
#   参数: {'to': '[email protected]', 'subject': 'Maintenance Complete'}
#
# 执行结果:
# 已删除旧日志文件，跳过了数据库备份，并发送了通知邮件。

"""
子代理中断配置 - 分层控制
"""
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

@tool
def delete_file(path: str) -> str:
    """删除文件"""
    return f"已删除：{path}"

@tool
def read_file(path: str) -> str:
    """读取文件"""
    with open(path, 'r') as f:
        return f.read()

# 创建 Agent
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="gpt-4o",
    tools=[delete_file, read_file],

    # 主 Agent 的中断配置
    interrupt_on={
        "delete_file": True,    # 主 Agent 中需要审批
        "read_file": False,     # 主 Agent 中无需审批
    },

    # 子代理配置
    subagents=[{
        "name": "file-manager",
        "description": "管理文件的专门代理",
        "tools": [delete_file, read_file],
        "system_prompt": "你是文件管理专家。",

        # 子代理的中断配置（覆盖主 Agent）
        "interrupt_on": {
            "delete_file": True,     # 子代理中也需要审批
            "read_file": True,       # ⚠️ 子代理中需要审批（与主 Agent 不同）
        }
    }],

    checkpointer=checkpointer
)

# 使用场景：
# 1. 主 Agent 直接调用 delete_file → 需要审批
# 2. 主 Agent 直接调用 read_file → 无需审批
# 3. 主 Agent 委派给 file-manager 子代理
#    → 子代理调用 delete_file → 需要审批
#    → 子代理调用 read_file → 需要审批（不同于主 Agent）

# 为什么子代理配置不同？
# - 子代理可能在更敏感的上下文中操作
# - 子代理可能处理更关键的数据
# - 不同的子代理有不同的审批策略

"""
风险等级配置策略
"""
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

# 定义不同风险等级的工具
@tool
def delete_production_data(table: str) -> str:
    """删除生产环境数据（高风险）"""
    return f"已删除生产表：{table}"

@tool
def send_marketing_email(recipients: list, content: str) -> str:
    """发送营销邮件（中等风险）"""
    return f"已发送邮件给 {len(recipients)} 个收件人"

@tool
def read_log_file(path: str) -> str:
    """读取日志文件（低风险）"""
    return f"日志内容：..."

@tool
def update_cache(key: str, value: str) -> str:
    """更新缓存（无风险）"""
    return f"缓存已更新：{key}"

# 风险等级配置
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[
        delete_production_data,
        send_marketing_email,
        read_log_file,
        update_cache
    ],
    interrupt_on={
        # 🔴 高风险：完全控制（批准、编辑、拒绝）
        "delete_production_data": {
            "allowed_decisions": ["approve", "edit", "reject"]
        },

        # 🟡 中等风险：仅批准/拒绝，不允许编辑
        "send_marketing_email": {
            "allowed_decisions": ["approve", "reject"]
        },

        # 🟢 低风险：仅需确认（只能批准）
        "read_log_file": {
            "allowed_decisions": ["approve"]
        },

        # ⚪ 无风险：无需审批
        "update_cache": False,
    },
    checkpointer=checkpointer,
    system_prompt="""你是系统管理助手。执行操作时需要遵循严格的审批策略：

- 删除生产数据：高风险，必须人工审批
- 发送营销邮件：中等风险，需要确认
- 读取日志文件：低风险，仅需确认
- 更新缓存：无风险，自动执行
"""
)

# 配置表格
print("风险等级配置策略：\n")
print("| 工具                    | 风险等级 | 允许的决策              | 说明")
print("|------------------------|---------|------------------------|--------")
print("| delete_production_data | 🔴 高   | approve, edit, reject  | 完全控制")
print("| send_marketing_email   | 🟡 中   | approve, reject        | 仅批准/拒绝")
print("| read_log_file          | 🟢 低   | approve                | 仅需确认")
print("| update_cache           | ⚪ 无   | -                      | 自动执行")

# ❌ 错误：没有 Checkpointer
agent = create_deep_agent(
    interrupt_on={"delete_file": True}
    # 缺少 checkpointer 参数
)
# 会导致运行时错误！

# ✅ 正确：包含 Checkpointer
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    interrupt_on={"delete_file": True},
    checkpointer=checkpointer  # 必需！
)

# ❌ 错误：使用不同的 thread_id
config1 = {"configurable": {"thread_id": "thread-1"}}
result = agent.invoke({...}, config=config1)

# 中断后...
config2 = {"configurable": {"thread_id": "thread-2"}}  # 不同的 ID！
result = agent.invoke(Command(resume={...}), config=config2)  # 会失败

# ✅ 正确：使用相同的 thread_id
config = {"configurable": {"thread_id": "thread-1"}}
result = agent.invoke({...}, config=config)

# 中断后...
result = agent.invoke(Command(resume={...}), config=config)  # 相同的 config

# ❌ 错误：参数不完整
decisions = [{
    "type": "edit",
    "edited_action": {
        "name": "send_email",
        "args": {"to": "[email protected]"}  # 缺少 subject 和 body
    }
}]

# ✅ 正确：完整参数
decisions = [{
    "type": "edit",
    "edited_action": {
        "name": "send_email",
        "args": {
            "to": "[email protected]",
            "subject": "Updated Subject",
            "body": "Updated content"
        }
    }
}]

# 使用持久化 Checkpointer
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver(connection_string="postgresql://...")

agent = create_deep_agent(
    interrupt_on={"delete_file": True},
    checkpointer=checkpointer
)

# 进程 1：触发中断
config = {"configurable": {"thread_id": "user-123-task-456"}}
result = agent.invoke({...}, config=config)

# ... 进程退出 ...

# 进程 2：恢复执行（可能在不同服务器）
config = {"configurable": {"thread_id": "user-123-task-456"}}  # 相同 ID
result = agent.invoke(Command(resume={...}), config=config)  # 成功恢复

# 拒绝所有操作
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]

    # 全部拒绝
    decisions = [{"type": "reject"} for _ in action_requests]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

    # Agent 会知道所有操作被拒绝，可以生成相应的消息

def create_user_agent(user_role: str):
    """根据用户角色创建 Agent"""

    if user_role == "admin":
        # 管理员：无需审批
        interrupt_on = {}
    elif user_role == "operator":
        # 操作员：仅高风险操作需审批
        interrupt_on = {"delete_file": True}
    else:
        # 普通用户：所有敏感操作需审批
        interrupt_on = {
            "delete_file": True,
            "send_email": True,
            "modify_database": True
        }

    return create_deep_agent(
        interrupt_on=interrupt_on,
        checkpointer=checkpointer
    )

# 使用
admin_agent = create_user_agent("admin")
user_agent = create_user_agent("user")

import time

# 触发中断
result = agent.invoke({...}, config=config)

if result.get("__interrupt__"):
    # 记录中断时间
    interrupt_time = time.time()
    timeout_seconds = 300  # 5 分钟超时

    # 等待用户决策（应该在异步环境中实现）
    # ... 等待用户输入 ...

    # 检查是否超时
    if time.time() - interrupt_time > timeout_seconds:
        # 超时，自动拒绝
        decisions = [{"type": "reject"} for _ in action_requests]
    else:
        # 使用用户决策
        decisions = user_decisions

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )