Langchain-Chatchat实现知识库操作审批流程设计

格拉摩根终身伯爵

725人浏览 · 2025-12-19 12:29:57

格拉摩根终身伯爵 · 2025-12-19 12:29:57 发布

Langchain-Chatchat 实现知识库操作审批流程设计

在企业智能化转型的浪潮中，一个日益突出的问题浮出水面：我们如何让 AI 助手既聪明又守规矩？

尤其是在金融、医疗、法律等对合规性要求极高的行业，知识库系统不再只是“能不能回答”的问题，而是“该不该现在回答”“谁能授权回答”的治理命题。许多企业已经部署了基于大模型的知识问答系统，但一旦出现敏感信息误输出——比如员工薪酬细节被随意查询，或是未公开的合同条款被自动披露——轻则引发内部争议，重则触发监管风险。

正是在这种背景下，Langchain-Chatchat 作为国内领先的本地化知识库解决方案，因其全流程离线运行、数据不出内网的特性，成为众多企业的首选。而在此基础上引入“操作审批流程”，则是将这套工具从“可用”推向“可信”的关键一步。

为什么需要审批？从一次真实场景说起

设想这样一个场景：某公司实习生在使用内部知识助手时，随口问了一句：“去年离职员工的补偿标准是什么？” 如果系统直接调用向量数据库检索并生成答案，可能会无意中泄露公司的人力资源政策细节。虽然AI没有恶意，但它也不懂“分寸”。

如果此时系统能暂停响应，转为“该请求需主管审批”，并将任务推送到HR负责人的待办列表，就能有效避免信息越权传播。这正是审批机制的核心价值——在自动化与安全性之间建立一道可控的闸门。

Langchain-Chatchat 的开源架构和模块化设计，使得这种“人机协同”模式的集成变得切实可行。它不像封闭云服务那样黑箱操作，而是允许开发者深入干预每一个处理环节，从而实现精细化的访问控制。

审批不是阻断，而是智能路由

很多人担心加入审批会拖慢响应速度，影响用户体验。但实际上，合理的审批机制并不是给所有请求都加上锁，而是做智能分流。

我们可以把整个问答流程想象成一条高速公路：

大部分常规问题（如“会议室怎么预约？”“报销流程是什么？”）走的是“快速通道”，秒级响应；
少数涉及敏感内容或高权限文档的请求，则被引导至“安检通道”，由规则引擎初步筛查后，交由人工审核决定是否放行。

这种设计的关键在于：只拦截该拦的，不打扰正常的使用节奏。

要实现这一点，系统必须具备多维度判断能力。例如：

内容层面：提问中是否包含“薪资”“机密”“客户名单”等关键词？
用户层面：提问者是正式员工还是外包人员？角色是否具备豁免权？
资源层面：目标知识文档本身是否被打上“confidential”标签？
行为层面：该用户今日已发起超过50次查询，是否存在异常爬取嫌疑？

这些条件可以组合成一套灵活的策略规则，动态决定是否触发审批。真正做到了“该严则严，该放则放”。

技术落地：如何在不重构的前提下植入审批逻辑？

Langchain-Chatchat 基于 FastAPI 构建后端服务，这为我们提供了天然的切入点——中间件（Middleware）机制。

通过编写一个轻量级的审批拦截中间件，我们可以在用户请求到达主处理逻辑之前，先进行一轮“前置检查”。如果命中审批规则，就中断流程、保存任务上下文，并返回提示信息；否则，让请求继续流向 RAG 流程。

下面是一个实际可运行的代码示例：

# middleware/approval_middleware.py
from fastapi import Request, HTTPException
from typing import Dict, List
import redis
import json
from datetime import datetime

# 初始化 Redis 连接
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

# 敏感关键词库（建议从数据库动态加载）
SENSITIVE_KEYWORDS: List[str] = ["薪酬", "薪资", "绩效", "裁员", "客户清单", "财务报表", "离职补偿"]

# 需要审批的角色列表
APPROVAL_REQUIRED_ROLES: List[str] = ["guest", "contractor", "intern"]

def contains_sensitive_keyword(text: str) -> bool:
    """检查文本是否包含敏感关键词"""
    return any(keyword in text for keyword in SENSITIVE_KEYWORDS)

def is_approval_needed(user_role: str, query_text: str) -> bool:
    """综合判断是否需要审批"""
    if user_role in APPROVAL_REQUIRED_ROLES:
        return True
    if contains_sensitive_keyword(query_text):
        return True
    return False

async def check_approval_status(request: Request) -> Dict:
    """
    审批中间件：拦截请求并检查是否需要审批
    """
    body = await request.json()
    question = body.get("query", "")
    user_id = body.get("user_id")
    user_role = body.get("role", "user")

    # 判断是否需要审批
    if not is_approval_needed(user_role, question):
        return {"proceed": True}  # 直接放行

    # 生成唯一任务ID
    task_id = f"approval:{user_id}:{int(datetime.now().timestamp())}"

    # 构建审批任务
    approval_task = {
        "task_id": task_id,
        "user_id": user_id,
        "question": question,
        "timestamp": datetime.now().isoformat(),
        "status": "pending"
    }

    # 存入Redis等待处理（保留24小时）
    redis_client.setex(task_id, 86400, json.dumps(approval_task))

    # 中断流程，返回403
    raise HTTPException(
        status_code=403,
        detail={
            "message": "此操作需要审批",
            "task_id": task_id,
            "status": "pending"
        }
    )

这段代码看似简单，却完成了几个关键动作：

非侵入式拦截：作为中间件挂载在 FastAPI 路由前，无需改动原有 RAG 逻辑；
上下文暂存：利用 Redis 临时存储待审请求，支持异步处理；
状态可追溯：每个任务都有独立 ID 和时间戳，便于后续审计；
超时自动清理：设置 TTL 为 24 小时，防止队列堆积。

配合另一个 /approve/{task_id} 接口，管理员批准后可主动触发后续的检索与生成流程，形成完整闭环。

系统架构升级：从单一流程到双轨并行

当审批机制嵌入后，整体系统架构也随之演进为更健壮的双轨结构：

graph TD
    A[用户提问] --> B{是否需审批?}
    B -- 否 --> C[RAG流程: 检索+生成]
    B -- 是 --> D[创建审批任务]
    D --> E[存入Redis队列]
    E --> F[通知审批员]
    F --> G[审批后台处理]
    G --> H{批准?}
    H -- 是 --> C
    H -- 否 --> I[返回拒绝信息]
    C --> J[返回回答给用户]
    I --> J
    G --> K[记录审计日志]
    J --> K

在这个新架构中，几个核心组件的作用更加清晰：