从“应该能用”到“真的在生产可用”
我为内部文档、客户支持和合规流程都搭过 RAG 系统。大多数第一次都失败了。有些第二次也失败了。
三个月前，我为我们的政策文档做了一个。全量 embedding，接好检索，接上 GPT-4。演示很顺利。
然后法务问了它一个关于数据保留政策的问题。
系统检索出了三个 chunk（分块）。其中两个来自我们早在几年前就替换掉的 2019 年政策。另一个来自 HR 入职文档，里面的“retention”指的是员工留存，语义完全不同，但词相同。
生成的答案把过时的政策和无关的上下文混在一起——信心满满，但完全错误。当着法务团队的面。
我把这个系统重建了四次。每次重建都修了一个之前没料到的失败模式。“演示能跑”和“生产能跑”之间的差距，不止一个点，是五个层次。
五个层次
- Naive RAG（朴素 RAG）
：教程版。一上真实查询就崩。 - Smart Chunking（智能分块）
：你如何切分文档，决定了你能检索到什么。 - Hybrid Search（混合检索）
：当“语义相似”不等于“真正相关”。 - Reranking（重排）
：第二遍打分，捞回初检错过的好内容。 - Production RAG（生产级 RAG）
：当检索失败会怎样？别让 LLM 即兴发挥。
好，开始。
等级 1：朴素 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（数据保留政策）”，会捞到“employee retention programs（员工留存项目）”的 chunk，因为 embedding 看到了词面重合。概念不相关，但向量很近。
你还会捞到“关于”正确主题、但并不“回答”问题的 chunk。三段都在讲数据保留，但都没提你需要的具体政策点。
如果你的演示能跑，是因为你测的都是你本来就知道答案的查询。
等级 2：智能分块（Chunking）
大多数 RAG 失败看起来像检索失败，其实是分块失败。
如果每 500 个 token 就切一刀，你会把一个政策条文拦腰切断。问题在一个 chunk，答案在另一个。上下文和结论被拆开。你会做出无法独立成段的 chunk。
chunk 尺寸比你想的更重要：
-
太小（100–200 个 token）：缺上下文。“90 天后删除”如果不说明“删什么”，就没意义。 -
太大（1000+ 个 token）：一个 chunk 涵盖多个主题。检索时信号与噪声一起进来。 -
甜蜜点（300–500 个 token）：上下文够用，又足够聚焦。
但尺寸不是关键。关键是 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 都包含它。落在边界上的答案，现在从两侧都能检到。
元数据技巧（metadata trick）： 别只存文本。把来源也存上。
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 年的 chunk 和 2024 年的 chunk 一起捞出来时，你一眼就能看见。你的 prompt 可以说“优先使用较新的来源”，或者你的代码可以在生成前先做过滤。
（我在这篇里更详细地讲了检索系统的数据源选择：here。）
光做这些，就修掉了大约 40% 的检索失败。垃圾进，垃圾出。更好的 chunk 带来更好的检索。
等级 3：混合检索（Hybrid Search）
查询：“5 年以上工龄员工的带薪休假（PTO）政策是什么？”
语义检索会找到一般性的休假政策。概念上相似。
关键词检索会找到包含“5+ years”和“tenure”的段落。完全匹配。
单独任一者都找不到准确答案。合在一起就能。
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，从失败的查询中调整
这不花哨。没人现在还写 BM25 博客了。但它能抓住纯向量检索漏掉的失败，尤其当你的用户会输入他们期望精确命中的短语时。
等级 4：重排（Reranking）
你检索到了 5 个 chunk。它们都“关于”这个话题。但哪些是真的“回答了这个问题”？
Embedding 相似度是独立计算的。每个文档各自针对查询打分。Reranker（交叉编码器）把查询和文档放在“一起”，问的是：“这个文档能回答这个问题吗？”
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。它需要把查询和文档一起看。这让它不适合做初检（你不可能给一万篇文档都打一次交叉分）。但用来对 20 个候选里选前 3？刚刚好。
我在内部文档上的测试中，把“正确 chunk 进 top 3”的命中率从 68% 提升到了 89%。检索其实找到了相关 chunk，只是没把它们排在最前面。
注意：重排救不了糟糕的初检。如果正确的 chunk 根本不在 20 个候选里，重排也浮不出来。先把等级 2 和 3 做好。
等级 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]
date_warning = ""
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.]"
# 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
评估
不能衡量就无法改进。构建一个有标准答案的查询测试集。
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
]
每次改动都跑一遍。跟踪检索准确率（有没有拿到对的文档？）和回答准确率（答案里有没有包含正确事实？）。当某个指标掉了，你就知道是哪一环出问题了。
即便这样，还是会有边界情况。用户的提问方式总能出乎意料。文档里也可能有你不知道的矛盾。
你一定会错过一些边界情况。确保系统承认它不知道，而不是胡编乱造。
何时该止步
不是每个用例都需要做到等级 5。
如何判断是否需要升一个等级：
还记得那个给法务演示吗？我的 RAG 系统把过时政策和无关上下文自信地混在一起的那次？
他们现在每天都在用这个系统。系统在没把握时会拒答。当来源冲突或过时时会标注。面对模糊问题，会先澄清而不是硬猜。
我重建了四次。每一次都修了一个我第一次没预见到的失败模式。流程就是这样：搭、踩雷、理解原因、升级。
从等级 1 开始。观察它在哪里崩。只有当你搞清楚“为什么崩”时，再往上走。
这就是如何把 RAG 做到真的能用。
正在用 RAG 搭东西？
如果你在把检索接到 Agent 里，我的 LangGraph 指南覆盖了编排这一侧： The Complete Guide to Building Your First AI Agent with LangGraph(https://medium.com/data-science-collective/the-complete-guide-to-building-your-first-ai-agent-its-easier-than-you-think-c87f376c84b2)。这篇被阅读了 10 万+，从基础图到生产部署都讲到了。