Langchain-Chatchat数据脱敏处理：敏感信息展示保护机制-开发者社区

Langchain-Chatchat数据脱敏处理：敏感信息展示保护机制

在金融、医疗和法律等行业，AI驱动的知识库系统正逐步成为提升效率的核心工具。然而，一个无法回避的现实是：这些领域往往涉及大量个人身份信息（PII）、商业机密或受监管数据。一旦处理不当，哪怕是在内网部署的系统，也可能因问答输出中“无意泄露”一段身份证号或薪资记录而引发合规危机。

Langchain-Chatchat 作为一款开源本地化知识库问答框架，其最大优势在于“数据不出内网”。但很多人忽略了一个关键点——本地化 ≠ 绝对安全。如果用户提问“张三的合同金额是多少”，而系统直接返回原文中的“人民币800,000元整”，即便整个流程都在私有服务器上运行，依然构成了隐私暴露风险。

真正的安全，不仅在于数据存储的位置，更在于信息如何被呈现。为此，构建一套细粒度、可编程的敏感信息展示保护机制，已成为企业级部署不可或缺的一环。

Langchain-Chatchat 的架构天然支持这种深度定制。它基于 LangChain 框架设计，本质上是一个由多个模块串联而成的数据流水线：文档加载 → 文本切分 → 向量化 → 检索 → 生成。这个链式结构的最大价值在于——你可以在任意节点插入自己的逻辑，实现对数据流的精准控制。

比如，在文档入库阶段，我们可以先用 NER 模型扫描所有文本块，标记出姓名、身份证、银行账号等实体位置，并建立“敏感区域索引”；当用户发起查询时，检索到的相关段落会先经过一道“安全网关”检查，若命中高危字段，则自动替换为泛化描述或触发权限验证；最后，在 LLM 输出回答的过程中，还能通过回调函数实时清洗每一个 token，防止流式输出中“漏出”敏感内容。

这三层防护——预处理标记、检索拦截、生成过滤——共同构成了一个闭环的脱敏体系。它的核心思想不是简单地屏蔽关键词，而是根据上下文语义判断是否构成风险，并采取分级响应策略。

以手机号为例，传统的正则脱敏可能只做格式匹配：

re.sub(r'1[3-9]\d{9}', '138****5678', text)

但这在实际场景中远远不够。考虑这句话：“测试号码为13800138000，仅供内部使用。” 如果不做上下文识别，系统可能会误判为真实用户信息并进行遮蔽，影响可读性。更进一步，“张三比李四多2000元工资”这类间接表达，虽无明文数字，却仍可能导致语义推断泄露。因此，仅靠规则引擎远远不足。

解决方案是引入命名实体识别（NER）模型。借助微调过的中文 BERT 模型（如hfl/chinese-bert-wwm-ext），我们能准确识别“张三”为人名、“13800138000”为电话号码，并结合业务标签定义其敏感等级。以下是一个典型实现：

from transformers import pipeline ner_pipeline = pipeline( "token-classification", model="my-finetuned-ner-model", aggregation_strategy="simple" ) def detect_sensitive_entities(text): results = ner_pipeline(text) sensitive_types = ["PER", "ID_CARD", "PHONE", "BANK_ACCOUNT"] detected = [] for ent in results: if ent["entity_group"] in sensitive_types: detected.append({ "text": ent["word"], "type": ent["entity_group"], "start": ent["start"], "end": ent["end"] }) return detected

该过程建议在文档导入阶段完成，将检测结果以元数据形式存入向量数据库。例如，在 FAISS 或 Chroma 中，每条文本片段可附加一个sensitivity_score字段和redaction_zones列表，标明哪些区间需要特别处理。这样做的好处是避免每次查询都重复推理，显著降低延迟。

接下来是关键环节：检索后的动态过滤。LangChain 允许我们自定义Retriever组件。通过继承BaseRetriever类，可以创建一个SafeRetriever，在获取原始结果后主动执行风险评估：

class SafeRetriever(BaseRetriever): vector_retriever: BaseRetriever sensitive_detector: callable threshold: int = 2 def _get_relevant_documents(self, query: str): raw_docs = self.vector_retriever.get_relevant_documents(query) safe_docs = [] for doc in raw_docs: score = self.sensitive_detector.assess_risk_level(doc.page_content) if score >= self.threshold: cleaned_content = self.sensitive_detector.redact(doc.page_content) doc.page_content = f"[已脱敏] {cleaned_content}" safe_docs.append(doc) return safe_docs

这种方式实现了“最小权限原则”下的内容暴露控制。不同角色可配置不同的阈值：普通员工只能看到模糊化结果，HR 管理员经认证后可查看部分明文，审计人员则保留完整访问日志。

而在生成层，还可以叠加最后一道防线——流式输出脱敏。利用 LangChain 的回调机制，我们可以监听每个 token 的生成过程：

class SanitizingCallback(StreamingStdOutCallbackHandler): def on_llm_new_token(self, token: str, **kwargs) -> None: sanitized_token = sanitize_text(token) # 包含正则与轻量级 NER super().on_llm_new_token(sanitized_token, **kwargs) qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=safe_retriever, callbacks=[SanitizingCallback()] )

这种设计尤其适用于 Web 前端流式渲染场景，确保即使网络传输过程中也不会暴露原始内容。

整个系统的运作流程如下：