深入RAG架構(gòu)：分塊策略、混合檢索與重排序的工程實現(xiàn)

把一個RAG系統(tǒng)從Demo做到生產(chǎn)，中間要解決5個問題。

最初的版本就是標準版：全量文檔 embedding，向量檢索，LLM生成。演示沒出過問題，但是翻車發(fā)生在數(shù)據(jù)留存政策的時候，因為系統(tǒng)召回了兩段2廢棄條款和一段聊"員工留存"的HR文檔，然后把這三段內(nèi)容揉成了一個看似完整實則全錯的回答。

這不是檢索的問題，也不純粹是模型的問題。從分塊方式到搜索策略，從排序邏輯到異常兜底，每一層都藏著獨立的故障模式。

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教程教的就是這套，大多數(shù)RAG系統(tǒng)也停在了這一步。

問題出在哪？語義相似度不等于相關(guān)性。查"data retention policy"，embedding 模型會把"employee retention programs"也拉進來，因為它看到了詞匯上的重疊。兩個概念八竿子打不著但向量空間里靠得很近。

還有一種情況更隱蔽：召回的 chunk 確實跟主題相關(guān)但根本沒在回答你的問題。三個 chunk 都在聊數(shù)據(jù)留存可沒一個提到你要查的那條具體政策。

Demo之所以看著沒問題，是因為測試用的 query 本身就是你已經(jīng)知道答案的。

Level 2：智能分塊

多數(shù)RAG故障看著像檢索出了問題，實際上是分塊出了問題。

按固定500 token切一刀會怎樣？一份政策聲明被劈成兩半，問題在上半截，答案在下半截。上下文和結(jié)論被強行拆開。切出來的 chunk 單獨看根本讀不通。

分塊尺寸這件事比想象中關(guān)鍵得多：100–200 tokens太碎chunk缺少語境，"90天后刪除"這句話脫離了上下文根本不知道刪的是什么；1000+ tokens又太長一個 chunk 里塞了好幾個主題，檢索的時候噪聲和有效信息一把抓；300–500 tokens是個比較舒服的區(qū)間，上下文夠用主題又足夠聚焦。

但尺寸還不是最關(guān)鍵的。重疊（overlap）才是。

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=100, # This is the key
separators=["\n\n", "\n", ". ", " ", ""]
)

設(shè)100 token的重疊區(qū)，一個句子即使被切斷了，兩個相鄰 chunk 里都有它的完整內(nèi)容。原本卡在邊界上的答案，現(xiàn)在從哪一側(cè)都能檢索到。

還有一個元數(shù)據(jù)的小技巧：不要只存文本本身，把來源信息也一起存進去。

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 同時出現(xiàn)在召回結(jié)果里的時候一眼就能看得出來。Prompt 里可以加"優(yōu)先引用最新來源"，代碼里也可以在生成前直接按時間過濾。

光是這一步就解決了大約40%的檢索故障。垃圾進垃圾出——chunk 質(zhì)量上去了檢索效果自然跟著上去。

Level 3：混合搜索

假設(shè)這樣一個查詢："What's our PTO policy for employees with 5+ years tenure?"

語義搜索能找到跟休假政策沾邊的 chunk，概念上確實接近。關(guān)鍵詞搜索能精確命中包含"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 的調(diào)法：如果語料里領(lǐng)域術(shù)語多（法律、醫(yī)學、公司內(nèi)部縮寫），alpha 調(diào)低一些讓 BM25 主導；如果用戶提的是自然語言問題，alpha 調(diào)高讓語義檢索權(quán)重大一些。初始值設(shè)0.5，然后看哪些 query 掛了再微調(diào)。

BM25是很老的技術(shù)了，也沒人再專門為它寫博客了。但它能兜住純向量搜索漏掉的那些 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 和文檔一起輸入。所以拿它做全量檢索不現(xiàn)實——一萬篇文檔逐條打分太慢了。但從20個候選里精選3個？這個開銷完全可以接受。

加入 reranking 之后"正確 chunk 出現(xiàn)在前3"的命中率從68%提到了89%。其實相關(guān)的 chunk 一直被檢索到了，只是排名不夠靠前。

不過有一點要清楚：reranking 救不了爛檢索。如果正確的 chunk 根本不在那20個候選里，reranker 也變不出來。先把 Level 2 和 Level 3 做扎實。

Level 5：生產(chǎn)級RAG

前面幾個級別都在提升檢索質(zhì)量。生產(chǎn)級RAG要處理的是另一件事：檢索已經(jīng)盡力了，但還是失敗了，怎么辦？

因為它一定會失敗，用戶會問文檔里根本沒覆蓋的問題。分塊策略會漏掉某個關(guān)鍵段落?；蛘邌栴}本身就很模糊，召回的幾個 chunk 互相矛盾。

真正該問的不是"怎么杜絕檢索失敗"，而是"檢索失敗的時候，系統(tǒng)該怎么表現(xiàn)"。

護欄

上下文不夠的時候，別讓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

評估

沒法度量的東西就沒法改進。先建一組測試 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
]

每次改動跑一遍。追蹤檢索精度（拿到正確文檔了嗎）和答案準確率（關(guān)鍵事實對了嗎）。哪個指標掉了，馬上能定位到是哪一步出了問題。

做到這一步仍然會有邊緣 case。用戶的表述方式超出預期，文檔里藏著你不知道的自相矛盾。

邊緣 case 漏不了。關(guān)鍵是讓系統(tǒng)在拿不準的時候老實說"不知道"，而不是胡編一個答案。

什么時候該停

不是所有場景都需要做到 Level 5。

判斷該不該升級，看用戶反饋就行：

用戶在抱怨什么RAG就壞在哪里。

從 Level 1 開始。記錄并監(jiān)控系統(tǒng)在哪翻車，搞清楚原因之后再往上走。

這才是構(gòu)建一個真正能用的RAG系統(tǒng)的路徑。

https://avoid.overfit.cn/post/93d89f1be12b421dbbb761198960bc76

by Paolo Perrone