把一个 RAG 系统从 Demo 做到生产，中间要解决 5 个问题。
最初的版本就是标准版：全量文档 embedding，向量检索，LLM 生成。演示没出过问题，但是翻车发生在数据留存政策的时候，因为系统召回了两段废弃条款和一段聊'员工留存'的 HR 文档，然后把这三段内容揉成了一个看似完整实则全错的回答。
这不是检索的问题，也不纯粹是模型的问题。从分块方式到搜索策略，从排序逻辑到异常兜底，每一层都藏着独立的故障模式。
Level 1：Naive RAG
文档做 embedding，存向量，按相似度取 top-k，丢给模型生成。流程就这么简单：
from openai import OpenAI
import chromadb
client = OpenAI()
chroma = chromadb.Client()
collection = chroma.create_collection("docs")
def index_document(doc_id: str, text: str):
response = client.embeddings.create(
model="text-embedding-3-small", input=text
)
collection.add(
ids=[doc_id],
embeddings=[response.data[0].embedding],
documents=[text]
)
def naive_rag(query: str, k: int = 3) -> str:
# Embed query
query_embedding = client.embeddings.create(
model="text-embedding-3-small", input=query
).data[0].embedding
# Retrieve
results = collection.query(
query_embeddings=[query_embedding], n_results=k
)
# Generate
context = "\n\n".join(results["documents"][0])
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"Answer based on this context:\n\n{context}"},
{"role": "user", "content": query}
]
)
return response.choices[0].message.content
所有 RAG 教程教的就是这套，大多数 RAG 系统也停在了这一步。
问题出在哪？语义相似度不等于相关性。查"data retention policy"，embedding 模型会把"employee retention programs"也拉进来，因为它看到了词汇上的重叠。两个概念八竿子打不着但向量空间里靠得很近。
还有一种情况更隐蔽：召回的 chunk 确实跟主题相关但根本没在回答你的问题。三个 chunk 都在聊数据留存可没一个提到你要查的那条具体政策。
Demo 之所以看着没问题，是因为测试用的 query 本身就是你已经知道答案的。
Level 2：智能分块
多数 RAG 故障看着像检索出了问题，实际上是分块出了问题。
按固定 500 token 切一刀会怎样？一份政策声明被劈成两半，问题在上半截，答案在下半截。上下文和结论被强行拆开。切出来的 chunk 单独看根本读不通。
分块尺寸这件事比想象中关键得多：100–200 tokens 太碎 chunk 缺少语境，"90 天后删除"这句话脱离了上下文根本不知道删的是什么；1000+ tokens 又太长一个 chunk 里塞了好几个主题，检索的时候噪声和有效信息一把抓；300–500 tokens 是个比较舒服的区间，上下文够用主题又足够聚焦。
但尺寸还不是最关键的。重叠（overlap）才是。
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=100, # This is the key
separators=["\n\n", "\n", ". ", " ", ""]
)
设 100 token 的重叠区，一个句子即使被切断了，两个相邻 chunk 里都有它的完整内容。原本卡在边界上的答案，现在从哪一侧都能检索到。
还有一个元数据的小技巧：不要只存文本本身，把来源信息也一起存进去。
def chunk_with_metadata(doc: str, source: str, doc_date: str) -> list[dict]:
chunks = splitter.split_text(doc)
return [
{
"text": chunk,
"source": source,
"date": doc_date,
"section": extract_section_header(chunk),
}
for chunk in chunks
]
这样当 2019 年和 2024 年的 chunk 同时出现在召回结果里的时候一眼就能看得出来。Prompt 里可以加"优先引用最新来源"，代码里也可以在生成前直接按时间过滤。
光是这一步就解决了大约 40% 的检索故障。垃圾进垃圾出——chunk 质量上去了检索效果自然跟着上去。
Level 3：混合搜索
假设这样一个查询：'What's our PTO policy for employees with 5+ years tenure?'
语义搜索能找到跟休假政策沾边的 chunk，概念上确实接近。关键词搜索能精确命中包含"5+ years"和"tenure"的 chunk。
单独用哪一个都不够。两路合并就可以了。
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
def __init__(self, documents: list[str]):
self.documents = documents
self.embeddings = self._embed_all(documents)
# BM25 for keyword matching
tokenized = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(tokenized)
def _embed_all(self, docs: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-small", input=docs
)
return [d.embedding for d in response.data]
def search(self, query: str, k: int = 5, alpha: float = 0.5) -> list[str]:
# Semantic scores (normalized)
q_emb = client.embeddings.create(
model="text-embedding-3-small", input=query
).data[0].embedding
sem_scores = np.dot(self.embeddings, q_emb)
sem_scores = (sem_scores - sem_scores.min()) / (sem_scores.max() - sem_scores.min() + 1e-8)
# BM25 scores (normalized)
bm25_scores = np.array(self.bm25.get_scores(query.lower().split()))
if bm25_scores.max() > 0:
bm25_scores = bm25_scores / bm25_scores.max()
# Combine: alpha controls semantic vs keyword weight
combined = alpha * sem_scores + (1 - alpha) * bm25_scores
top_k = np.argsort(combined)[::-1][:k]
return [self.documents[i] for i in top_k]
alpha 的调法：如果语料里领域术语多（法律、医学、公司内部缩写），alpha 调低一些让 BM25 主导；如果用户提的是自然语言问题，alpha 调高让语义检索权重大一些。初始值设 0.5，然后看哪些 query 挂了再微调。
BM25 是很老的技术了，也没人再专门为它写博客了。但它能兜住纯向量搜索漏掉的那些 case，尤其是用户输入的恰好是文档里的原始表述时。
Level 4：Reranking
检索回来 5 个 chunk，跟主题都沾边。但哪些真正在回答问题？
Embedding 相似度是单独算的，每份文档独立跟 query 打分。Reranker 不一样——它把 query 和文档放在一起看，问的是：'这份文档是不是在回答这个问题？'
from sentence_transformers import CrossEncoder
class RerankedRetriever:
def __init__(self, documents: list[str]):
self.hybrid = HybridRetriever(documents)
self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def search(self, query: str, k: int = 3) -> list[str]:
# Get 20 candidates (cheap, fast)
candidates = self.hybrid.search(query, k=20)
# Rerank with cross-encoder (expensive, accurate)
pairs = [(query, doc) for doc in candidates]
scores = self.reranker.predict(pairs)
# Return top k after reranking
reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in reranked[:k]]
Cross-encoder 没办法预先算好文档 embedding，必须 query 和文档一起输入。所以拿它做全量检索不现实——一万篇文档逐条打分太慢了。但从 20 个候选里精选 3 个？这个开销完全可以接受。
加入 reranking 之后"正确 chunk 出现在前 3"的命中率从 68% 提到了 89%。其实相关的 chunk 一直被检索到了，只是排名不够靠前。
不过有一点要清楚：reranking 救不了烂检索。如果正确的 chunk 根本不在那 20 个候选里，reranker 也变不出来。先把 Level 2 和 Level 3 做扎实。
Level 5：生产级 RAG
前面几个级别都在提升检索质量。生产级 RAG 要处理的是另一件事：检索已经尽力了，但还是失败了，怎么办？
因为它一定会失败，用户会问文档里根本没覆盖的问题。分块策略会漏掉某个关键段落。或者问题本身就很模糊，召回的几个 chunk 互相矛盾。
真正该问的不是"怎么杜绝检索失败"，而是"检索失败的时候，系统该怎么表现"。
护栏
上下文不够的时候，别让 LLM 自己编。
Air Canada 在这件事上付出了代价——他们输了一场官司，原因是聊天机器人编造了一条根本不存在的退款政策。
def guarded_rag(query: str, retriever, min_score: float = 0.6) -> str:
results = retriever.search_with_scores(query, k=3)
# Check: Do we have ANY confident results?
top_score = results[0][1] if results else 0
if top_score < min_score:
return (
"I don't have enough information to answer that confidently. "
"Could you rephrase, or is there a specific document I should look at?"
)
# Check: Are sources from different time periods?
dates = [r["date"] for r, _ in results]
if len(set(dates)) > 1:
newest = max(dates)
if any(d < newest for d in dates):
date_warning = "\n\n[Note: Some sources are older. The most recent policy takes precedence.]"
else:
date_warning = ""
# Generate with explicit grounding instruction
context = "\n\n---\n\n".join([r["text"] for r, _ in results])
response = client.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "system",
"content": f"""Answer based ONLY on the provided context. If the context doesn't contain enough information, say so explicitly. Never infer or make up information not directly stated. Context: {context}"""
},
{"role": "user", "content": query}
]
)
return response.choices[0].message.content + date_warning
评估
没法度量的东西就没法改进。先建一组测试 query，每条都带上已知的正确答案：
test_cases = [
{
"query": "What's our data retention policy for customer records?",
"must_retrieve": ["data-retention-policy-2024.md"],
"answer_must_contain": ["7 years", "deletion request"],
"answer_must_not_contain": ["2019", "employee retention"]
},
# ... 50+ more cases covering your actual use cases
]
每次改动跑一遍。追踪检索精度（拿到正确文档了吗）和答案准确率（关键事实对了吗）。哪个指标掉了，马上能定位到是哪一步出了问题。
做到这一步仍然会有边缘 case。用户的表述方式超出预期，文档里藏着你不知道的自相矛盾。
边缘 case 漏不了。关键是让系统在拿不准的时候老实说"不知道"，而不是胡编一个答案。
什么时候该停
不是所有场景都需要做到 Level 5。
判断该不该升级，看用户反馈就行：
用户在抱怨什么 RAG 就坏在哪里。
从 Level 1 开始。记录并监控系统在哪翻车，搞清楚原因之后再往上走。
这才是构建一个真正能用的 RAG 系统的路径。
