👤 Human-in-the-Loop（人工介入）

"""
HumanInTheLoopMiddleware 基础配置
功能：为敏感工具添加人工审核
"""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool

@tool
def write_file(filename: str, content: str) -> str:
    """写入文件（敏感操作）"""
    with open(filename, "w") as f:
        f.write(content)
    return f"文件 {filename} 写入成功"

@tool
def read_file(filename: str) -> str:
    """读取文件（安全操作）"""
    with open(filename, "r") as f:
        return f.read()

@tool
def delete_file(filename: str) -> str:
    """删除文件（危险操作）"""
    import os
    os.remove(filename)
    return f"文件 {filename} 已删除"

# 创建带 HITL 的 Agent
agent = create_agent(
    model="gpt-4o",
    tools=[write_file, read_file, delete_file],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,      # 写文件需要审核
                "delete_file": True,     # 删除文件需要审核
                "read_file": False,      # 读文件无需审核
            },
            description_prefix="工具执行等待审批"
        )
    ],
    checkpointer=InMemorySaver(),  # 必需！
)

# 配置 thread_id（用于持久化）
config = {"configurable": {"thread_id": "user_001"}}

"""
步骤 1：运行 Agent 直到遇到需要审核的操作
"""
from langgraph.types import Command

config = {"configurable": {"thread_id": "user_001"}}

# 调用 Agent
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "删除 old_data.txt 文件"}
        ]
    },
    config=config
)

# 检查是否有中断
if "__interrupt__" in result:
    print("⏸️ Agent 已暂停，等待人工决策")
    interrupt_info = result["__interrupt__"]

    # 查看中断详情
    action_requests = interrupt_info[0]["value"]["action_requests"]
    for req in action_requests:
        print(f"\n工具: {req['name']}")
        print(f"参数: {req['args']}")
        print(f"描述: {req.get('description', '无')}")
else:
    # 无需审核，直接执行完成
    print("✅ 执行完成（无需审核）")
    print(result["messages"][-1].content)

"""
批准执行：同意 Agent 的提议
"""
from langgraph.types import Command

# 恢复执行并批准
result = agent.invoke(
    Command(
        resume={
            "decisions": [
                {"type": "approve"}  # ✅ 批准执行
            ]
        }
    ),
    config=config  # 使用相同的 thread_id
)

print("✅ 操作已执行")
print(result["messages"][-1].content)

"""
编辑执行：修改参数后再执行
"""
from langgraph.types import Command

# 假设 Agent 想删除 important_file.txt
# 我们修改为删除 backup_file.txt

result = agent.invoke(
    Command(
        resume={
            "decisions": [
                {
                    "type": "edit",  # ✏️ 编辑模式
                    "edited_action": {
                        "name": "delete_file",  # 工具名称（可选修改）
                        "args": {
                            "filename": "backup_file.txt"  # 修改参数
                        }
                    }
                }
            ]
        }
    ),
    config=config
)

print("✏️ 操作已修改并执行")
print(result["messages"][-1].content)

"""
拒绝执行：不同意操作并提供反馈
"""
from langgraph.types import Command

result = agent.invoke(
    Command(
        resume={
            "decisions": [
                {
                    "type": "reject",  # ❌ 拒绝
                    "message": "不能删除该文件，因为它包含重要数据。请先备份后再操作。"
                }
            ]
        }
    ),
    config=config
)

# 拒绝消息会被加入对话历史
# Agent 会根据反馈重新思考
print("❌ 操作已拒绝")
print(result["messages"][-1].content)

"""
多工具审批示例
功能：同时处理多个工具调用的审批
"""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langgraph.types import Command

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

@tool
def delete_records(table: str, condition: str) -> str:
    """删除数据库记录"""
    return f"已从 {table} 删除符合 {condition} 的记录"

@tool
def create_backup(source: str) -> str:
    """创建备份"""
    return f"已备份 {source}"

agent = create_agent(
    model="gpt-4o",
    tools=[send_email, delete_records, create_backup],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": True,
                "delete_records": True,
                "create_backup": True,
            }
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "multi_approval"}}

# Agent 可能同时调用多个工具
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "备份数据库，删除旧记录，并发邮件通知管理员"}
        ]
    },
    config=config
)

if "__interrupt__" in result:
    action_requests = result["__interrupt__"][0]["value"]["action_requests"]

    print(f"需要审批 {len(action_requests)} 个操作：\n")
    for i, req in enumerate(action_requests):
        print(f"{i+1}. {req['name']}: {req['args']}")

    # 为每个工具提供决策（顺序必须匹配）
    result = agent.invoke(
        Command(
            resume={
                "decisions": [
                    {"type": "approve"},  # 批准第1个工具
                    {
                        "type": "edit",    # 修改第2个工具
                        "edited_action": {
                            "name": "delete_records",
                            "args": {
                                "table": "logs",
                                "condition": "created_at < '2024-01-01'"  # 修改条件
                            }
                        }
                    },
                    {
                        "type": "reject",  # 拒绝第3个工具
                        "message": "请先确认备份成功后再发送邮件"
                    }
                ]
            }
        ),
        config=config
    )

print("\n决策已提交，Agent 继续执行")

"""
流式输出 + HITL 示例
功能：实时显示执行过程，遇到中断时暂停
"""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langgraph.types import Command

@tool
def execute_sql(query: str) -> str:
    """执行 SQL 查询"""
    return f"SQL 执行成功：{query}"

agent = create_agent(
    model="gpt-4o",
    tools=[execute_sql],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"execute_sql": True}
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "streaming_hitl"}}

print("开始执行...\n")

# 使用流式模式
interrupted = False

for mode, chunk in agent.stream(
    {
        "messages": [
            {"role": "user", "content": "删除 users 表中所有未激活的用户"}
        ]
    },
    config=config,
    stream_mode=["updates", "messages"]
):
    if mode == "messages":
        # 显示 Token 流
        token, metadata = chunk
        if token:
            print(token, end="", flush=True)

    elif mode == "updates":
        # 检查是否有中断
        for step, data in chunk.items():
            if "__interrupt__" in data:
                interrupted = True
                print("\n\n⏸️ 检测到需要审批的操作！")

                action_requests = data["__interrupt__"][0]["value"]["action_requests"]
                for req in action_requests:
                    print(f"\n工具: {req['name']}")
                    print(f"SQL: {req['args']['query']}")

                # 模拟人工决策
                print("\n等待人工审批...")
                break

if interrupted:
    # 继续执行（这里简化为自动批准）
    print("✅ 批准执行\n")

    for mode, chunk in agent.stream(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,
        stream_mode=["messages"]
    ):
        token, metadata = chunk
        if token:
            print(token, end="", flush=True)

print("\n\n执行完成")

"""
限制决策类型示例
功能：某些工具只允许批准或拒绝，不允许编辑
"""
from langchain.agents.middleware import HumanInTheLoopMiddleware

middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        # 写文件：允许所有决策类型
        "write_file": True,

        # 删除文件：只允许批准或拒绝，不允许编辑
        "delete_file": {
            "allowed_decisions": ["approve", "reject"]
        },

        # 发送邮件：只允许批准（必须执行或手动取消）
        "send_email": {
            "allowed_decisions": ["approve"]
        },
    }
)

"""
自定义描述文本示例
功能：为审批请求提供更详细的说明
"""
from langchain.agents.middleware import HumanInTheLoopMiddleware

def custom_description(action_request):
    """动态生成描述文本"""
    tool_name = action_request["name"]
    args = action_request["args"]

    if tool_name == "delete_file":
        filename = args.get("filename", "未知文件")
        return f"⚠️ 危险操作：即将删除文件 '{filename}'，此操作不可撤销！"

    elif tool_name == "send_email":
        to = args.get("to", "未知收件人")
        subject = args.get("subject", "无主题")
        return f"📧 邮件审批：发送给 {to}，主题：{subject}"

    else:
        return f"操作审批：{tool_name}"

middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "delete_file": True,
        "send_email": True,
    },
    description=custom_description  # 使用自定义函数
)

"""
敏感操作审批系统完整示例
功能：模拟真实的审批工作流
"""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langgraph.types import Command
import json

# ==================== 工具定义 ====================

@tool
def query_database(sql: str) -> str:
    """查询数据库（安全操作）"""
    return f"查询结果：[{'user1', 'user2', 'user3'}]"

@tool
def update_database(sql: str) -> str:
    """更新数据库（需审批）"""
    return f"数据库更新成功：{sql}"

@tool
def delete_from_database(sql: str) -> str:
    """删除数据库记录（危险操作）"""
    return f"删除成功：{sql}"

@tool
def send_notification(recipients: list[str], message: str) -> str:
    """发送通知（需审批）"""
    return f"通知已发送给 {len(recipients)} 人"

# ==================== 自定义描述 ====================

def approval_description(action_request):
    """生成审批描述"""
    tool_name = action_request["name"]
    args = action_request["args"]

    descriptions = {
        "update_database": f"🔸 数据库更新\nSQL: {args.get('sql', 'N/A')}",
        "delete_from_database": f"⚠️ 数据库删除（危险）\nSQL: {args.get('sql', 'N/A')}",
        "send_notification": f"📧 批量通知\n收件人: {len(args.get('recipients', []))} 人\n内容: {args.get('message', 'N/A')[:50]}...",
    }

    return descriptions.get(tool_name, f"操作: {tool_name}")

# ==================== 创建 Agent ====================

approval_agent = create_agent(
    model="gpt-4o",
    tools=[
        query_database,
        update_database,
        delete_from_database,
        send_notification
    ],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "query_database": False,           # 查询不需要审批
                "update_database": True,           # 更新需要审批
                "delete_from_database": {          # 删除只允许批准/拒绝
                    "allowed_decisions": ["approve", "reject"]
                },
                "send_notification": True,         # 发送通知需要审批
            },
            description=approval_description
        )
    ],
    checkpointer=InMemorySaver(),
    system_prompt="""你是数据库管理助手。

职责：
1. 查询数据库
2. 更新或删除记录（需要人工审批）
3. 发送通知

注意：
- 删除操作特别危险，需谨慎
- 发送通知前确认收件人和内容"""
)

# ==================== 审批工作流 ====================

def execute_with_approval(user_request: str, thread_id: str):
    """执行带审批的请求"""
    config = {"configurable": {"thread_id": thread_id}}

    print("\n" + "="*60)
    print(f"用户请求: {user_request}")
    print("="*60 + "\n")

    # 第一次调用：运行直到中断
    result = approval_agent.invoke(
        {"messages": [{"role": "user", "content": user_request}]},
        config=config
    )

    # 检查是否需要审批
    if "__interrupt__" not in result:
        print("✅ 执行完成（无需审批）")
        print(result["messages"][-1].content)
        return

    # 显示审批请求
    print("⏸️ 需要人工审批\n")
    action_requests = result["__interrupt__"][0]["value"]["action_requests"]

    decisions = []
    for i, req in enumerate(action_requests):
        print(f"审批请求 #{i+1}:")
        print(f"  工具: {req['name']}")
        print(f"  参数: {json.dumps(req['args'], ensure_ascii=False, indent=4)}")
        print(f"  说明: {req.get('description', '无')}")

        # 模拟人工决策（实际应用中这里是交互式输入）
        if req['name'] == "delete_from_database":
            # 删除操作：拒绝
            decision = {
                "type": "reject",
                "message": "删除操作过于危险，请先备份数据"
            }
            print("  决策: ❌ 拒绝\n")
        elif req['name'] == "send_notification":
            # 发送通知：批准
            decision = {"type": "approve"}
            print("  决策: ✅ 批准\n")
        else:
            # 其他操作：批准
            decision = {"type": "approve"}
            print("  决策: ✅ 批准\n")

        decisions.append(decision)

    # 提交决策并继续执行
    print("提交审批决策...\n")
    result = approval_agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

    print("✅ 执行完成")
    print(result["messages"][-1].content)

# ==================== 测试场景 ====================

if __name__ == "__main__":
    # 场景 1：查询操作（无需审批）
    execute_with_approval(
        "查询所有活跃用户",
        thread_id="scenario_1"
    )

    # 场景 2：更新操作（需要审批）
    execute_with_approval(
        "更新所有未激活用户的状态为已激活",
        thread_id="scenario_2"
    )

    # 场景 3：删除操作（会被拒绝）
    execute_with_approval(
        "删除所有测试数据，并发送通知给管理员",
        thread_id="scenario_3"
    )

# ❌ 错误：缺少 checkpointer
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[HumanInTheLoopMiddleware(...)]
    # 缺少 checkpointer！
)

# ✅ 正确
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[HumanInTheLoopMiddleware(...)],
    checkpointer=InMemorySaver()  # 必需
)

import time
from threading import Thread

def wait_for_approval_with_timeout(agent, config, timeout=300):
    """等待审批，超时自动拒绝"""
    decision = {"type": None}

    def wait_input():
        # 实际应用中这里是获取用户输入
        user_input = input("批准(y)/拒绝(n): ")
        decision["type"] = "approve" if user_input == "y" else "reject"

    thread = Thread(target=wait_input)
    thread.start()
    thread.join(timeout=timeout)

    if decision["type"] is None:
        print("⏰ 审批超时，自动拒绝")
        decision["type"] = "reject"
        decision["message"] = "审批超时"

    return decision

# 记录审批日志
import logging
from datetime import datetime

logger = logging.getLogger("approval_system")

def log_approval_decision(action_request, decision, user_id):
    """记录审批决策"""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "user_id": user_id,
        "tool": action_request["name"],
        "args": action_request["args"],
        "decision": decision["type"],
        "message": decision.get("message", ""),
    }

    logger.info(f"Approval: {log_entry}")

    # 存储到数据库
    # db.save_approval_log(log_entry)

# 生产环境推荐配置
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/langchain"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    agent = create_agent(
        model="gpt-4o",
        tools=[...],
        middleware=[
            HumanInTheLoopMiddleware(
                interrupt_on={...},
                description=custom_description_function,
            ),
        ],
        checkpointer=checkpointer,  # 生产级持久化
    )