用户请求 → Agent 规划 → [暂停：展示计划给人类] → 人类批准 → 工具执行 → 返回结果

用户请求 → Agent 执行 → 工具调用完成 → [暂停：展示结果给人类] → 人类审核 → 对外发布

用户请求 → Agent 执行 → 工具连续失败 → [暂停：报告困境] → 人类提供指引 → 继续执行

from typing import Annotated, TypedDict
import json
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

# 定义 Agent 状态
class AgentState(TypedDict):
    messages: list
    pending_tool_call: dict | None  # 待确认的工具调用
    user_approved: bool | None       # 人类的审批结果

# 工具定义（危险操作）
def delete_database_records(table: str, condition: str) -> str:
    """实际执行数据库删除（此处为模拟）"""
    return f"Deleted records from {table} WHERE {condition}"

def send_notification_email(to: str, subject: str, body: str) -> str:
    """发送邮件（此处为模拟）"""
    return f"Email sent to {to}"

DANGEROUS_TOOLS = {"delete_database_records", "send_notification_email"}

llm = ChatOpenAI(model="gpt-4o")
tools = [delete_database_records, send_notification_email]
llm_with_tools = llm.bind_tools(tools)


# Node 1：LLM 推理节点
def agent_node(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": state["messages"] + [response], "pending_tool_call": None}


# Node 2：工具执行前的审批节点（interrupt_before 模式）
def approval_gate(state: AgentState) -> AgentState:
    """
    检查最新 AI 消息是否包含危险工具调用。
    如果包含，暂停并等待人类审批。
    """
    last_message = state["messages"][-1]

    if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
        return state

    for tool_call in last_message.tool_calls:
        if tool_call["name"] in DANGEROUS_TOOLS:
            # 关键：调用 interrupt()，Graph 执行暂停
            # interrupt() 的参数会传递给等待审批的客户端
            human_decision = interrupt({
                "type": "approval_required",
                "tool_name": tool_call["name"],
                "tool_args": tool_call["args"],
                "message": (
                    f"Agent 计划执行高风险操作：\n"
                    f"工具：{tool_call['name']}\n"
                    f"参数：{json.dumps(tool_call['args'], ensure_ascii=False, indent=2)}\n"
                    f"请批准（approve）或拒绝（reject）。"
                ),
            })

            # human_decision 是恢复时传入的值
            if human_decision != "approve":
                # 拒绝：插入一条工具结果消息，告知 LLM 操作被取消
                rejection_msg = ToolMessage(
                    content=f"Operation cancelled by human reviewer. Reason: {human_decision}",
                    tool_call_id=tool_call["id"],
                )
                return {
                    "messages": state["messages"] + [rejection_msg],
                    "user_approved": False,
                }

    return {"user_approved": True, **state}


# Node 3：工具执行节点
def tool_executor(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    new_messages = []

    if hasattr(last_message, "tool_calls"):
        for tool_call in last_message.tool_calls:
            # 根据工具名称分发执行
            tool_map = {
                "delete_database_records": delete_database_records,
                "send_notification_email": send_notification_email,
            }
            tool_fn = tool_map.get(tool_call["name"])
            if tool_fn:
                result = tool_fn(**tool_call["args"])
            else:
                result = f"Unknown tool: {tool_call['name']}"

            new_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"],
            ))

    return {"messages": state["messages"] + new_messages}


# 路由函数：判断是否有工具调用
def should_use_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "approval_gate"
    return END


# 构建 Graph
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("approval_gate", approval_gate)
builder.add_node("tool_executor", tool_executor)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_use_tools)
builder.add_edge("approval_gate", "tool_executor")
builder.add_edge("tool_executor", "agent")

# 关键：必须使用 Checkpointer 才能支持 interrupt/resume
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

import uuid

def run_with_approval(user_request: str) -> None:
    """演示完整的 interrupt → 人工审批 → resume 流程"""

    # 每次对话使用唯一的 thread_id，保证状态隔离
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    print(f"=== 开始执行，Thread: {thread_id[:8]} ===")

    # 第一次运行：可能在中断点暂停
    for chunk in graph.stream(
        {"messages": [HumanMessage(content=user_request)]},
        config=config,
        stream_mode="values",
    ):
        # 检查是否有中断
        if "__interrupt__" in chunk:
            interrupt_data = chunk["__interrupt__"][0].value
            print("\n[!] Agent 请求人工审批：")
            print(interrupt_data["message"])

            # 模拟人类审批决策（实际场景通过 API/UI 获取）
            decision = input("\n请输入决策（approve / reject原因）：").strip()

            # 恢复执行：通过 Command 传入审批结果
            print("\n=== 恢复执行 ===")
            for resume_chunk in graph.stream(
                Command(resume=decision),
                config=config,  # 必须使用相同的 thread_id
                stream_mode="values",
            ):
                if "messages" in resume_chunk:
                    last = resume_chunk["messages"][-1]
                    if hasattr(last, "content") and isinstance(last.content, str):
                        print(f"Agent: {last.content}")
            return

    # 无中断，直接完成
    state = graph.get_state(config)
    final_message = state.values["messages"][-1]
    print(f"Agent: {final_message.content}")


# 测试：触发审批流程
run_with_approval("删除 users 表中 status='inactive' 的所有记录，这些账号已超过3年未登录")

from enum import Enum

class InterruptReason(str, Enum):
    APPROVAL_REQUIRED = "approval_required"
    CLARIFICATION_NEEDED = "clarification_needed"
    ERROR_RECOVERY = "error_recovery"

def clarification_node(state: AgentState) -> AgentState:
    """
    当 Agent 检测到任务描述存在关键歧义时，主动请求澄清。
    与 approval_gate 不同，这里是 Agent 主动发起中断，而非系统检测危险操作。
    """
    last_message = state["messages"][-1]

    # 检查 LLM 是否在回复中表达了需要澄清的意图
    # 实际实现可通过结构化输出判断
    if hasattr(last_message, "content") and "NEED_CLARIFICATION:" in last_message.content:
        question = last_message.content.split("NEED_CLARIFICATION:")[1].strip()

        user_response = interrupt({
            "type": InterruptReason.CLARIFICATION_NEEDED,
            "question": question,
            "context": "Agent needs additional information to proceed safely.",
        })

        # 将用户回答注入消息历史，让 Agent 继续推理
        return {
            "messages": state["messages"] + [
                HumanMessage(content=f"Clarification: {user_response}")
            ]
        }

    return state

SYSTEM_PROMPT = """
You are a helpful assistant with access to powerful tools.

IMPORTANT RULES:
1. For ambiguous requests, output "NEED_CLARIFICATION: <your question>" before using tools.
2. Never assume the scope of destructive operations—always clarify the exact target.
3. When you identify a dangerous operation, describe exactly what will happen before proceeding.

Examples of when to ask for clarification:
- "Delete old records" → Ask: which table? What defines 'old'? Is there a backup?
- "Send email to the team" → Ask: which team? What address? Is this urgent?
"""

from dataclasses import dataclass
from enum import Enum

class UserRole(str, Enum):
    VIEWER = "viewer"       # 只读
    EDITOR = "editor"       # 可写，不可删
    ADMIN = "admin"         # 全权限，高危操作仍需确认
    SUPER_ADMIN = "super"   # 全权限，无需确认

@dataclass
class PermissionPolicy:
    """权限策略：定义哪些操作需要确认"""
    auto_execute: set[str]       # 可自动执行的操作
    confirm_required: set[str]   # 需要确认的操作
    forbidden: set[str]          # 禁止执行的操作

PERMISSION_POLICIES: dict[UserRole, PermissionPolicy] = {
    UserRole.VIEWER: PermissionPolicy(
        auto_execute={"read_file", "search_database", "get_user_profile"},
        confirm_required=set(),
        forbidden={"delete_records", "send_email", "deploy", "execute_sql"},
    ),
    UserRole.EDITOR: PermissionPolicy(
        auto_execute={"read_file", "search_database", "write_file", "create_record"},
        confirm_required={"send_email", "update_record"},
        forbidden={"delete_records", "deploy"},
    ),
    UserRole.ADMIN: PermissionPolicy(
        auto_execute={"read_file", "search_database", "write_file", "create_record", "update_record"},
        confirm_required={"send_email", "delete_records", "deploy"},
        forbidden=set(),
    ),
    UserRole.SUPER_ADMIN: PermissionPolicy(
        auto_execute={"read_file", "search_database", "write_file", "create_record",
                      "update_record", "send_email", "delete_records", "deploy"},
        confirm_required=set(),
        forbidden=set(),
    ),
}

def check_permission(
    tool_name: str,
    user_role: UserRole,
) -> tuple[bool, bool]:
    """
    返回：(is_allowed, requires_confirmation)
    """
    policy = PERMISSION_POLICIES[user_role]
    if tool_name in policy.forbidden:
        return False, False
    if tool_name in policy.confirm_required:
        return True, True
    return True, False  # auto_execute 或未明确列出的工具默认允许

# FastAPI 后端示例
from fastapi import FastAPI
from pydantic import BaseModel
import asyncio

app = FastAPI()
pending_approvals: dict[str, dict] = {}  # thread_id → 中断数据

@app.post("/agent/run")
async def start_agent(request: dict):
    thread_id = str(uuid.uuid4())
    # 异步启动 Agent（可能中断）
    asyncio.create_task(run_agent_async(request["query"], thread_id))
    return {"thread_id": thread_id, "status": "running"}

@app.get("/agent/{thread_id}/status")
async def get_status(thread_id: str):
    """前端轮询此接口，检查是否有待审批项"""
    if thread_id in pending_approvals:
        return {
            "status": "waiting_approval",
            "approval_data": pending_approvals[thread_id],
        }
    # 检查是否完成...
    return {"status": "running"}

class ApprovalDecision(BaseModel):
    decision: str  # "approve" 或 拒绝原因

@app.post("/agent/{thread_id}/approve")
async def approve_action(thread_id: str, body: ApprovalDecision):
    """人类提交审批决策"""
    config = {"configurable": {"thread_id": thread_id}}
    # 恢复 LangGraph 执行
    graph.invoke(Command(resume=body.decision), config=config)
    del pending_approvals[thread_id]
    return {"status": "resumed"}

from fastapi.responses import StreamingResponse
import asyncio

@app.get("/agent/{thread_id}/stream")
async def stream_events(thread_id: str):
    """
    SSE 端点：Agent 状态变更实时推送给前端。
    中断时推送 approval_required 事件，前端弹出确认对话框。
    """
    async def event_generator():
        config = {"configurable": {"thread_id": thread_id}}

        async for event in graph.astream_events(
            None,  # 已有 thread_id，无需重新输入
            config=config,
            version="v2",
        ):
            if event["event"] == "on_chain_start":
                yield f"data: {json.dumps({'type': 'step_start', 'node': event['name']})}\n\n"
            elif event["event"] == "on_chain_end":
                yield f"data: {json.dumps({'type': 'step_end', 'node': event['name']})}\n\n"
            elif "__interrupt__" in event.get("data", {}):
                interrupt_data = event["data"]["__interrupt__"][0].value
                yield f"data: {json.dumps({'type': 'approval_required', **interrupt_data})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

import os
import shutil

# 文件操作工具（区分安全与危险）
def list_files(directory: str) -> list[str]:
    """列出目录下的文件（安全操作）"""
    return os.listdir(directory)

def read_file_content(path: str) -> str:
    """读取文件内容（安全操作）"""
    with open(path) as f:
        return f.read()

def write_file(path: str, content: str) -> str:
    """写入文件（中等风险，如文件已存在则覆盖）"""
    with open(path, "w") as f:
        f.write(content)
    return f"Written to {path}"

def delete_file(path: str) -> str:
    """删除文件（高风险，不可逆）"""
    os.remove(path)
    return f"Deleted {path}"

def delete_directory(path: str) -> str:
    """删除目录（极高风险，不可逆）"""
    shutil.rmtree(path)
    return f"Deleted directory {path}"

# 工具风险分级配置
TOOL_RISK_LEVELS = {
    "list_files": "safe",
    "read_file_content": "safe",
    "write_file": "medium",      # 覆盖时需提示
    "delete_file": "dangerous",  # 必须确认
    "delete_directory": "critical",  # 必须确认，加倍警示
}

def build_file_manager_agent() -> tuple:
    """构建文件管理 Agent，内置 HitL 审批"""

    class FileAgentState(TypedDict):
        messages: list
        working_directory: str

    def file_agent_node(state: FileAgentState):
        tools_schema = [
            # 此处定义工具 schema...
        ]
        llm = ChatOpenAI(model="gpt-4o")
        response = llm.invoke(state["messages"])
        return {"messages": state["messages"] + [response]}

    def file_approval_gate(state: FileAgentState):
        last = state["messages"][-1]
        if not hasattr(last, "tool_calls"):
            return state

        for tc in last.tool_calls:
            risk = TOOL_RISK_LEVELS.get(tc["name"], "safe")
            if risk in ("dangerous", "critical"):
                warning = "⚠️ 极高风险" if risk == "critical" else "高风险"
                decision = interrupt({
                    "type": "approval_required",
                    "risk_level": risk,
                    "tool_name": tc["name"],
                    "tool_args": tc["args"],
                    "warning": warning,
                    "message": (
                        f"{warning} 操作需要确认\n"
                        f"操作：{tc['name']}\n"
                        f"目标：{tc['args']}\n"
                        f"此操作不可撤销，请谨慎确认。"
                    ),
                })
                if decision != "approve":
                    return {
                        "messages": state["messages"] + [
                            ToolMessage(
                                content=f"Operation cancelled: {decision}",
                                tool_call_id=tc["id"],
                            )
                        ]
                    }
        return state

    return file_agent_node, file_approval_gate