场景 2: 敏感操作审批

模块：Interrupts（中断/人机协作）优先级：🟠 P1（中高）业务价值：提高系统安全性，防止误操作

一、业务背景

1.1 当前风险

当前项目允许用户执行以下敏感操作，但缺少安全确认机制：

具体风险：

• 批量删除会话 → 误操作导致重要对话丢失
• 注销账号 → 用户后悔无法恢复
• 导出数据 → 隐私泄露风险
• 修改关键设置 → 影响系统使用

1.2 期望效果

---

二、敏感操作识别

2.1 敏感操作分类

类别	操作	风险等级	审批方式
数据删除	删除单个会话	🟡 中	简单确认
数据删除	批量删除会话	🔴 高	详细确认
数据删除	清空所有数据	🔴 高	二次确认 + 输入验证
账号操作	注销账号	🔴 高	二次确认 + 延迟执行
数据导出	导出所有对话	🟡 中	简单确认
系统设置	修改关键配置	🟡 中	简单确认

2.2 识别流程

---

三、代码实现

3.1 审批工具定义

创建文件: services/tools/sensitive_tools.py

"""敏感操作相关工具

这些工具在执行前会触发 interrupt，等待用户确认。
"""
from langchain.tools import tool
from typing import Optional, Dict, Any, List
import logging

logger = logging.getLogger(__name__)


@tool
def delete_conversation(
    conversation_id: str,
    user_id: str,
    confirm: bool = False
) -> Dict[str, Any]:
    """
    删除指定的对话。

    ⚠️ 这是一个敏感操作，需要用户确认后才会执行。

    Args:
        conversation_id: 要删除的对话 ID
        user_id: 用户 ID
        confirm: 是否确认删除（由审批流程设置）

    Returns:
        操作结果信息
    """
    if not confirm:
        # 返回需要审批的信息
        return {
            "action": "delete_conversation",
            "requires_approval": True,
            "risk_level": "medium",
            "details": {
                "conversation_id": conversation_id,
                "message": "删除后对话将无法恢复"
            }
        }

    # 用户已确认，执行删除
    from core.database import get_db
    from models import Conversation, Message

    db = next(get_db())
    try:
        # 验证所有权
        conv = db.query(Conversation).filter_by(
            id=conversation_id,
            user_id=user_id
        ).first()

        if not conv:
            return {"success": False, "error": "对话不存在"}

        # 删除消息
        db.query(Message).filter_by(conversation_id=conversation_id).delete()
        # 删除对话
        db.delete(conv)
        db.commit()

        return {
            "success": True,
            "message": f"已删除对话「{conv.title}」"
        }
    except Exception as e:
        db.rollback()
        logger.error(f"删除对话失败: {e}")
        return {"success": False, "error": str(e)}


@tool
def delete_all_conversations(
    user_id: str,
    confirm: bool = False,
    confirmation_text: Optional[str] = None
) -> Dict[str, Any]:
    """
    删除用户所有对话。

    ⚠️ 这是一个高风险操作，需要用户输入确认文字。

    Args:
        user_id: 用户 ID
        confirm: 是否确认删除
        confirmation_text: 用户需要输入"确认删除所有对话"

    Returns:
        操作结果信息
    """
    if not confirm:
        return {
            "action": "delete_all_conversations",
            "requires_approval": True,
            "risk_level": "high",
            "details": {
                "message": "⚠️ 这将删除您的所有对话记录",
                "warning": "此操作不可恢复！",
                "confirmation_required": "请输入「确认删除所有对话」以继续"
            }
        }

    # 验证确认文字
    if confirmation_text != "确认删除所有对话":
        return {
            "success": False,
            "error": "确认文字不正确，操作已取消"
        }

    # 执行删除
    from core.database import get_db
    from models import Conversation, Message
    from sqlalchemy import delete

    db = next(get_db())
    try:
        # 获取对话 ID 列表
        conv_ids = [c.id for c in db.query(Conversation.id).filter_by(user_id=user_id).all()]

        # 删除消息
        db.execute(delete(Message).where(Message.conversation_id.in_(conv_ids)))
        # 删除对话
        db.execute(delete(Conversation).where(Conversation.user_id == user_id))
        db.commit()

        return {
            "success": True,
            "message": f"已删除 {len(conv_ids)} 个对话"
        }
    except Exception as e:
        db.rollback()
        logger.error(f"批量删除失败: {e}")
        return {"success": False, "error": str(e)}


@tool
def export_conversations(
    user_id: str,
    conversation_ids: Optional[List[str]] = None,
    export_all: bool = False,
    confirm: bool = False
) -> Dict[str, Any]:
    """
    导出对话数据。

    Args:
        user_id: 用户 ID
        conversation_ids: 要导出的对话 ID 列表
        export_all: 是否导出全部
        confirm: 是否确认导出

    Returns:
        导出结果（包含下载链接）
    """
    if export_all and not confirm:
        return {
            "action": "export_all_conversations",
            "requires_approval": True,
            "risk_level": "medium",
            "details": {
                "message": "将导出您的所有对话数据",
                "warning": "导出文件包含敏感信息，请妥善保管"
            }
        }

    # 执行导出
    from core.database import get_db
    from models import Conversation, Message
    import json
    from datetime import datetime

    db = next(get_db())
    try:
        query = db.query(Conversation).filter_by(user_id=user_id)
        if conversation_ids:
            query = query.filter(Conversation.id.in_(conversation_ids))

        conversations = query.all()
        export_data = []

        for conv in conversations:
            messages = db.query(Message).filter_by(conversation_id=conv.id).order_by(Message.id).all()
            export_data.append({
                "id": conv.id,
                "title": conv.title,
                "created_at": conv.created_at.isoformat(),
                "messages": [
                    {
                        "role": msg.role,
                        "content": msg.content,
                        "created_at": msg.created_at.isoformat()
                    }
                    for msg in messages
                ]
            })

        # 生成导出文件
        filename = f"conversations_{user_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        # TODO: 保存到 COS 并返回下载链接

        return {
            "success": True,
            "message": f"已导出 {len(export_data)} 个对话",
            "download_url": f"/api/download/{filename}",
            "conversation_count": len(export_data)
        }
    except Exception as e:
        logger.error(f"导出失败: {e}")
        return {"success": False, "error": str(e)}

3.2 审批 Agent

修改 services/langgraph_approval.py（在集成方案中定义的基础上，添加实际业务逻辑）：

"""审批 Agent - 敏感操作人机协作

针对项目实际业务需求实现。
"""
from typing import Literal, Optional, Dict, Any
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import interrupt, Command
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import os
import logging

from services.tools.sensitive_tools import (
    delete_conversation,
    delete_all_conversations,
    export_conversations,
)

logger = logging.getLogger(__name__)


class ApprovalState(MessagesState):
    """审批状态"""
    user_id: str
    pending_action: Optional[Dict[str, Any]] = None
    approved: Optional[bool] = None
    action_result: Optional[str] = None


class ApprovalAgent:
    """敏感操作审批 Agent"""

    SENSITIVE_TOOLS = [delete_conversation, delete_all_conversations, export_conversations]

    def __init__(self):
        self.tools_by_name = {t.name: t for t in self.SENSITIVE_TOOLS}

    def _should_interrupt(self, tool_name: str, tool_args: dict) -> bool:
        """判断是否需要中断"""
        # 所有敏感工具都需要中断
        return tool_name in self.tools_by_name

    def _approval_node(self, state: ApprovalState) -> Command[Literal["execute", "cancel"]]:
        """审批节点 - 中断等待用户确认"""
        last_message = state["messages"][-1]
        tool_calls = last_message.tool_calls

        if not tool_calls:
            return Command(goto=END)

        tool_call = tool_calls[0]
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # 获取工具的风险信息
        tool_result = self.tools_by_name[tool_name].invoke(tool_args)

        if not tool_result.get("requires_approval"):
            # 不需要审批，直接执行
            return Command(goto="execute")

        # 触发中断
        decision = interrupt({
            "action": tool_name,
            "risk_level": tool_result.get("risk_level", "medium"),
            "details": tool_result.get("details", {}),
            "confirmation_required": tool_result.get("details", {}).get("confirmation_required")
        })

        if decision:
            # 用户批准
            return Command(
                goto="execute",
                update={"approved": True, "pending_action": tool_call}
            )
        else:
            # 用户拒绝
            return Command(
                goto="cancel",
                update={"approved": False, "pending_action": tool_call}
            )

    def _execute_node(self, state: ApprovalState) -> ApprovalState:
        """执行节点 - 执行已批准的操作"""
        tool_call = state.get("pending_action")
        if not tool_call:
            return {"action_result": "没有待执行的操作"}

        # 添加确认标记
        args = tool_call["args"].copy()
        args["confirm"] = True

        # 执行工具
        result = self.tools_by_name[tool_call["name"]].invoke(args)

        return {
            "action_result": result.get("message", "操作完成"),
            "approved": None,
            "pending_action": None
        }

    def _cancel_node(self, state: ApprovalState) -> ApprovalState:
        """取消节点"""
        return {
            "action_result": "操作已取消",
            "approved": None,
            "pending_action": None
        }

    def build_graph(self, checkpointer):
        """构建审批工作流"""
        llm = ChatOpenAI(
            model=os.getenv("OPENROUTER_MODEL", "openai/gpt-4o-mini"),
            api_key=os.getenv("OPENROUTER_API_KEY"),
            base_url=os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        ).bind_tools(self.SENSITIVE_TOOLS)

        def agent_node(state: ApprovalState):
            system_prompt = """你是一个智能助手。当用户要求执行以下操作时，使用相应的工具：

- 删除对话 → 使用 delete_conversation
- 删除所有对话 → 使用 delete_all_conversations
- 导出对话 → 使用 export_conversations

对于敏感操作，系统会自动请求用户确认。"""
            messages = [SystemMessage(content=system_prompt)] + state["messages"]
            response = llm.invoke(messages)
            return {"messages": [response]}

        def should_approve(state: ApprovalState) -> Literal["approval", END]:
            last_message = state["messages"][-1]
            if last_message.tool_calls:
                tool_name = last_message.tool_calls[0]["name"]
                if tool_name in self.tools_by_name:
                    return "approval"
            return END

        workflow = StateGraph(ApprovalState)
        workflow.add_node("agent", agent_node)
        workflow.add_node("approval", self._approval_node)
        workflow.add_node("execute", self._execute_node)
        workflow.add_node("cancel", self._cancel_node)

        workflow.add_edge(START, "agent")
        workflow.add_conditional_edges("agent", should_approve, {"approval": "approval", END: END})
        workflow.add_edge("execute", END)
        workflow.add_edge("cancel", END)

        return workflow.compile(checkpointer=checkpointer)

---

四、前端审批组件

4.1 审批对话框

// static/js/approval-dialog.js

class ApprovalDialog {
    constructor() {
        this.dialog = null;
    }

    show(interruptData, onDecision) {
        this.close();

        const { action, risk_level, details, confirmation_required } = interruptData;

        // 根据风险等级选择样式
        const riskConfig = {
            high: {
                icon: '⚠️',
                color: '#f44336',
                title: '高风险操作'
            },
            medium: {
                icon: '⚡',
                color: '#ff9800',
                title: '敏感操作'
            },
            low: {
                icon: 'ℹ️',
                color: '#2196f3',
                title: '确认操作'
            }
        };

        const config = riskConfig[risk_level] || riskConfig.medium;

        this.dialog = document.createElement('div');
        this.dialog.className = 'approval-dialog-overlay';
        this.dialog.innerHTML = `
            <div class="approval-dialog ${risk_level}-risk">
                <div class="approval-header" style="border-color: ${config.color}">
                    <span class="approval-icon">${config.icon}</span>
                    <h3>${config.title}</h3>
                </div>
                <div class="approval-body">
                    <div class="approval-action">
                        操作: ${this.getActionDisplayName(action)}
                    </div>
                    <div class="approval-details">
                        ${details.message ? `<p>📝 ${details.message}</p>` : ''}
                        ${details.warning ? `<p class="warning">⚠️ ${details.warning}</p>` : ''}
                    </div>
                    ${confirmation_required ? `
                        <div class="confirmation-input">
                            <label>${confirmation_required}</label>
                            <input type="text" id="confirmation-text" placeholder="请输入确认文字">
                        </div>
                    ` : ''}
                </div>
                <div class="approval-footer">
                    <button class="btn-cancel" onclick="approvalDialog.cancel()">取消</button>
                    <button class="btn-confirm" onclick="approvalDialog.confirm()" style="background: ${config.color}">
                        确认执行
                    </button>
                </div>
            </div>
        `;

        document.body.appendChild(this.dialog);
        this.onDecision = onDecision;
    }

    confirm() {
        const confirmationInput = this.dialog.querySelector('#confirmation-text');
        if (confirmationInput && confirmationInput.value) {
            this.onDecision(true, { confirmation_text: confirmationInput.value });
        } else {
            this.onDecision(true);
        }
        this.close();
    }

    cancel() {
        this.onDecision(false);
        this.close();
    }

    close() {
        if (this.dialog) {
            this.dialog.remove();
            this.dialog = null;
        }
    }

    getActionDisplayName(action) {
        const names = {
            'delete_conversation': '删除对话',
            'delete_all_conversations': '删除所有对话',
            'export_all_conversations': '导出所有对话'
        };
        return names[action] || action;
    }
}

const approvalDialog = new ApprovalDialog();

4.2 CSS 样式

/* 审批对话框样式 */
.approval-dialog-overlay {
    position: fixed;
    top: 0;
    left: 0;
    right: 0;
    bottom: 0;
    background: rgba(0, 0, 0, 0.6);
    display: flex;
    align-items: center;
    justify-content: center;
    z-index: 2000;
    backdrop-filter: blur(4px);
}

.approval-dialog {
    background: white;
    border-radius: 16px;
    max-width: 480px;
    width: 90%;
    box-shadow: 0 8px 32px rgba(0, 0, 0, 0.2);
    overflow: hidden;
    animation: dialog-appear 0.2s ease-out;
}

@keyframes dialog-appear {
    from {
        opacity: 0;
        transform: scale(0.9);
    }
    to {
        opacity: 1;
        transform: scale(1);
    }
}

.approval-header {
    padding: 20px 24px;
    border-bottom: 3px solid;
    display: flex;
    align-items: center;
    gap: 12px;
}

.approval-icon {
    font-size: 24px;
}

.approval-header h3 {
    margin: 0;
    font-size: 18px;
    color: #333;
}

.approval-body {
    padding: 24px;
}

.approval-action {
    font-size: 16px;
    font-weight: 600;
    margin-bottom: 16px;
    color: #333;
}

.approval-details p {
    margin: 8px 0;
    font-size: 14px;
    color: #666;
}

.approval-details .warning {
    color: #f44336;
    font-weight: 500;
}

.confirmation-input {
    margin-top: 16px;
    padding-top: 16px;
    border-top: 1px solid #eee;
}

.confirmation-input label {
    display: block;
    font-size: 13px;
    color: #666;
    margin-bottom: 8px;
}

.confirmation-input input {
    width: 100%;
    padding: 12px;
    border: 2px solid #ddd;
    border-radius: 8px;
    font-size: 14px;
    transition: border-color 0.2s;
}

.confirmation-input input:focus {
    border-color: #4a90d9;
    outline: none;
}

.approval-footer {
    padding: 16px 24px;
    background: #f8f9fa;
    display: flex;
    gap: 12px;
    justify-content: flex-end;
}

.btn-cancel, .btn-confirm {
    padding: 12px 24px;
    border: none;
    border-radius: 8px;
    font-size: 14px;
    font-weight: 500;
    cursor: pointer;
    transition: all 0.2s;
}

.btn-cancel {
    background: #e0e0e0;
    color: #666;
}

.btn-cancel:hover {
    background: #d0d0d0;
}

.btn-confirm {
    color: white;
}

.btn-confirm:hover {
    filter: brightness(1.1);
}

/* 高风险特殊样式 */
.approval-dialog.high-risk .approval-header {
    background: #ffebee;
}

.approval-dialog.high-risk .btn-confirm {
    animation: pulse-warning 2s infinite;
}

@keyframes pulse-warning {
    0%, 100% { box-shadow: 0 0 0 0 rgba(244, 67, 54, 0.4); }
    50% { box-shadow: 0 0 0 8px rgba(244, 67, 54, 0); }
}

---

五、API 集成

# api/approval.py

from fastapi import APIRouter, Depends, Request
from pydantic import BaseModel
from typing import Optional, Dict, Any

router = APIRouter(prefix="/api", tags=["approval"])


class ApprovalRequest(BaseModel):
    thread_id: str
    approved: bool
    confirmation_text: Optional[str] = None


@router.post("/approval/resume")
async def resume_approval(request: ApprovalRequest, req: Request):
    """
    恢复中断的审批流程

    用户在前端确认或取消后，调用此接口继续执行。
    """
    from services.langgraph_approval import get_approval_agent
    from services.checkpointer import get_checkpointer

    agent = get_approval_agent()

    with get_checkpointer() as checkpointer:
        graph = agent.build_graph(checkpointer)
        config = {"configurable": {"thread_id": request.thread_id}}

        # 构建恢复命令
        if request.approved:
            resume_data = True
            if request.confirmation_text:
                resume_data = {
                    "approved": True,
                    "confirmation_text": request.confirmation_text
                }
        else:
            resume_data = False

        result = graph.invoke(
            Command(resume=resume_data),
            config
        )

        return {
            "success": True,
            "action_result": result.get("action_result", ""),
            "approved": request.approved
        }

---

六、预期收益

6.1 安全提升

场景	改进前	改进后
误删除	无法恢复	二次确认防误操作
批量删除	一键删除无提示	明确警告 + 输入验证
账号注销	即时生效	延迟确认期

6.2 用户体验

---

七、实施计划

步骤	任务	预估时间
1	创建 services/tools/sensitive_tools.py	2h
2	实现 services/langgraph_approval.py	2h
3	添加 API 端点 /api/approval/resume	1h
4	前端审批对话框组件	2h
5	集成测试	1h
总计		8h (1天)