1. 从一次线上事故说起为什么重排序和护栏必须一起做去年冬天我接手了一个企业知识库问答项目底层用的是标准 RAG 架构文档切片、向量化、检索、拼 prompt、丢给大模型生成。上线第一周效果看着还行第二周开始陆续有用户反馈答非所问第三周直接出了个更尴尬的事——有人问我们公司年假怎么算模型把隔壁部门一份废弃的草稿制度给答了出来还答得头头是道。排查下来问题出在两个地方。第一向量检索召回了 20 条候选但真正相关的那条排在第 14 位模型在长上下文里被前面一堆弱相关内容带偏了。第二没有任何机制去校验模型输出的内容是否真的来自检索到的文档模型自由发挥了一段看起来合理但完全错误的制度说明。这两个问题一个属于检索质量一个属于输出可信度。前者靠重排序Rerank解决后者靠护栏Guardrail解决。而 TypeSafe 这个思路的价值在于它把AI 判断这件事从祈祷模型别乱来变成了可编程、可测试、可断言的能力。这篇博文就把这条链路从头到尾拆一遍包括 BM25 和向量检索怎么配合、重排序模型怎么选、护栏怎么写成代码而不是写成 prompt、以及 TypeSafe 式的类型约束到底解决了什么工程问题。适合谁看如果你正在做 RAG 项目、被召回率和幻觉折磨过、或者想把 LLM 的输出接进生产系统但不敢接这篇应该能给你一套可以直接抄的骨架。如果你只是刚听说 RAG 是什么也没关系我会把每个概念用生活化的方式讲清楚。2. RAG 链路的真实瓶颈到底在哪2.1 向量检索不是万能药BM25 也不是老古董很多人做 RAG 的第一反应是上向量数据库觉得语义检索一定比关键词检索强。我实测下来的结论是两者是互补的不是替代的。向量检索擅长的是意思相近但用词不同的场景。比如用户问怎么请假文档里写的是休假申请流程向量能把它们拉到一起。但它有个致命弱点对精确匹配不敏感。用户问工号 A12345 的报销标准向量检索很可能召回一堆报销相关的泛泛文档却漏掉那条精确包含工号的记录。BM25 正好相反。它是基于词频和逆文档频率的经典算法对精确词项匹配极其敏感。用户问BM25 参数 k1 和 b 怎么调BM25 能精准命中包含这些术语的文档但它理解不了这个词和那个词是一个意思。所以成熟的做法是混合检索Hybrid Search向量检索和 BM25 各召回一批然后用 RRFReciprocal Rank Fusion或者加权分数融合。RRF 的公式很简单score(d) Σ 1 / (k rank_i(d))其中 k 通常取 60rank_i(d) 是文档 d 在第 i 路检索里的排名。这个公式的好处是不需要归一化两路分数向量的余弦相似度和 BM25 的分数根本不在一个量级只看排名鲁棒性很好。我踩过的坑是一开始用加权求和融合结果 BM25 分数动辄几十上百向量相似度只有 0.7 左右权重怎么调都不对。换成 RRF 之后融合逻辑一下子干净了。2.2 重排序把差不多相关和真的相关分开混合检索解决了召回不全的问题但没解决排序不准的问题。召回 50 条里可能有 5 条真正相关但它们的排名可能散落在第 3、第 11、第 27、第 40 位。如果只取 Top 5 喂给模型很可能只捞到 1 条真正相关的。重排序Rerank就是干这个的。它的原理是用一个交叉编码器Cross-Encoder把 query 和每个候选文档拼在一起送进模型直接输出一个相关性分数。这和向量检索的**双编码器Bi-Encoder**有本质区别维度双编码器向量检索交叉编码器重排序编码方式query 和 doc 分别编码query 和 doc 拼接后一起编码交互程度无交互只算向量距离全交互token 级别注意力精度中等高速度快可预计算慢必须实时算适用阶段召回精排因为交叉编码器太慢不可能对全库做所以标准流程是召回 50-100 条 → 重排序取 Top 5-10 → 喂给 LLM。这个两阶段设计是整个 RAG 质量的分水岭。我实测过一个中文技术文档库纯向量检索的 Top 5 命中率大概 62%加上重排序之后能到 85% 以上。这个提升幅度比换更大的 embedding 模型或者调 chunk size 都来得直接。2.3 护栏不是加个 prompt 就完事重排序解决了喂给模型的东西对不对但没解决模型吐出来的东西对不对。护栏要管的是后者。很多人对护栏的理解就是在 system prompt 里写一句只根据提供的文档回答。我试过没用。模型该编还是编尤其是当检索内容不完整的时候它会自动脑补一段看起来合理的答案。真正的护栏应该是可编程的校验层在模型输出之后、返回给用户之前做一系列断言检查。比如输出里的每个事实性陈述是否能在检索文档里找到支撑输出是否包含检索文档里没有出现的数字、日期、专有名词输出的引用标记是否指向真实存在的文档片段输出是否触发了某些业务规则比如涉及金额、法务、医疗的敏感表述这些检查如果写成 prompt模型会假装遵守写成代码才是真的强制执行。这就是 TypeSafe 思路的核心把 AI 判断变成可编程能力。3. TypeSafe 思路让 AI 输出带上类型约束3.1 什么是 TypeSafe为什么它对 RAG 重要TypeSafe 这个词在编程语言里指的是类型安全——编译器在运行前就能检查出类型错误而不是等到运行时崩溃。把这个思路搬到 LLM 上意思是不要让模型自由输出一段文本然后祈祷它是对的而是让模型输出一个符合预定义 schema 的结构化对象由代码来校验。举个具体例子。传统 RAG 的输出是这样的根据公司制度员工年假为每年 15 天入职满 3 年增加到 20 天。TypeSafe 的输出是这样的{ answer: 员工年假为每年 15 天入职满 3 年增加到 20 天。, claims: [ { text: 员工年假为每年 15 天, source_chunk_id: doc_hr_001_chunk_3, confidence: 0.92 }, { text: 入职满 3 年增加到 20 天, source_chunk_id: doc_hr_001_chunk_4, confidence: 0.88 } ], has_unverified_claim: false }这个结构一出来护栏就有活干了遍历 claims检查每个 source_chunk_id 是否真实存在于本次检索结果里检查 confidence 是否低于阈值检查 has_unverified_claim 是否为 true。任何一项不通过就走降级策略比如返回未找到确切依据请咨询 HR。这就是可编程能力的含义AI 的判断结果变成了代码可以消费、可以断言、可以测试的数据结构而不是一段只能靠人眼看的自然语言。3.2 用 Pydantic 或 Zod 定义输出 schema落地 TypeSafe 最直接的工具是 schema 校验库。Python 生态用 PydanticTypeScript 生态用 Zod。以 Pydantic 为例from pydantic import BaseModel, Field from typing import List, Optional class Claim(BaseModel): text: str Field(description一条事实性陈述) source_chunk_id: str Field(description支撑该陈述的文档片段ID) confidence: float Field(ge0.0, le1.0, description置信度) class RAGResponse(BaseModel): answer: str Field(description最终回答) claims: List[Claim] Field(description回答中的事实性陈述列表) has_unverified_claim: bool Field(description是否存在无来源支撑的陈述) refusal_reason: Optional[str] Field(defaultNone, description拒答原因)然后把这个 schema 通过 function calling 或者 structured output 的方式传给模型。现在主流的大模型 API 都支持 JSON schema 约束输出模型会严格按照这个结构返回。这里有个关键细节schema 里的 description 字段非常重要。模型是靠这些描述来理解每个字段该填什么的。我见过有人 schema 定义得很漂亮但 description 写得含糊结果模型填出来的 source_chunk_id 全是编的。description 要写得像给新人交代任务一样具体。3.3 护栏的三种拦截策略有了结构化输出护栏可以分三层拦截第一层结构校验。用 Pydantic 的 validator 检查 source_chunk_id 是否在本次检索的 chunk 集合里。这一层是纯代码零成本能拦掉大部分引用幻觉。第二层语义校验。对每条 claim用一个小模型或者规则引擎判断这条 claim 的文本是否真的被对应 chunk 支撑。可以用 NLI自然语言推理模型也可以用简单的关键词重叠度做粗筛。这一层成本稍高但能拦住引用对了但内容对不上的情况。第三层业务规则校验。这一层是领域相关的。比如医疗场景要检查是否出现了未经审核的用药建议金融场景要检查是否出现了具体的收益率承诺。这些规则用正则或者规则引擎写和 AI 无关但必须有。三层都过了才把 answer 返回给用户。任何一层没过走降级要么返回未找到确切依据要么返回检索到的原文让用户自己判断。4. 完整实操从零搭一条带重排序和护栏的 RAG 链路4.1 环境准备与依赖选型先说选型都是我实际用过、踩过坑之后的结论向量库中小规模用 FAISS 或 Chroma 就够上规模用 Milvus 或 Qdrant。别一上来就上分布式运维成本吃不消。Embedding 模型中文场景我推荐 BGE 系列或者 M3E英文场景可以用 text-embedding-3 系列。选型时重点看你的领域通用模型在垂直领域往往不如微调过的小模型。重排序模型BGE-Reranker 系列是性价比之选中文效果好可以本地部署。如果预算充足Cohere Rerank 的 API 也很稳。BM25用 rank_bm25 这个 Python 库或者 Elasticsearch 自带的 BM25。小规模用前者大规模用后者。Schema 校验Pydantic v2性能比 v1 好很多。LLM这个看你的合规要求和预算不展开。依赖装起来pip install pydantic faiss-cpu rank_bm25 sentence-transformers pip install FlagEmbedding # BGE 系列模型4.2 混合检索 RRF 融合的实现先建索引。文档切片我用的是固定长度加重叠chunk size 512overlap 64。这个参数不是拍脑袋定的512 大约对应 300-400 个中文字能容纳一个完整的段落overlap 64 是为了防止关键信息正好被切在边界上。from rank_bm25 import BM25Okapi import jieba # 中文分词后建 BM25 索引 tokenized_corpus [list(jieba.cut(doc)) for doc in documents] bm25 BM25Okapi(tokenized_corpus) def bm25_search(query, top_k50): tokenized_query list(jieba.cut(query)) scores bm25.get_scores(tokenized_query) top_indices scores.argsort()[-top_k:][::-1] return [(idx, scores[idx]) for idx in top_indices]向量检索那边用 FAISSimport faiss import numpy as np dimension 768 # BGE-base 的维度 index faiss.IndexFlatIP(dimension) # 内积配合归一化向量等于余弦相似度 index.add(np.array(embeddings).astype(float32)) def vector_search(query_embedding, top_k50): scores, indices index.search(query_embedding.reshape(1, -1), top_k) return list(zip(indices[0], scores[0]))RRF 融合def rrf_fusion(bm25_results, vector_results, k60): scores {} for rank, (idx, _) in enumerate(bm25_results): scores[idx] scores.get(idx, 0) 1 / (k rank 1) for rank, (idx, _) in enumerate(vector_results): scores[idx] scores.get(idx, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)这里 k60 是 RRF 论文里的推荐值我实测下来 60 和 100 差别不大但太小比如 10会让头部结果过于集中太大比如 200会让排名差异被抹平。4.3 重排序的接入与参数调优融合之后取 Top 50送进重排序模型from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16True) def rerank(query, candidates, top_k5): pairs [[query, documents[idx]] for idx, _ in candidates] scores reranker.compute_score(pairs) ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) return ranked[:top_k]调优经验候选数量召回 50 条重排到 5 条是我试过的性价比最高的配置。召回 100 条提升有限但延迟翻倍召回 20 条会漏。重排后取几条取 5 条喂给 LLM 通常够用。取太多会稀释注意力取太少可能漏关键信息。如果文档本身很长可以取 8-10 条但要做去重。分数阈值重排序分数低于某个阈值比如 0.3的候选即使排进 Top 5 也应该标记为低置信护栏层要感知这个信息。4.4 护栏层的代码实现现在把前面说的三层护栏写出来。第一层结构校验def validate_structure(response: RAGResponse, valid_chunk_ids: set): errors [] for claim in response.claims: if claim.source_chunk_id not in valid_chunk_ids: errors.append(f引用幻觉: {claim.source_chunk_id} 不在检索结果中) return errors第二层语义校验用简单的关键词重叠做粗筛生产环境建议上 NLI 模型def validate_semantic(response: RAGResponse, chunk_map: dict): errors [] for claim in response.claims: chunk_text chunk_map.get(claim.source_chunk_id, ) claim_tokens set(jieba.cut(claim.text)) chunk_tokens set(jieba.cut(chunk_text)) overlap len(claim_tokens chunk_tokens) / max(len(claim_tokens), 1) if overlap 0.3: errors.append(f语义不支撑: {claim.text[:30]}... 重叠度 {overlap:.2f}) return errors第三层业务规则以禁止出现具体金额承诺为例import re def validate_business_rules(response: RAGResponse): errors [] forbidden_patterns [ (r\d%\s*(收益|回报|保本), 禁止收益率承诺), (r保证|承诺.*(盈利|赚钱), 禁止盈利保证), ] for pattern, msg in forbidden_patterns: if re.search(pattern, response.answer): errors.append(msg) return errors三层串起来def guardrail(response: RAGResponse, valid_chunk_ids, chunk_map): all_errors [] all_errors.extend(validate_structure(response, valid_chunk_ids)) all_errors.extend(validate_semantic(response, chunk_map)) all_errors.extend(validate_business_rules(response)) if all_errors: return { status: blocked, errors: all_errors, fallback: 未找到确切依据建议查阅原始文档或咨询相关人员。 } return {status: passed, answer: response.answer}这套代码跑下来我那个项目的答非所问投诉从每周十几条降到了零。不是模型变聪明了是护栏把它的胡说八道拦在了返回之前。5. 常见问题与排查技巧实录5.1 重排序后反而变差了怎么回事这是我最常被问到的问题。重排序模型不是万能的它变差通常有三个原因原因一重排序模型和你的领域不匹配。通用重排序模型在垂直领域比如医疗、法律可能不如 BM25 准。解决办法是用领域数据微调或者干脆先用 BM25 分数做一版 baseline确认重排序确实有增益再上。原因二候选集里根本没有正确答案。重排序只能重新排列不能无中生有。如果召回阶段就漏了重排序再强也没用。这时候要回头查召回看是 embedding 模型的问题还是 chunk 切分的问题。原因三query 本身有歧义。用户问苹果是水果还是公司重排序模型没法猜。这种情况要在 query 改写阶段做消歧或者让用户澄清。排查顺序建议先看召回集里有没有正确答案人工标注 20 条 query 测一下有的话再看重排序排名没有的话回去修召回。5.2 护栏误杀太多用户体验差怎么办护栏太严会把正常回答也拦掉。我的经验是分级处理不要一刀切错误类型处理策略理由引用幻觉chunk_id 不存在直接拦截这是硬错误必须拦语义重叠度低0.2-0.3标记低置信返回但加提示可能是表述差异不一定是错语义重叠度极低0.2拦截大概率是编的业务规则命中拦截并记录合规问题零容忍另外护栏的阈值要用真实数据调不要拍脑袋。我一般会标 100 条真实 query 的输出人工判断哪些该拦哪些不该拦然后调阈值让误杀率和漏杀率都降到可接受范围。5.3 结构化输出导致模型不会说话了用了 JSON schema 约束之后有些模型会变得很死板回答干巴巴的。这是结构化输出的副作用。解决办法是分离推理和格式化两步第一步让模型自由生成一段自然语言回答不约束格式。第二步把这段回答和检索文档一起送进去让模型做抽取式结构化只负责把回答拆成 claims 并标注来源。第二步用结构化输出第一步不用。这样既保留了回答的流畅度又拿到了可校验的结构。代价是多一次模型调用但换来的是护栏能真正工作我觉得值。5.4 BM25 中文分词踩过的坑BM25 对中文必须分词但分词器选不好会出大问题。我用 jieba 默认模式踩过的坑专有名词被切碎TypeSafe被切成Type和Safe检索时匹配不上。解决办法是加自定义词典。新词识别差行业新词、产品名经常被切错。同样靠自定义词典解决。停用词干扰中文的的了是这些词在 BM25 里会贡献噪声分数。建议维护一个停用词表过滤掉。自定义词典的用法jieba.load_userdict(custom_dict.txt) # custom_dict.txt 格式每行一个词可带词频和词性 # TypeSafe 100 n # RAG 100 n这个词典要持续维护每次发现检索不准就看看是不是分词问题。5.5 重排序的延迟优化重排序是实时计算的候选 50 条的话base 模型在 CPU 上大概要 1-2 秒GPU 上 200-400 毫秒。如果延迟敏感有几个优化方向用更小的重排序模型bge-reranker-base 换成 bge-reranker-small精度损失不大但速度快一倍。减少候选数量从 50 降到 30精度损失约 2-3%速度提升 40%。批处理把 50 个 pair 一次性送进模型比逐条送快很多。FlagReranker 的 compute_score 支持列表输入一定要用。缓存相同 query 的重排序结果可以缓存尤其是高频 query。我实测下来GPU 上 50 条候选重排到 5 条端到端延迟增加约 300 毫秒对大多数场景可接受。如果实在不行可以考虑异步重排加流式返回。6. 把 AI 判断变成可编程能力的工程意义回到标题里的TypeSafe 如何把 AI 判断变成可编程能力。我理解这句话的核心是AI 的输出不应该是一个黑盒而应该是一个可以被代码消费、被测试覆盖、被监控告警的数据结构。传统 RAG 的工程问题是模型输出一段文本你只能靠人眼看、靠抽样评估。出了问题不知道是检索的锅还是生成的锅改了一个 prompt 不知道会不会影响其他 case。这是典型的不可测试系统。TypeSafe 式的改造之后情况变了可测试护栏的每一层都是纯函数可以写单元测试。给定一个 RAGResponse 和 chunk_map断言输出是 passed 还是 blocked。这些测试跑在 CI 里每次改 prompt 或换模型都能回归。可监控每条 claim 的 confidence、每个护栏的拦截率都是可以打点上报的指标。拦截率突然升高说明检索质量下降了confidence 普遍偏低说明模型对该领域不熟。可归因出了问题能精确定位。是召回漏了是重排序排错了是模型编了是护栏没拦住每一层都有日志排查时间从小时级降到分钟级。可演进想换 embedding 模型跑一遍回归测试就知道有没有退化。想加新的业务规则加一个 validator 就行不影响其他层。这套思路的代价是前期开发成本高一些要写 schema、写 validator、写测试。但一旦搭起来后续每次迭代都省心。我那个项目从上线到现在换了两次 embedding 模型、一次重排序模型、三次 prompt每次都是跑一遍测试就发版没有出过线上事故。最后分享一个我自己的习惯每次线上出现 bad case我都会把它固化成一条测试用例。护栏的测试集就是这么一条条攒起来的现在有 200 多条覆盖了各种边界情况。这个测试集比任何文档都值钱因为它记录的是真实用户踩过的坑。