RAG 数据治理

RAG 的效果不只取决于模型。很多线上答错、越权、引用过期、召回混乱的问题，本质都不是“模型不够聪明”，而是知识库数据没有治理好。

一句话理解：

RAG 数据治理，就是让进入知识库的资料来源清楚、内容干净、权限正确、版本可控、更新及时、删除同步、质量可评估、问题可追溯。

如果没有数据治理，RAG 很容易从“企业知识库问答”退化成“随机片段拼接器”。模型看到什么就会基于什么回答，资料错、权限错、切分错、版本错，最终答案就会错。

学习目标

学完本页，你应该能回答：

1. 为什么 RAG 数据治理比 Prompt 更长期、更重要。
2. 文档从采集、解析、清洗、脱敏、切分、向量化、发布、删除的完整生命周期是什么。
3. Chunk 元数据应该设计哪些字段，为什么不能只存文本和向量。
4. 增量更新、全量重建、索引发布、索引回滚分别怎么做。
5. 权限过滤为什么必须发生在检索前或检索中。
6. 文档去重、Chunk 去重、版本去重有什么区别。
7. 如何处理文档删除、权限变更、资料冲突和过期知识。
8. 如何建设 RAG 数据质量指标和质量报告。
9. 如何写一个最小可运行的数据治理 Demo。
10. 面试时如何说明 RAG 数据治理和线上排查。

为什么 RAG 必须做数据治理

RAG 的完整链路可以简化成：

flowchart TD
    A["企业资料"] --> B["切分和向量化"]
    B --> C["用户提问时检索"]
    C --> D["模型基于检索片段回答"]

模型的回答质量，受上游资料质量强烈影响。

数据问题	线上表现
旧制度和新制度同时存在	模型引用过期制度
同一内容重复入库	TopK 被重复片段占满
Chunk 太碎	模型看到半句话，回答断章取义
Chunk 太长	检索命中不精确，Prompt 噪声大
元数据缺失	无法引用来源、无法按权限过滤
权限标签错误	普通用户检索到敏感文档
原文删除但向量未删	已下线内容仍被回答
Embedding 模型换了但未重建	查询向量和文档向量空间不一致

RAG 项目早期看 Prompt，长期看数据治理。Prompt 能约束模型“怎么答”，但数据治理决定模型“看到了什么资料”。

数据治理总流程

flowchart TD
    A["接入数据源"] --> B["解析文本和结构"]
    B --> C["清洗噪声和重复"]
    C --> D["脱敏和安全分级"]
    D --> E["按语义切分 Chunk"]
    E --> F["补充元数据"]
    F --> G["生成内容哈希"]
    G --> H["生成 Embedding"]
    H --> I["写入候选索引"]
    I --> J["质量检查和评估"]
    J --> K{"是否通过发布门槛"}
    K -- "通过" --> L["发布为线上索引"]
    K -- "不通过" --> M["修复数据或规则"]
    L --> N["监控召回和用户反馈"]

这个流程里每一步都不能只做“能跑”。商业系统要能回答：

1. 这段答案来自哪个文档、哪个版本、哪个 Chunk？
2. 用户为什么有权限看到这段资料？
3. 这份文档什么时候入库、由谁维护、是否已经过期？
4. 如果答案错了，是解析错、切分错、召回错、排序错，还是模型生成错？
5. 如果知识库发布出问题，能否切回上一版？

文档生命周期

知识库不是一次性导入文件。真实项目里文档会新增、修改、删除、下线、权限变更、版本升级。

flowchart TD
    A["草稿或原始资料"] --> B["审核通过"]
    B --> C["入库和向量化"]
    C --> D["发布到线上索引"]
    D --> E["内容更新"]
    E --> F["重新切分和向量化"]
    F --> G["新索引评估"]
    G --> H["灰度切换"]
    D --> I["权限变更"]
    I --> J["同步权限元数据"]
    D --> K["文档删除或下线"]
    K --> L["删除结构化记录和向量"]

生命周期治理的关键点：

事件	必须处理	不处理会怎样
新增文档	解析、切分、补元数据、入库	用户检索不到新知识
更新文档	生成新版本 Chunk，替换旧版本	新旧答案混用
删除文档	删除原文、Chunk、向量、缓存	下线内容继续被回答
权限变更	更新 metadata filter 字段	越权或误拒答
Embedding 模型变更	全量重建向量	查询和文档向量空间不一致
切分策略变更	重切 Chunk 并评估	召回质量不可控

元数据设计

向量库里不能只存 text 和 vector。没有元数据，RAG 很快就会不可维护。

推荐元数据：

字段	示例	作用
chunkId	doc-1001:3:v2	Chunk 唯一标识
docId	doc-1001	文档唯一标识
title	LIS 数据资产目录	用于引用和排序
path	/assets/lis	用于跳转来源
headingPath	资产目录 > 检验结果	保留语义层级
chunkIndex	3	保留片段顺序
version	v2	支持版本控制
contentHash	sha256...	判断内容是否变更
sourceType	markdown/pdf/db	区分来源
owner	信息科	负责人和数据治理责任
tenantId	hospital-a	多租户隔离
permissionTags	dept:信息科,level:secret	检索前权限过滤
effectiveAt	2026-07-01	生效时间
expiredAt	2026-12-31	过期时间
embeddingModel	bge-m3	判断是否需要重建
ingestBatchId	batch-20260706	关联入库批次

为什么 contentHash 很重要

内容哈希用于判断文档片段是否真的发生变化。

同一个 docId + chunkIndex，如果 contentHash 没变，可以跳过重新 Embedding。
如果 contentHash 变了，必须重新生成向量并替换旧 Chunk。

这样可以降低增量更新成本，也能避免重复写入。

为什么要保存 embeddingModel

不同 Embedding 模型生成的向量不在同一个语义空间里。文档用模型 A 入库，问题用模型 B 查询，距离计算就不可靠。

所以每个 Chunk 都要记录 embeddingModel。如果线上查询模型切换了，要么保持兼容索引，要么全量重建。

权限治理

权限是 RAG 生产落地里最不能糊弄的部分。

错误链路：

flowchart TD
    A["全库检索"] --> B["召回敏感资料"]
    B --> C["放入 Prompt"]
    C --> D["要求模型不要泄露"]

这条链路的问题是：敏感资料已经进入模型上下文，风险已经发生。

正确链路：

flowchart TD
    A["用户身份"] --> B["计算租户、部门、角色、密级"]
    B --> C["生成 metadata filter"]
    C --> D["只检索用户可见 Chunk"]
    D --> E["组装 Prompt"]
    E --> F["模型基于可见资料回答"]

权限过滤要尽量下推到检索层：

过滤维度	示例
租户	tenantId = hospital-a
部门	permissionTags contains dept:信息科
角色	permissionTags contains role:data_admin
密级	securityLevel <= user.securityLevel
数据范围	ownerOrg in user.orgTree
有效期	effectiveAt <= now < expiredAt

如果向量库不支持复杂 filter，可以在结构化数据库里先算可访问 docId，再把 docId 作为过滤条件传给向量检索。

去重治理

RAG 里的重复有三层。

类型	例子	处理方式
文档重复	同一制度上传了两份 PDF	按 docId、路径、标题、内容指纹去重
Chunk 重复	页眉、页脚、免责声明重复出现	按 contentHash 或相似度去重
版本重复	v1 和 v2 同时在线	按生效时间、版本状态过滤

重复内容的危害：

1. TopK 被相同内容占满，召回多样性下降。
2. 模型误以为重复内容更重要。
3. token 成本增加。
4. 旧版本和新版本混在一起，答案互相冲突。

去重不能简单“相同文本删掉”。有些文本相同但权限不同、租户不同、版本不同，不能合并。正确做法是先按业务身份判断，再按内容指纹处理。

增量更新与全量重建

增量更新

增量更新适合文档少量变化。

flowchart TD
    A["读取当前文档"] --> B["解析并切分"]
    B --> C["计算每个 Chunk 的 Hash"]
    C --> D{"Hash 是否变化"}
    D -- "未变化" --> E["跳过 Embedding"]
    D -- "变化" --> F["重新向量化"]
    F --> G["写入新 Chunk"]
    G --> H["删除旧 Chunk 或标记失效"]

增量更新的关键是要有稳定的文档 ID、Chunk ID 和内容哈希。如果每次入库都生成随机 ID，就很难判断哪些内容变了。

全量重建

以下情况建议全量重建：

1. 更换 Embedding 模型。
2. 切分策略大改。
3. 元数据权限模型大改。
4. 大量历史数据质量差，需要重新清洗。
5. 向量库索引参数或存储结构大改。

全量重建不要直接覆盖线上索引。推荐使用“双索引发布”：

flowchart TD
    A["线上索引 v1"] --> B["后台构建索引 v2"]
    B --> C["离线评估 v2"]
    C --> D{"评估通过"}
    D -- "否" --> E["废弃 v2，继续使用 v1"]
    D -- "是" --> F["灰度切换少量流量到 v2"]
    F --> G{"线上指标稳定"}
    G -- "是" --> H["全量切换到 v2"]
    G -- "否" --> I["回滚到 v1"]

删除同步和过期治理

删除是很多 RAG 系统最容易漏的环节。

文档删除时要处理：

1. 原文表记录下线或删除。
2. Chunk 表记录标记失效。
3. 向量库删除对应向量。
4. 缓存中的召回结果失效。
5. 评测集和引用链接更新。

如果只删除原文，不删向量库，用户仍然可能检索到旧 Chunk。更危险的是，页面已经无法访问来源，但模型仍然能回答旧内容。

推荐做软删除：

status = active / inactive / deleted

检索时只允许 status = active，后台异步清理 deleted 向量。这样既能快速阻断线上召回，又能保留审计记录。

质量指标

RAG 数据治理要有可观察指标，不能靠感觉。

指标	含义	风险信号
文档解析成功率	成功解析文档 / 总文档	低说明格式或解析器有问题
空 Chunk 比例	空内容 Chunk 数 / 总 Chunk	高说明清洗或解析异常
超长 Chunk 比例	超过长度上限的 Chunk	高说明切分策略不合理
元数据完整率	必填元数据齐全比例	低会影响权限和引用
权限标签覆盖率	有权限标签 Chunk 比例	低可能越权或误拒答
重复 Chunk 比例	重复内容比例	高会污染 TopK
过期文档比例	已过期仍 active 的文档	高会引用旧内容
召回命中率	正确答案 Chunk 是否进入 TopK	低说明检索质量差
引用有效率	答案引用是否能打开	低说明来源不可追溯

每次入库完成后应该生成质量报告：

{
  "batchId": "batch-20260706",
  "totalDocs": 120,
  "successDocs": 118,
  "failedDocs": 2,
  "totalChunks": 3560,
  "emptyChunks": 3,
  "duplicateChunks": 82,
  "missingPermissionTags": 0,
  "expiredActiveDocs": 0,
  "embeddingModel": "bge-m3",
  "status": "passed"
}

商业 Demo：RAG 入库治理脚本

下面 Demo 演示一个简化的 RAG 入库治理流程：标准化文档、切分 Chunk、生成元数据、计算哈希、去重、做质量检查。

from __future__ import annotations

from dataclasses import dataclass
import hashlib
from typing import Iterable


@dataclass
class Document:
    doc_id: str
    title: str
    path: str
    content: str
    tenant_id: str
    permission_tags: set[str]
    version: str
    owner: str


@dataclass
class Chunk:
    chunk_id: str
    doc_id: str
    text: str
    metadata: dict


def sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def normalize_text(text: str) -> str:
    lines = [line.strip() for line in text.splitlines()]
    lines = [line for line in lines if line]
    return "\n".join(lines)


def split_by_paragraph(text: str, max_chars: int = 180) -> list[str]:
    paragraphs = text.split("\n")
    chunks: list[str] = []
    current = ""

    for paragraph in paragraphs:
        candidate = paragraph if not current else current + "\n" + paragraph
        if len(candidate) <= max_chars:
            current = candidate
        else:
            if current:
                chunks.append(current)
            current = paragraph

    if current:
        chunks.append(current)
    return chunks


def build_chunks(doc: Document, embedding_model: str, batch_id: str) -> list[Chunk]:
    normalized = normalize_text(doc.content)
    texts = split_by_paragraph(normalized)
    chunks: list[Chunk] = []

    for index, text in enumerate(texts):
        content_hash = sha256(text)
        chunk_id = f"{doc.doc_id}:{doc.version}:{index}"
        chunks.append(
            Chunk(
                chunk_id=chunk_id,
                doc_id=doc.doc_id,
                text=text,
                metadata={
                    "docId": doc.doc_id,
                    "title": doc.title,
                    "path": doc.path,
                    "version": doc.version,
                    "chunkIndex": index,
                    "tenantId": doc.tenant_id,
                    "permissionTags": sorted(doc.permission_tags),
                    "owner": doc.owner,
                    "contentHash": content_hash,
                    "embeddingModel": embedding_model,
                    "ingestBatchId": batch_id,
                    "status": "active",
                },
            )
        )
    return chunks


def deduplicate_chunks(chunks: Iterable[Chunk]) -> list[Chunk]:
    seen: set[tuple[str, str, str]] = set()
    result: list[Chunk] = []

    for chunk in chunks:
        key = (
            chunk.metadata["tenantId"],
            ",".join(chunk.metadata["permissionTags"]),
            chunk.metadata["contentHash"],
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(chunk)
    return result


def validate_chunks(chunks: list[Chunk]) -> list[str]:
    errors: list[str] = []
    required = ["docId", "title", "path", "tenantId", "permissionTags", "contentHash"]

    for chunk in chunks:
        if not chunk.text.strip():
            errors.append(f"{chunk.chunk_id}: empty text")
        if len(chunk.text) > 500:
            errors.append(f"{chunk.chunk_id}: chunk too long")
        for field in required:
            if not chunk.metadata.get(field):
                errors.append(f"{chunk.chunk_id}: missing metadata {field}")
    return errors


docs = [
    Document(
        doc_id="lis-assets",
        title="LIS 数据资产目录",
        path="/assets/lis",
        content="""
        LIS 检验结果资产包含 patient_id、report_id、result_value。
        patient_id 属于高敏字段，仅信息科和数据管理员可见。
        检验报告用于临床辅助诊断，普通统计应使用脱敏结果。
        """,
        tenant_id="hospital-a",
        permission_tags={"dept:信息科", "role:data_admin"},
        version="v1",
        owner="信息科",
    )
]

all_chunks: list[Chunk] = []
for doc in docs:
    all_chunks.extend(build_chunks(doc, embedding_model="bge-m3", batch_id="batch-20260706"))

deduped = deduplicate_chunks(all_chunks)
errors = validate_chunks(deduped)

print("chunks:", len(deduped))
print("errors:", errors)
for chunk in deduped:
    print(chunk.chunk_id, chunk.metadata)

这个 Demo 没接真实向量库，但已经体现了商业 RAG 入库最重要的骨架：

1. 文本标准化。
2. 按段落合并切分。
3. 生成稳定 Chunk ID。
4. 保存租户、权限、版本、Hash、Embedding 模型、批次。
5. 按租户、权限、Hash 去重。
6. 入库前做质量校验。

生产排查流程

当用户反馈“RAG 答错了”，不要直接改 Prompt。先判断问题来自哪一层。

flowchart TD
    A["RAG 答错或越权"] --> B["根据 requestId 查日志"]
    B --> C["确认用户身份和权限过滤条件"]
    C --> D{"正确文档是否存在且 active"}
    D -- "否" --> E["补文档、修同步、检查删除状态"]
    D -- "是" --> F{"正确 Chunk 是否入库"}
    F -- "否" --> G["检查解析、清洗、切分、入库批次"]
    F -- "是" --> H{"检索是否召回正确 Chunk"}
    H -- "否" --> I["检查 Embedding、TopK、关键词、Rerank"]
    H -- "是" --> J{"答案是否被引用支持"}
    J -- "否" --> K["收紧 Prompt、引用校验、拒答策略"]
    J -- "是" --> L["检查资料冲突和业务口径"]

排查证据：

证据	用途
requestId	串起一次线上请求
userId/tenantId/roles	判断权限过滤是否正确
permissionFilter	判断是否误放或误拦
retrievedChunkIds	判断正确片段是否召回
ingestBatchId	判断这批数据是否入库成功
contentHash	判断线上片段是否是最新内容
embeddingModel	判断查询和文档向量是否同模型
promptVersion	判断是不是 Prompt 改动导致
references	判断答案是否有来源支撑

常见坑

坑	后果	正确做法
只存文本和向量	无法权限过滤、引用、排查	设计完整元数据
更新文档只改原文	向量库仍召回旧内容	同步重切和重建向量
删除文档不删向量	下线内容仍被回答	软删除加异步物理删除
先全库检索再过滤	敏感内容已进入上下文	检索前 metadata filter
Embedding 模型混用	相似度不可靠	记录模型并全量重建
Chunk 切分只按固定长度	语义被切断	按标题、段落、条款、表格切
不做入库质量报告	上线后才发现大量失败	每批生成质量指标
直接覆盖线上索引	出错无法回滚	双索引发布和灰度切换

面试标准回答

RAG 数据治理是什么？

可以这样答：

RAG 数据治理是对知识库资料从采集、解析、清洗、脱敏、切分、元数据、权限、向量化、发布、更新、删除到评估的全生命周期管理。它的目标是保证资料来源清楚、内容干净、权限正确、版本可控、可引用、可回滚、可排查。RAG 答案质量不仅取决于模型，也取决于检索到的 Chunk 是否正确、是否最新、是否有权限、是否能引用。

RAG 为什么不能只存文本和向量？

因为文本和向量只能做相似度检索，不能支持生产需要的权限过滤、引用来源、版本控制、增量更新、删除同步和问题排查。至少要保存 docId、chunkId、标题、路径、版本、权限标签、租户、内容哈希、Embedding 模型和入库批次。

文档更新后 RAG 怎么同步？

先解析新文档并重新切分，计算每个 Chunk 的内容哈希。未变化的 Chunk 可以跳过，变化的 Chunk 重新生成 Embedding 并写入新版本，同时把旧 Chunk 标记失效或删除。发布前要做质量检查和召回评估，最好用双索引灰度切换，避免直接覆盖线上。

文档删除后为什么还可能被 RAG 回答？

因为很多系统只删除了原文，没有删除向量库里的 Chunk，或者缓存里仍保留旧召回结果。正确做法是文档删除时同步标记 Chunk 失效、删除向量、清理缓存，并在检索条件里只允许 status=active 的资料。

RAG 如何防越权？

权限过滤必须发生在检索前或检索过程中。系统根据用户租户、部门、角色、密级生成 metadata filter，只在用户有权访问的 Chunk 中检索。不能先全库检索再让模型不要泄露，因为敏感内容已经进入上下文。

关联知识点

知识点	作用
RAG知识库	理解 RAG 的离线建库和在线问答主流程
RAG流程	深入学习检索、重排、生成和引用链路
Embedding与向量库	理解文本向量化和相似度检索
向量数据库选型	理解 metadata filter、索引和压测
AI评估	建立 RAG 召回和答案质量评估集
Prompt评测	用回归评测验证 Prompt 和知识库变更
AI安全	理解权限、注入、数据泄露和输出安全
LLMOps	管理知识库发布、灰度、监控和回滚
AI面试题	查看数据治理相关标准回答

本章小结

RAG 数据治理的核心不是“多塞点文档给模型”，而是让知识库成为一个可管理的数据系统。资料要能追溯来源，Chunk 要有元数据，权限要在检索前生效，更新和删除要同步到向量库，索引发布要可评估、可灰度、可回滚。只有这些基础扎实，Prompt 和模型优化才有意义。