RAG安全问答系统实战：数据脱敏、防幻觉与审计追踪

前两天把一个内部知识库问答系统接到大模型上时，团队差点被一个问题劝退：模型回答得越来越“像人”，但检索出来的用户隐私数据也越来越多地出现在对话里。后来我们回头审视整个 AI 工程链路，发现真正要解决的并不是“模型够不够聪明”，而是“我们是否正在为了 AI 的效果，牺牲掉原本不该牺牲的东西”。

最近看到英文标题 “Are We Sacrificing Cemeteries for AI?”，直译是“我们是否在为了 AI 牺牲墓地？”。这里的“墓地”当然不是字面意思，而是一种比喻——我们多年积累的数据资产、历史文档、旧系统，以及用户隐私和模型可解释性。在 AI 应用中，这些“公墓”正在被悄悄挖开、搬进训练集和上下文窗口。本文从一个 RAG（检索增强生成）问答系统的完整实战出发，讨论大模型应用中的隐私泄露、模型幻觉、内容安全与审计追踪，并提供一套可落地的工程实现方案。

1. 背景与核心概念

1.1 什么是“数字墓地”与 AI 的代价

Cemetery 这个词放在 AI 语境下，通常指代那些已经“沉淀”下来的数据与知识：

• 多年运营产生的大量历史文档、工单、邮件；
• 已下线的旧系统里保存的用户行为记录；
• 人工维护但早已没人敢动的老数据库；
• 个人身份信息、联系方式、健康数据等敏感资产。

这些数据不会自己说话，但一旦被大模型“挖掘”，就可能以两种方式出现：一是作为训练语料被卷进模型参数；二是作为 RAG 检索的内容被拼进 Prompt，跟随用户的提问一起送到模型面前。

这两种方式都会带来一个隐患：我们为了 AI 的“好用”，可能正在牺牲数据所有权、隐私边界和可解释性。用工程的话说，就是数据安全与模型效果的trade-off。真正合格的 AI 系统不应该让这种牺牲成为默认选项，而应该靠架构设计去控制风险。

1.2 大模型应用中的三大风险

落地大模型应用时，笔者最关注三个技术风险：

这三个风险不是独立存在的。大模型没有“边界感”，它会忠实地理解用户想要什么，然后尽力生成一个看起来合理的回答。如果我们的知识库里有隐私数据，模型会把隐私数据缝合进答案；如果知识库里没有对应事实，模型会自己“脑补”一个事实。这也是为什么今天所有大模型应用都必须把“工程防护”放在模型能力之前。

1.3 为什么 RAG 是一把双刃剑

RAG 是当前大模型落地的标配方案。它先通过向量检索找到最相关的知识片段，再把片段拼进 Prompt，让模型基于这些片段回答。它的好处显而易见：答案可以被溯源，模型不依赖参数记忆，更新知识只要改数据库。

但 RAG 也把风险放大了：

• 检索结果一旦包含隐私数据，模型会原样引用；
• 检索结果如果互相矛盾，模型可能自动“调和”出错误结论；
• Prompt 里的上下文越长，模型越容易被干扰，输出不可控的片段。

所以，RAG 不是简单把文档扔进向量库就完事。它需要数据脱敏、权限控制、内容过滤、提示词约束，以及完整的审计链路。下面我们就从工程角度，一步步构建一套这样的系统。

2. 环境准备与版本说明

2.1 运行环境与依赖

本文示例以 Python 3.10 为主，操作系统推荐 Ubuntu 22.04 或 macOS，Windows 下同样可运行，但需要注意本地模型和 ChromaDB 的路径写法。版本不必完全复制，但建议保持相近：

• Python 3.10+
• Ollama 0.5+（本地模型推理）
• ChromaDB 0.5+
• LangChain 仅用于理解思路，本示例尽量不用重量级框架，方便看清内部逻辑

需要安装的 Python 依赖：

pip install chromadb ollama

其中ollama是 Ollama 的 Python 客户端，我们用它调用本地模型和嵌入模型；chromadb用于向量存储与检索。

2.2 本地模型与向量库选型

为了数据不离开本地环境，本示例全部使用 Ollama 加载本地模型：

• 对话模型：qwen2.5:7b，负责最终回答生成；
• 嵌入模型：nomic-embed-text，负责把文档和用户问题转成向量。

先确保 Ollama 已启动，并拉取模型：

ollama pull qwen2.5:7b ollama pull nomic-embed-text

注意，模型名称会随版本更新变化，请以你本地ollama list的结果为准。

2.3 项目结构

ai_guard/ ├── data/ │ └── source_docs.txt ├── src/ │ ├── __init__.py │ ├── anonymizer.py │ ├── retriever.py │ ├── safe_llm.py │ ├── audit.py │ └── main.py ├── requirements.txt └── README.md

src/anonymizer.py负责隐私数据脱敏；src/retriever.py负责文档加载、向量化与检索；src/safe_llm.py负责输入输出安全过滤与模型调用；src/audit.py负责审计日志；src/main.py负责组装整个问答流程。

3. 核心设计：从数据到答案的完整链路

3.1 数据采集与缺失整理

很多团队会把原始文档直接喂给 RAG，这是最大的错误。原始文档可能包含：

• 用户手机号、身份证、邮箱；
• 内部会议记录与未公开决策；
• 互相矛盾的旧版本说明。

所以在进入向量库之前，必须做一次数据准入：能入库的文档要满足可公开、低敏感、结构清晰。这一步在代码里体现为“脱敏 + 清洗”，但更重要的是制度上明确什么数据可以进 RAG。

3.2 数据脱敏与匿名化

数据脱敏的常见方式有：

• 替换：把手机号替换为[手机号]；
• 掩码：保留前3位后4位，中间用****替代；
• 删除：直接移除敏感字段。

在 RAG 链路里，最推荐“替换”和“掩码”，因为我们需要保留上下文逻辑，但又不暴露真实值。比如“李四的电话是 138****1234” 比“李四的电话是 [手机号]” 更自然，同时保护了隐私。

3.3 文档切分与向量化

文档进入向量库前需要切分。切分大小直接影响检索效果：

• 切分太短，缺少上下文，检索结果碎片化；
• 切分太长，噪声太多，相关片段淹没在过长的文本里。

一般建议按章节或段落切分，每个 chunk 控制在 200 到 500 个 token。示例中为了简化，我们按固定字数切分。

向量化阶段，使用nomic-embed-text把每个 chunk 转成向量。实际项目中可以选择更专业的 embedding 模型，但原理一致。

3.4 检索增强生成与引用溯源

RAG 的标准流程是：

1. 用户提问；
2. 向量化用户问题；
3. 在向量库中检索 top-k 相关 chunk；
4. 把 chunk 拼进 Prompt；
5. 大模型基于 Prompt 生成回答。

为了控制幻觉，Prompt 中必须明确要求模型只基于给定片段回答，如果片段不足以支撑答案，就回答“暂无相关信息”。同时，我们可以让模型返回引用来源序号，帮助用户验证。

3.5 内容安全过滤

内容安全不能只靠模型“自律”。必须设置两个环节：

• 输入过滤：用户问题中如果包含敏感词或越权内容，直接拦截；
• 输出过滤：模型回答中如果出现敏感词或不合规内容，再次拦截并返回兜底文案。

这里需要注意，关键词过滤只是最基础的防线。生产级系统还需要引入分类模型、语义向量相似度，甚至人工审核抽检。

3.6 审计日志

AI 应用必须可追溯。每次问答都应记录：

• 用户标识（脱敏后）；
• 完整问题（脱敏后）；
• 检索到的文档片段 ID；
• 模型完整回答；
• 时间戳与操作结果。

日志本身也是敏感数据，建议加密存储并设置访问权限。

4. 完整实战案例：构建一个带安全护栏的 RAG 问答系统

4.1 创建项目与依赖

创建项目目录：

mkdir ai_guard cd ai_guard mkdir -p data src touch src/__init__.py

写入requirements.txt：

chromadb>=0.5.0 ollama>=0.3.0

安装依赖：

pip install -r requirements.txt

4.2 数据准备

在data/source_docs.txt中放入少量测试文档。注意不要放真实敏感数据，示例仅为演示：

员工手册： 公司实行弹性工作制，核心工作时间为上午10点到下午4点。 员工年度体检可在任何三甲医院完成，报销上限为1000元。 客户服务规范： 客户投诉应在24小时内响应，并在3个工作日内给出解决方案。 联系电话：13812345678，联系邮箱：service@example.com。 假期政策： 员工每年享有12天带薪年假，入职满一年后开始计算。 法定节假日按国家规定执行。

可以看到文档里原本有手机号和邮箱，我们会用脱敏模块把它们替换掉。

4.3 数据脱敏模块

文件路径：src/anonymizer.py

import re class Anonymizer: """对文本中的手机号、身份证号、邮箱进行掩码脱敏。""" def __init__(self): self.patterns = [ # 手机号：1开头 + 10位数字 (re.compile(r'(1[3-9]\d)\d{4}(\d{4})'), r'\1****\2'), # 身份证号：17位数字 + 数字或X (re.compile(r'(\d{6})\d{8}(\d{3}[\dXx])'), r'\1********\2'), # 邮箱：用户名@域名 (re.compile(r'([\w.+-]+)@([\w-]+\.[\w.]+)'), r'***@\2'), ] def anonymize(self, text: str) -> str: """返回脱敏后的文本。""" for pattern, repl in self.patterns: text = pattern.sub(repl, text) return text

这段代码的思路是：先匹配敏感数据，再用正则替换为掩码形式。re.sub替换时保留部分结构，例如手机号前三位和后四位，中间用****代替。这样既保护隐私，又不让语义完全丢失。

4.4 向量库与检索模块

文件路径：src/retriever.py

import os import chromadb import ollama EMBED_MODEL = "nomic-embed-text" COLLECTION_NAME = "docs" CHROMA_DIR = "./chroma_db" class Retriever: def __init__(self, source_file: str, chunk_size: int = 200): self.chunk_size = chunk_size self.client = chromadb.PersistentClient(path=CHROMA_DIR) self.collection = self.client.get_or_create_collection(name=COLLECTION_NAME) self.source_file = source_file def _read_and_chunk(self) -> list[str]: """读取文档并按固定长度切分。""" with open(self.source_file, "r", encoding="utf-8") as f: text = f.read() chunks = [] current = "" for line in text.splitlines(): current += line + "\n" if len(current) >= self.chunk_size: chunks.append(current.strip()) current = "" if current.strip(): chunks.append(current.strip()) return chunks def build_index(self): """将文档切分、向量化并写入 ChromaDB。""" chunks = self._read_and_chunk() ids = [f"chunk_{i}" for i in range(len(chunks))] embeddings = [] for chunk in chunks: resp = ollama.embeddings(model=EMBED_MODEL, prompt=chunk) embeddings.append(resp["embedding"]) self.collection.add( ids=ids, embeddings=embeddings, documents=chunks, ) print(f"索引完成：共 {len(chunks)} 个片段。") def query(self, question: str, top_k: int = 3) -> list[str]: """查询最相关的 top_k 个文档片段。""" resp = ollama.embeddings(model=EMBED_MODEL, prompt=question) query_embedding = resp["embedding"] result = self.collection.query( query_embeddings=[query_embedding], n_results=top_k, ) return result["documents"][0]

这个模块有两个关键点：

1. PersistentClient(path)指定向量数据库持久化目录，避免每次启动重建索引；
2. build_index()会重复写入相同片段，若重新执行需先删除已有集合，否则会报 ID 冲突。生产环境建议使用文档哈希作为 ID，避免重复入库。

4.5 安全过滤与模型调用

文件路径：src/safe_llm.py

import ollama CHAT_MODEL = "qwen2.5:7b" # 示例敏感词列表，请按业务场景替换 SENSITIVE_WORDS = ["示例敏感词1", "示例敏感词2"] class SafeLLM: def __init__(self): self.sensitive_words = SENSITIVE_WORDS def check_text(self, text: str) -> bool: """输入输出安全过滤：含敏感词返回 False。""" for word in self.sensitive_words: if word in text: return False return True def generate(self, question: str, context_chunks: list[str]) -> str: # 输入过滤 if not self.check_text(question): return "抱歉，您的问题包含不当内容，无法回答。" # 组装上下文 context = "\n\n".join( [f"[片段{i+1}]\n{chunk}" for i, chunk in enumerate(context_chunks)] ) system_prompt = ( "你是一个严谨的智能助手。请只能基于以下提供的知识片段回答问题。" "如果片段中没有提到相关信息，请明确回答“根据现有知识无法回答”。" "不要编造事实，不要泄露任何上下文之外的隐私信息。" "回答时可引用片段编号，例如“根据片段1”。" ) user_prompt = f"知识片段：\n{context}\n\n用户问题：{question}" response = ollama.chat( model=CHAT_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) answer = response["message"]["content"] # 输出过滤 if not self.check_text(answer): return "抱歉，模型生成内容未通过安全审核。" return answer

这里把安全过滤和模型调用放在一起，是为了演示“输入检查 -> 检索 -> Prompt 组装 -> 模型调用 -> 输出检查”的完整流程。注意，真正的生产级过滤不能只有关键词，还要有语义模型和人工审核。

4.6 审计日志模块

文件路径：src/audit.py

import logging from datetime import datetime class AuditLogger: def __init__(self, log_file: str = "audit.log"): self.logger = logging.getLogger("audit") self.logger.setLevel(logging.INFO) fh = logging.FileHandler(log_file, encoding="utf-8") fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s") fh.setFormatter(fmt) self.logger.addHandler(fh) def log(self, user_id: str, question: str, answer: str, retrieval_ids: list[str]): record = { "user_id": user_id, "question": question, "answer": answer, "retrieval_ids": retrieval_ids, "time": datetime.now().isoformat(), } self.logger.info(record)

在实际系统中，user_id应来自统一身份认证，且存储前需要做哈希或脱敏；日志文件要定期归档，并设置文件权限防止未授权读取。

4.7 主流程串联

文件路径：src/main.py

from anonymizer import Anonymizer from retriever import Retriever from safe_llm import SafeLLM from audit import AuditLogger def main(): source_file = "data/source_docs.txt" anonymizer = Anonymizer() retriever = Retriever(source_file) safe_llm = SafeLLM() audit = AuditLogger() # 首次运行需要构建索引，之后可注释掉 retriever.build_index() print("AI 安全问答系统已启动，输入 exit 退出。") while True: question = input("\n请输入问题：") if question.strip().lower() == "exit": break # 1. 输入脱敏与过滤 question_clean = anonymizer.anonymize(question) # 2. 检索 chunks = retriever.query(question_clean, top_k=3) # 3. 生成 answer = safe_llm.generate(question_clean, chunks) # 4. 输出 print("\n--- 回答 ---") print(answer) # 5. 审计 audit.log( user_id="anonymous", question=question_clean, answer=answer, retrieval_ids=[f"chunk_{i}" for i in range(len(chunks))], ) if __name__ == "__main__": main()

4.8 运行与验证

在项目根目录下运行：

python src/main.py

首次运行会构建向量索引，控制台会输出：

索引完成：共 6 个片段。 AI 安全问答系统已启动，输入 exit 退出。

接着输入问题，例如：

请输入问题：客户服务热线是什么？

系统会查询向量库，把最相关的几个片段作为上下文，交给大模型生成回答。由于我们已经对文档做过脱敏，source_docs.txt中的手机号会以138****5678的形式出现，而不是完整号码。

如果输入问题中包含“示例敏感词1”，系统会直接拦截，不会调用模型。

整个流程的日志会写入audit.log，你可以打开文件确认每条问答都有记录。

5. 常见问题与排查思路

5.1 模型幻觉仍然存在

还可以对模型回答做“事实一致性校验”：用另一轮模型判断回答是否严格基于给定片段。虽然会额外耗时，但能显著降低对外发布时的幻觉风险。

5.2 敏感数据绕过脱敏

这里要特别强调：脱敏不是一次性操作，而是要在每个数据出口都执行一次。包括 Prompt 进入模型前、模型输出返回用户前。

5.3 内容安全误杀过多

生产环境建议把“直接拦截”和“告警后继续”分开，并配合人工审核队列。

5.4 检索召回效果差

可以通过召回率、命中率等指标评估检索效果，再针对性地调参。

5.5 本地模型性能不足

本地模型的好处是数据不出内网，适合隐私要求高的场景；但如果性能不足，也可以采用私有化部署 API 网关的方式，在网关层做同样的安全控制和审计。

6. 最佳实践与工程建议

6.1 数据治理与最小化原则

不要把所有数据都接入 AI 应用。遵循最小化原则：

• 只放当前业务需要的、适合公开给模型的数据；
• 敏感性高的数据放到另一套权限控制更强的系统中，不直接接入 RAG；
• 文档入库前由负责人审批，而不是让开发人员自己决定。

可以用一张清单做准入检查：

[ ] 是否包含可公开信息？ [ ] 是否包含个人敏感字段？ [ ] 是否已经过脱敏处理？ [ ] 是否明确数据归属与更新责任人？

6.2 提示词工程与安全护栏

提示词不是随便写写。关键要写明：

• 模型只能使用给定片段；
• 不知道时主动承认；
• 禁止输出 Prompt 本身；
• 禁止尝试绕过系统限制。

同时，输入输出过滤要独立于提示词，不能只依赖模型“自觉”。可借鉴 AWS 的 “guardrails” 思路，把内容策略定义从业务代码中剥离出来，便于统一管理。

6.3 可解释性与来源引用

对外提供服务时，用户需要知道答案的根据。建议在回答中附带文档编号或来源链接：

根据员工手册（片段2），核心工作时间为上午10点到下午4点。

如果模型生成内容没有引用来源，宁可返回“无法回答”，也不要直接展示。这也是控制幻觉的有效手段。

6.4 权限控制与审计

AI 应用必须做用户级权限控制。不同的用户应该看到不同的知识范围。实现方式有两种：

• 检索前过滤：根据用户角色过滤向量库中文档的可见性；
• 生成后过滤：对模型输出再次做权限裁剪。

第一种更安全，因为不安全的上下文根本不会进入 Prompt。审计日志则要包含用户 ID、问题、回答、检索片段、时间戳和审核状态，并且日志本身要防篡改。

6.5 模型迭代与灰度发布

大模型的版本升级可能带来不可预见的输出变化。上线前要做红队测试，用一批“危险问题”检验模型的越狱能力、幻觉率和安全过滤召回率。建议：

• 每次切换模型前，先跑一遍自动化测试集；
• 采用 A/B 或灰度发布，让少量用户先体验；
• 保留旧模型回滚通道，出现问题时快速切换。

7. 下一步学习路线

如果你刚开始接触 AI 工程，建议按这个顺序深入：

1. 掌握 RAG 原理：理解向量检索、文档切分、Prompt 拼装；
2. 学习提示词工程：学会设计 system prompt，降低幻觉；
3. 实践数据脱敏：从正则到 PII 识别模型，逐步完善；
4. 了解向量数据库：从 Chroma 到 Milvus、Qdrant 等生产级方案；
5. 了解内容安全：语义过滤、多模态审核、人工审核流程；
6. 体系化评估：建立离线测试集和线上监控指标，持续优化。

真正成熟的 AI 应用，一定是在模型效果和数据安全之间找到了平衡。不要把“提高准确率”作为唯一目标，还要多想想——你正在用什么样的数据去喂养模型，那些数据背后的“数字墓地”，是否真的允许被这样打开。