RAG从离线建库到在线检索与索引发布完整管道
RAG不是“Embedding后写向量库,再取TopK”。生产RAG有两条独立但通过版本连接的管道:离线管道把有权限、有来源、可删除的原始资料发布为不可变索引版本;在线管道在用户身份和Token预算约束下改写问题、混合召回、Rerank、组装上下文,再验证答案与引用。
本页专注管道执行和故障恢复。RAG概念、Chunk和向量基础见 RAG知识库,数据所有权、生命周期和删除政策见 RAG数据治理。
学习目标
完成本页后,你应该能够:
- 画出离线建库与在线问答两条完整链路。
- 设计数据源登记、原始快照、作业状态和失败重试。
- 解释解析、清洗、脱敏、切分每一步的输入输出契约。
- 设计稳定docId、chunkId、contentHash和版本字段。
- 批量调用Embedding并处理限流、部分失败和幂等重试。
- 使用蓝绿索引、Alias和Manifest完成原子发布与回滚。
- 解释为什么ACL必须在检索前或检索中生效。
- 组合向量、关键词、元数据过滤、去重和Rerank。
- 管理Query Rewrite、多轮指代、Token预算和上下文顺序。
- 由应用生成真实引用,并在资料不足时拒答。
- 评估Recall、MRR、Rerank、Faithfulness、引用和SLO。
- 通过对账发现原文、结构化库、向量库和缓存不一致。
一、两条管道和版本边界
flowchart TD
subgraph Offline["离线建库管道"]
A["登记数据源和ACL"] --> B["抓取不可变原始快照"]
B --> C["解析/OCR/版面/表格"]
C --> D["清洗、脱敏和规范化"]
D --> E["语义切分和稳定ID"]
E --> F["批量Embedding"]
F --> G["写入候选索引版本"]
G --> H["质量校验和Manifest"]
H --> I["Alias原子发布"]
end
subgraph Online["在线问答管道"]
J["认证用户和问题"] --> K["Rewrite/实体/意图"]
K --> L["生成ACL与版本Filter"]
L --> M["向量+关键词混合召回"]
M --> N["去重、Rerank和阈值"]
N --> O["Token预算和上下文"]
O --> P["模型生成"]
P --> Q["引用、忠实度和安全校验"]
end
I -->|"publishedIndexVersion"| L在线请求必须记录实际读取的索引版本。若Alias在请求中途切换,检索层应在一次请求内使用同一版本/快照,避免召回与引用来自不同发布。
二、离线作业状态机
flowchart TD
A["DISCOVERED"] --> B["SNAPSHOTTED"]
B --> C["PARSED"]
C --> D["CLEANED"]
D --> E["CHUNKED"]
E --> F["EMBEDDING"]
F --> G["INDEXED"]
G --> H["VALIDATED"]
H --> I["PUBLISHED"]
B --> J["FAILED_RETRYABLE"]
C --> J
F --> J
J --> B
C --> K["QUARANTINED"]
D --> K每阶段保存输入引用、输出引用、版本、计数、错误分类和重试次数。不要用一个processed=true表示所有步骤,否则无法从Embedding第800批失败处恢复。
作业字段:
jobId、sourceId、tenantId
sourceRevision、snapshotHash
parserVersion、cleanerVersion、chunkPolicyVersion
embeddingModel、embeddingRevision、dimension
targetIndexVersion、status、version
attempt、nextRunAt、leaseOwner、leaseUntil
documentCount、chunkCount、embeddedCount、indexedCount
errorCode、safeError、createdAt、finishedAt三、数据源登记和原始快照
3.1 数据源登记
记录Owner、抓取方式、授权用途、租户、ACL来源、更新频率、删除语义和保留期。爬到内容不代表允许用于模型问答。
3.2 原始快照
把本次抓取的文件或数据库导出保存为不可变快照,并记录:
sourceUri
sourceRevision/ETag/updatedAt
rawObjectKey
SHA-256
mediaType
byteSize
capturedAt
permissionSnapshot解析器升级后可从同一快照重放。若只保留解析文本,无法判断错误来自原文件还是解析器。
3.3 增量发现
用业务ID、ETag、更新时间和Hash判断新增、修改、删除。更新时间不可靠时内容Hash兜底,但不能用Hash合并ACL不同的文档。
四、解析层是第一处质量上限
| 来源 | 解析重点 | 常见失败 |
|---|---|---|
| Markdown | 标题、代码块、链接 | 代码和正文混乱 |
| HTML | 主正文、标题、列表 | 菜单广告重复 |
| PDF文本 | 页码、阅读顺序、表格 | 双栏错序、字体映射 |
| 扫描PDF | OCR、方向、版面 | 小字、倾斜、印章 |
| Word | 标题、表格、批注 | 合并单元格、修订 |
| 数据库 | 主键、字段、更新时间 | 大字段、分页漂移 |
解析输出不要只有字符串:
{
"docId": "policy-1024",
"blocks": [
{
"type": "paragraph",
"page": 3,
"headingPath": ["退款制度", "到账时间"],
"text": "...",
"bbox": [80, 120, 920, 260]
}
],
"parserVersion": "pdf-parser-v4"
}页码、坐标和标题路径支撑引用与排查。解析失败文档进入隔离区,不能用空文本标记成功。
五、清洗、脱敏和规范化
5.1 清洗
删除重复页眉页脚、导航和乱码,但保留错误码、代码、表格单位和否定词。清洗前后保存计数和抽样差异。
5.2 脱敏
根据用途删除不必要的手机号、身份证、病历号、Token、密码和连接串。脱敏规则要版本化;高敏数据不应仅靠正则,需数据分类和人工抽查。
5.3 规范化
统一Unicode、空白、换行和可安全归一的标点。不能把大小写敏感代码、SQL或医学缩写全部改写。
5.4 Prompt Injection标记
文档中的“忽略规则、调用工具”是资料,不是指令。可标记风险片段并在生成阶段隔离,但真正安全边界是工具后端鉴权和最小权限。
六、Chunk切分要同时服务召回和回答
flowchart TD
A["结构化Blocks"] --> B["按标题/条款/表格语义边界"]
B --> C{"是否超过Token上限"}
C -->|"否"| D["形成候选Chunk"]
C -->|"是"| E["按段落/句子二次拆分"]
E --> D
D --> F["加入必要标题和父上下文"]
F --> G["计算contentHash和stable chunkId"]6.1 太短
丢失定义、条件和例外;检索到“不得退款”却漏掉前文适用范围。
6.2 太长
一个Chunk包含多个主题,向量语义被稀释,召回和Token成本变差。
6.3 Overlap
可保留边界上下文,但增加重复召回和索引成本。Rerank前要按docId、邻接关系和内容指纹去重。
6.4 表格和代码
表格保留标题、表头、行列关系和单位;代码保留语言、文件、函数和错误信息。不能按字符把表头与数据或函数签名与函数体拆开。
七、稳定docId和chunkId
推荐:
docId = 业务系统稳定文档ID
docVersion = 源版本/内容版本
chunkId = hash(tenantId + docId + chunkPolicyVersion + logicalPath + normalizedContent)
contentHash = hash(normalizedContent)不要每次随机UUID,否则无法:
- 判断Chunk是否变化。
- 幂等写入。
- 删除旧Chunk。
- 复用Embedding。
- 对账。
- 保持引用稳定。
内容相同但租户、ACL或文档身份不同,不能仅按contentHash合并成同一权限对象。
八、元数据和ACL
至少保存:
tenantId、docId、docVersion、chunkId、contentHash
title、sourceUri、headingPath、page/bbox
chunkIndex、language、contentType
permissionTags、departmentIds、classification
status、validFrom、validTo
parser/chunker/embedding版本
indexVersion、ingestionBatchId、createdAt向量库Metadata Filter能力必须在选型和压测中验证。只把ACL存普通数据库、先全库召回再过滤,可能泄露TopK分数、缓存和模型上下文,也会导致过滤后结果不足。
九、Embedding批处理和幂等
flowchart TD
A["选择changed chunks"] --> B["按Token/条数组成Batch"]
B --> C["预留配额并调用Embedding"]
C --> D{"结果数量和维度正确"}
D -->|"否"| E["标记批次失败并隔离"]
D -->|"是"| F["按chunkId写Embedding结果"]
F --> G["记录模型Revision、维度和Usage"]
G --> H{"还有批次"}
H -->|"是"| B
H -->|"否"| I["进入索引阶段"]重试必须按 (chunkId, contentHash, embeddingRevision) 幂等。若内容和模型未变可复用向量;Embedding模型或规范化变化通常需要重建。
处理:429 Retry-After、批次部分失败、响应数量不等、NaN/维度错误、超时和费用结算。不要把错误向量或零向量写成成功。
十、候选索引和蓝绿发布
直接写当前生产索引会暴露半成品:前10万Chunk已更新,后10万仍旧。更可靠:
flowchart TD
A["当前Alias -> index_v17"] --> B["创建候选index_v18"]
B --> C["全量/增量写入v18"]
C --> D["计数、维度、ACL和抽样检索"]
D --> E["固定评估集回归"]
E --> F{"发布门槛通过"}
F -->|"否"| G["保留报告并删除/隔离候选"]
F -->|"是"| H["原子切Alias到v18"]
H --> I["监控Canary/线上指标"]
I --> J{"是否回滚"}
J -->|"是"| K["Alias切回v17"]10.1 Manifest
发布Manifest记录文档/Chunk清单摘要、解析/切分/Embedding版本、维度、计数、ACL统计、评估报告和创建时间。
10.2 Alias切换
Alias或路由配置原子更新。应用缓存Key包含indexVersion,发布后旧缓存自然隔离。
10.3 保留旧索引
按回滚窗口保留旧版本,确认稳定后再删除。删除前检查是否有进行中请求、异步评估或审计引用。
十一、增量更新和删除
11.1 更新
发现文档新版本
→ 解析和切分
→ 比较旧/新chunkId与contentHash
→ 复用未变Embedding
→ 新增/修改Chunk入候选索引
→ 标记旧版本失效
→ 校验后发布若切分策略改变,logicalPath和边界大量变化,可能更适合全量重建。
11.2 删除
删除传播:
源系统删除/权限撤销
→ 结构化元数据标记DELETING/INACTIVE
→ 检索Filter立即排除
→ 删除/失效向量Chunk
→ 清理查询和答案缓存
→ 更新索引Manifest
→ 对账确认不可召回
→ 按政策处理原始快照和审计保留先让Filter排除可以快速止血,但最终仍需完成物理或合规要求的删除。
十二、在线入口和Query Rewrite
12.1 输入治理
认证用户、租户、场景、长度、语言和速率。用户传来的tenantId不能作为权限事实。
12.2 多轮改写
历史:
用户:Spring事务传播有哪些?
用户:第二个什么时候用?独立查询应改为:
Spring事务传播行为列表中的第二种传播行为是什么,适用于什么场景?Rewrite必须保留用户意图,不能加入历史中不存在的实体或权限。保存原问题与改写问题,评估改写是否让召回变好。
12.3 不必所有问题都改写
独立明确问题直接检索,避免额外延迟和模型误改。可由规则或轻量分类器判断是否含指代、省略或上下文依赖。
十三、ACL Filter必须先于召回结果暴露
flowchart TD
A["SecurityContext中的tenant/roles/departments"] --> B["后端构造结构化Filter"]
B --> C["向量/关键词查询携带Filter"]
C --> D["只在授权候选集合召回"]
D --> E["Rerank仍保持权限边界"]
E --> F["有权Chunk进入Prompt"]模型不能生成Filter中的用户身份;Filter使用参数化构造器,禁止把用户自由文本拼成过滤表达式。缓存Key包含权限摘要和索引版本。
十四、混合召回
14.1 向量召回
擅长语义相似:“连接池耗尽”和“数据库连接拿不到”可能接近。弱点是错误码、字段名和编号精确匹配。
14.2 关键词/BM25
擅长 ORA-01555、patient_id、接口名等精确Token。对同义改写可能弱。
14.3 融合
可以使用加权归一化分数或RRF。RRF示意:
RRF(d) = Σ 1 / (k + rank_i(d))它融合排名而非直接比较不同检索器不可比的原始分数。k和各路候选数量需要评估。
十五、去重和Rerank
先召回较多候选,例如向量和关键词各TopN,融合、按chunkId/contentHash去重,再Rerank到较少候选。
Reranker判断“这个Chunk是否能回答这个问题”,比单纯Embedding相似度更精细,但增加延迟和费用。记录召回前排名、Rerank分数和最终入选原因。
相邻Overlap Chunk可合并上下文,但要防止重复内容占满Token。
十六、阈值和无答案判断
固定相似度阈值不能跨Embedding模型通用。组合信号:
召回/Rerank分数
Top1与后续差距
是否覆盖问题关键实体
资料版本和有效期
多路检索是否一致
答案是否能被引用支持证据不足时返回“资料不足/需要澄清”,而不是让模型用参数知识补齐企业事实。拒答是正确行为,需要单独评估。
十七、Token预算和上下文组装
总窗口
- System与安全规则
- 用户问题和必要历史
- Tool定义/结果
- 输出预留
- Safety Margin
= 可用于RAG的Token预算在RAG预算中按Rerank顺序选择Chunk,去重、限制单文档占比,并保留标题、来源和引用ID。不能拼接后从尾部截断,因为可能截掉引用或完整条款。
上下文结构:
[SOURCE chunkId=c1 docId=d1 page=3]
标题路径:...
内容:...
[/SOURCE]资料文本是数据,不是系统指令。
十八、生成和引用验证
模型输入中使用内部引用ID,模型只能引用候选集合内ID。应用返回前校验:
引用ID是否存在于本次候选
用户是否有权
索引版本是否一致
引用Chunk是否实际支持对应断言
sourceUri是否可访问不要让模型自由生成URL和页码。对外引用由应用根据真实Metadata渲染。
Faithfulness检查答案中的关键断言是否被上下文支持。高风险场景可结构化提取断言并规则/人工复核;模型自评不能成为唯一证据。
十九、缓存
| 缓存 | Key必须包含 | 失效条件 |
|---|---|---|
| Query Embedding | query、embeddingRevision | 模型/规范化变化 |
| 召回结果 | tenant/ACL、query、indexVersion、参数 | 索引/权限变化 |
| Rerank | query、candidateHashes、rerankerVersion | 候选/模型变化 |
| 最终答案 | 权限、所有版本、解码和场景 | 任一事实/权限/策略变化 |
删除文档必须清理或版本隔离缓存。最终答案缓存风险最高,实时和高敏场景慎用。
二十、可运行Demo:稳定ID、ACL、混合召回和Alias发布
下面用标准库构造可运行的教学管道。它使用词集合Jaccard模拟语义分数,不是真实Embedding,但可以验证稳定ID、检索前ACL、RRF融合和蓝绿Alias语义。
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import re
def stable_hash(*parts: str) -> str:
raw = "\x1f".join(parts).encode("utf-8")
return hashlib.sha256(raw).hexdigest()[:20]
def tokenize(text: str) -> set[str]:
return set(re.findall(r"[A-Za-z0-9_-]+|[\u4e00-\u9fff]", text.lower()))
@dataclass(frozen=True)
class Chunk:
tenant_id: str
doc_id: str
chunk_id: str
content: str
acl_roles: frozenset[str]
version: str
def make_chunk(tenant: str, doc: str, logical_path: str, content: str,
roles: set[str], version: str) -> Chunk:
normalized = " ".join(content.split())
chunk_id = stable_hash(tenant, doc, "chunk-policy-v1", logical_path, normalized)
return Chunk(tenant, doc, chunk_id, normalized, frozenset(roles), version)
def vector_like_rank(query: str, chunks: list[Chunk]) -> list[str]:
q = tokenize(query)
scored = []
for chunk in chunks:
c = tokenize(chunk.content)
score = len(q & c) / len(q | c) if q | c else 0.0
scored.append((score, chunk.chunk_id))
return [chunk_id for score, chunk_id in sorted(scored, reverse=True) if score > 0]
def keyword_rank(query: str, chunks: list[Chunk]) -> list[str]:
terms = tokenize(query)
scored = []
for chunk in chunks:
lower = chunk.content.lower()
score = sum(lower.count(term) for term in terms)
scored.append((score, chunk.chunk_id))
return [chunk_id for score, chunk_id in sorted(scored, reverse=True) if score > 0]
def rrf(rankings: list[list[str]], constant: int = 60) -> list[str]:
scores: dict[str, float] = {}
for ranking in rankings:
for rank, chunk_id in enumerate(ranking, start=1):
scores[chunk_id] = scores.get(chunk_id, 0.0) + 1 / (constant + rank)
return [chunk_id for chunk_id, _ in sorted(scores.items(), key=lambda x: x[1], reverse=True)]
class IndexRegistry:
def __init__(self) -> None:
self.indices: dict[str, list[Chunk]] = {}
self.alias: dict[str, str] = {}
def build(self, version: str, chunks: list[Chunk]) -> None:
if len({chunk.chunk_id for chunk in chunks}) != len(chunks):
raise ValueError("候选索引存在重复chunkId")
self.indices[version] = list(chunks)
def publish(self, alias: str, version: str) -> None:
if version not in self.indices or not self.indices[version]:
raise ValueError("候选索引不存在或为空")
self.alias[alias] = version
def search(self, alias: str, tenant: str, roles: set[str], query: str) -> list[Chunk]:
version = self.alias[alias]
# ACL在任何打分前生效。
allowed = [
chunk for chunk in self.indices[version]
if chunk.tenant_id == tenant and bool(chunk.acl_roles & roles)
]
ids = rrf([vector_like_rank(query, allowed), keyword_rank(query, allowed)])
by_id = {chunk.chunk_id: chunk for chunk in allowed}
return [by_id[chunk_id] for chunk_id in ids]
if __name__ == "__main__":
registry = IndexRegistry()
v18 = [
make_chunk("hospital-a", "doc-1", "redis/hot-key", "Redis 热key会造成单分片高负载", {"ops"}, "18"),
make_chunk("hospital-a", "doc-2", "security/key", "数据库密码轮换流程", {"security"}, "18"),
make_chunk("hospital-b", "doc-3", "redis/hot-key", "另一个租户的Redis热key资料", {"ops"}, "18"),
]
registry.build("index_v18", v18)
registry.publish("knowledge_current", "index_v18")
results = registry.search("knowledge_current", "hospital-a", {"ops"}, "Redis 热key")
assert len(results) == 1
assert results[0].doc_id == "doc-1"
assert all(result.tenant_id == "hospital-a" for result in results)
print({"alias": registry.alias, "results": [r.chunk_id for r in results]})真实系统将模拟分数替换为Embedding/向量库、BM25和Reranker,但ACL、稳定ID、版本和Alias边界不变。
二十一、质量门槛和评估
21.1 检索
- Recall@K:正确Chunk是否在前K。
- MRR:第一个正确结果排名。
- nDCG:多级相关性的排序质量。
- ACL正确率:越权召回必须为0。
- 无答案召回噪声。
21.2 Rerank
比较Rerank前后Recall保留、MRR/nDCG、延迟和费用。Reranker可能把正确Chunk降下去。
21.3 生成
- Answer correctness。
- Faithfulness。
- Citation precision/recall。
- 拒答正确率。
- 格式和安全。
21.4 端到端
TTFT、总耗时、Token、费用、错误、缓存命中和用户反馈。每次改解析器、Chunk、Embedding、索引、Rerank、Prompt或模型都运行相同评估集。
二十二、可观测性和版本矩阵
Trace:
rag.request
├── query.rewrite
├── acl.build
├── embedding.query
├── retrieve.vector
├── retrieve.keyword
├── fusion.deduplicate
├── rerank
├── context.compose
├── model.generate
└── citation.validate记录原问题、改写版本(脱敏)、Filter摘要、候选chunkId/排名、Rerank、最终入选、Token预算、实际indexVersion和所有模型/策略版本。
二十三、对账和一致性
RAG有多个副本:源系统、原始快照、元数据数据库、对象存储、向量索引和缓存。定期对账:
源active文档数 vs 元数据active数
每文档预期Chunk vs 索引Chunk
Embedding成功数 vs 索引向量数
Manifest计数 vs 实际索引
删除清单 vs 可召回结果
ACL统计 vs Filter抽样
Alias目标 vs 发布记录发现差异后按source/doc/chunk/batch/version重建或删除,不能直接手工改一个向量后不留审计。
二十四、生产Runbook
24.1 文档已更新但仍回答旧内容
检查源Revision、抓取Job、快照Hash、docVersion、候选索引、Alias、查询实际indexVersion和缓存。不要只看原文件更新时间。
24.2 正确文档存在但没有召回
逐层检查解析文本、Chunk边界、Embedding模型、ACL Filter、向量/关键词候选、TopN、融合和Rerank。先证明正确Chunk在哪一步消失。
24.3 召回正确但答案仍错误
检查上下文是否因Token预算被截断、资料是否冲突、Prompt、模型版本、引用和Faithfulness。正确Chunk出现在Top20不代表最终进入Prompt。
24.4 发布后大面积质量下降
冻结Alias扩展,记录v旧/v新评估和在线样本,原子切回旧索引。检查Parser、Chunk、Embedding、ACL和Rerank版本,不要在新索引上边修边服务。
24.5 文档删除后仍可回答
检查status Filter、向量Chunk、Alias版本、召回/答案缓存和旧异步请求。立即Filter排除止血,再完成删除传播和对账。
24.6 出现跨租户召回
立即停止相关索引/路由并审计。检查Filter构造、缓存Key、批量入库tenantId、Alias、Rerank候选和日志访问。Prompt要求“不要泄露”不能修复权限漏洞。
24.7 Embedding批次部分失败
按batchId核对输入数量、响应数量、chunkId、维度、NaN、429和费用。只重试失败批次并保持幂等;在全部校验前不得发布候选索引。
24.8 RAG突然变慢
拆Rewrite、Embedding、Filter、向量、关键词、Rerank、上下文和模型;检查候选数、索引规模、Filter选择性、缓存、Provider和排队。TopK变大可能同时增加Rerank与生成耗时。
二十五、常见误区
写入向量库就算建库完成
错误。还需计数、ACL、维度、抽样、评估、Manifest和原子发布。
先全库召回再过滤权限也可以
错误。越权内容已进入候选、缓存和日志,过滤后还可能结果不足。
TopK越大越不容易漏
错误。噪声、Token、Rerank和模型干扰同时增加。
引用交给模型生成即可
错误。模型会编造URL和页码,应用必须绑定真实Metadata。
增量更新永远优于全量重建
错误。Embedding/Chunk策略变化和大范围修复通常需要新索引全量重建。
Alias回滚会撤销所有影响
错误。旧回答缓存、异步请求和已导出的结果仍需处理。
二十六、面试标准回答
RAG离线建库怎样保证可发布
数据源先登记授权和ACL并保存不可变快照,经过版本化解析、清洗、脱敏和语义切分生成稳定docId/chunkId;Embedding按chunkId、contentHash和模型Revision幂等批处理,写入独立候选索引。候选通过计数、维度、ACL、抽样检索和固定评估后,用Alias原子发布,旧索引保留回滚窗口。
在线RAG完整链路
认证后判断是否需要Query Rewrite,从SecurityContext构造tenant/role/department Filter,在授权候选集合中执行向量和关键词混合召回,融合去重后Rerank;按Token预算选择Chunk并带内部引用ID生成,最后校验引用、忠实度、安全和实际索引版本,证据不足则拒答。
文档更新后怎样避免半新半旧
不直接修改生产索引,而是构建新的不可变索引版本,完成全量或增量写入和评估,再原子切换Alias。一次查询固定读取同一indexVersion,缓存Key带版本;异常时Alias切回旧版本。
RAG为什么需要对账
源文件、快照、元数据、向量索引和缓存是多个副本,写入、删除和重试可能部分成功。要按docId/chunkId/batch/indexVersion定期比较active文档、Chunk、Embedding、索引和删除结果,发现差异后重建或清理。
二十七、学习验收
不看答案完成:
- 画出离线状态机和在线请求链。
- 为数据源设计快照、Hash、ACL和删除字段。
- 为PDF解析输出页码、坐标和标题路径。
- 设计稳定docId/chunkId/contentHash。
- 模拟Embedding第3批失败并幂等恢复。
- 设计index_v1到v2的Alias发布和回滚。
- 证明ACL在打分前生效。
- 手算两个排名的RRF融合。
- 为32K窗口分配RAG和输出预算。
- 设计应用生成的真实引用结构。
- 运行标准库Demo并增加越权资料验证不可召回。
- 为更新、删除、越权和发布退化执行Runbook。
