RAG 完整解析:檢索、重排、生成
從原理、Chunking 策略到 Retriever、Reranker 與進階 RAG 技術,理解 Agent 最強大的武器。
RAG 完整解析:檢索、重排、生成
上一篇我們給 Agent 加上了記憶,讓它能記住使用者的偏好與過去的互動。
但記憶解決的是「關於使用者」的資訊,如果 Agent 需要回答的是專業知識呢?
例如:公司內部的產品文件、最新的法律條文、某篇論文的細節。
這些知識不可能全部塞進模型的參數裡,也不可能全部放進上下文窗口。
這一篇,我們要談的是 Agent 最強大的武器:RAG(Retrieval-Augmented Generation,檢索增強生成)。
我們會從原理、完整流程、Chunking 策略、Retriever 與 Reranker,一路談到進階 RAG 技術與實作。
一、為什麼需要 RAG?
LLM 有三個限制:
1. 知識有截止日期
模型的知識來自訓練資料,它不知道昨天發布的新聞、上個月更新的法規。
2. 無法存取私有資料
它不知道你公司的內部文件、你的產品手冊、你的客戶資料。
3. 容易產生幻覺
當模型不確定答案時,它可能生成一段聽起來很合理但完全錯誤的內容。
傳統的解法是微調(Fine-Tuning):用你自己的資料繼續訓練模型。
但微調有幾個問題:
- 成本高:需要大量算力與標註資料。
- 更新慢:每次資料變動都要重新訓練。
- 無法引用來源:模型不會告訴你答案來自哪份文件。
- 可能災難性遺忘:微調後可能失去原有的通用能力。
RAG 提供了一個更輕量、更靈活的替代方案:
不改變模型參數,而是在生成之前,先從外部知識庫檢索相關資訊,再把這些資訊放進上下文,讓模型根據它們生成回答。
這樣做的好處:
- 知識可即時更新:只要更新知識庫,不需要重新訓練。
- 可引用來源:每條回答都能追溯到具體文件。
- 成本低:不需要 GPU 訓練,只需要向量檢索。
- 減少幻覺:模型有具體的參考資料,不用憑空猜測。
二、RAG 的完整流程
一個標準的 RAG 系統,可以拆成兩個階段:
階段一:索引(Indexing)
這是離線階段,在建立知識庫時執行一次。
原始文件
↓ 1. 載入(Load)
↓ 2. 切塊(Chunking)
↓ 3. 嵌入(Embedding)
↓ 4. 存入向量資料庫(Store)
向量資料庫
階段二:檢索與生成(Retrieval & Generation)
這是線上階段,每次使用者提問時執行。
使用者查詢
↓ 1. 查詢處理(Query Processing)
↓ 2. 檢索(Retrieve)
↓ 3. 重排(Rerank)
↓ 4. 生成(Generate)
最終回答
完整流程如下圖:
三、階段一:索引(Indexing)
步驟 1:載入文件(Load)
RAG 的知識來源可以是各種格式:
- PDF、Word、Markdown
- 網頁、HTML
- 資料庫、CSV
- API 回傳的 JSON
from langchain_community.document_loaders import PyPDFLoader, TextLoader
# 載入 PDF
loader = PyPDFLoader("product_manual.pdf")
documents = loader.load()
# 載入文字檔
loader = TextLoader("faq.txt", encoding="utf-8")
documents = loader.load()
步驟 2:切塊(Chunking)
文件通常太長,無法整份放進上下文。
我們需要把它切成較小的塊(Chunk),每塊是一個獨立的檢索單位。
Chunking 是 RAG 中最關鍵、也最容易被忽略的環節。
切得太粗,檢索不精準;切得太細,失去上下文。
常見的 Chunking 策略
1. 固定長度切塊(Fixed-size Chunking)
最簡單的做法:每 N 個字元切一塊,可設定重疊(Overlap)。
def fixed_size_chunk(text: str, chunk_size: int = 500, overlap: int = 50):
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
優點:簡單、可控。
缺點:可能在句子中間切斷,破壞語意。
2. 遞迴切塊(Recursive Chunking)
按照層級切割:先按段落,再按句子,最後按字元。
這是 LangChain 的 RecursiveCharacterTextSplitter 的做法。
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", "!", "?", ",", " ", ""],
)
chunks = splitter.split_text(text)
優點:盡量保持語意完整。
缺點:chunk 大小不固定。
3. 語意切塊(Semantic Chunking)
用嵌入模型計算句子之間的相似度,在語意轉折處切開。
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
splitter = SemanticChunker(
OpenAIEmbeddings(),
breakpoint_threshold_type="percentile",
)
chunks = splitter.split_text(text)
優點:每個 chunk 語意連貫。
缺點:需要額外的嵌入計算,成本較高。
4. 文件結構切塊(Document-based Chunking)
如果文件有明確結構(如 Markdown 標題、程式碼函式),按結構切。
from langchain.text_splitter import MarkdownHeaderTextSplitter
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
)
chunks = splitter.split_text(markdown_text)
Chunking 的最佳實踐
- Chunk size:通常 200 到 1000 個 token。太小失去上下文,太大檢索不精準。
- Overlap:通常為 chunk size 的 10% 到 20%,避免切斷重要資訊。
- 保留 metadata:每個 chunk 都應該記錄來源、頁碼、標題等資訊。
- 考慮查詢類型:如果查詢是「這個產品的保固期是多久」,chunk 應該包含完整的保固條款。
# 保留 metadata 的範例
chunks_with_metadata = []
for i, chunk in enumerate(chunks):
chunks_with_metadata.append({
"text": chunk,
"metadata": {
"source": "product_manual.pdf",
"chunk_index": i,
"total_chunks": len(chunks),
}
})
步驟 3:嵌入(Embedding)
把每個 chunk 轉成向量,這是檢索的基礎。
from openai import OpenAI
client = OpenAI()
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list:
response = client.embeddings.create(
model=model,
input=text,
)
return response.data[0].embedding
常見的嵌入模型:
| 模型 | 維度 | 特點 |
|---|---|---|
| text-embedding-3-small | 1536 | OpenAI,便宜、快速 |
| text-embedding-3-large | 3072 | OpenAI,效果更好 |
| BGE-M3 | 1024 | 開源,支援多語言 |
| E5-large | 1024 | 開源,英文效果好 |
| Cohere Embed v3 | 1024 | 商業,多語言 |
選擇嵌入模型時要考慮:
- 語言:是否支援中文?
- 維度:維度越高,表達能力越強,但儲存與檢索成本也越高。
- 成本:API 呼叫的價格。
- 速度:嵌入計算與檢索的速度。
步驟 4:存入向量資料庫(Store)
import chromadb
client = chromadb.PersistentClient(path="./rag_db")
collection = client.get_or_create_collection(name="knowledge_base")
# 批次寫入
collection.add(
documents=[c["text"] for c in chunks_with_metadata],
metadatas=[c["metadata"] for c in chunks_with_metadata],
ids=[f"chunk_{i}" for i in range(len(chunks_with_metadata))],
)
四、階段二:檢索與生成
步驟 1:查詢處理(Query Processing)
使用者的查詢通常很短、很模糊。
我們可以對查詢做一些處理,提升檢索效果。
查詢改寫(Query Rewriting)
用 LLM 把模糊的查詢改寫成更明確的檢索查詢。
REWRITE_PROMPT = """請將以下使用者問題改寫成適合檢索的查詢。
使用者問題:{query}
請輸出 2-3 個不同的檢索查詢,每行一個。
只輸出查詢,不要其他文字。"""
def rewrite_query(query: str) -> list:
result = call_llm([
{"role": "user", "content": REWRITE_PROMPT.format(query=query)}
])
return [q.strip() for q in result.strip().split("\n") if q.strip()]
查詢擴展(Query Expansion)
加入同義詞或相關詞,提升召回率。
HyDE(Hypothetical Document Embeddings)
先讓 LLM 生成一個「假設性的答案」,再用這個答案去檢索。
因為答案的用詞通常比問題更接近文件內容。
HYDE_PROMPT = """請根據以下問題,寫一段假設性的回答。
問題:{query}
回答:"""
def hyde_retrieve(query: str, collection, n_results: int = 5):
# 生成假設性答案
hypothetical = call_llm([
{"role": "user", "content": HYDE_PROMPT.format(query=query)}
])
# 用假設性答案去檢索
results = collection.query(
query_texts=[hypothetical],
n_results=n_results,
)
return results
步驟 2:檢索(Retrieve)
最常見的是向量檢索:把查詢轉成向量,找出最相似的 chunk。
def vector_retrieve(query: str, collection, n_results: int = 5):
results = collection.query(
query_texts=[query],
n_results=n_results,
)
return results
但向量檢索不是萬能的。
它擅長捕捉語意相似性,但對於精確關鍵字(如產品型號、人名)可能不夠好。
所以實務上常用混合檢索(Hybrid Search):
- 向量檢索:捕捉語意相似性。
- 關鍵字檢索(BM25):捕捉精確詞彙匹配。
然後把兩者的結果合併,用 RRF(Reciprocal Rank Fusion) 等演算法排序。
def hybrid_retrieve(query: str, collection, bm25_index, n_results: int = 5):
# 向量檢索
vector_results = collection.query(
query_texts=[query],
n_results=n_results * 2,
)
# BM25 關鍵字檢索
bm25_results = bm25_index.search(query, top_k=n_results * 2)
# 用 RRF 合併
combined = reciprocal_rank_fusion(
[vector_results["ids"][0], bm25_results],
k=60,
)
return combined[:n_results]
步驟 3:重排(Rerank)
檢索出來的 chunk 不一定都相關。
Reranker 是一個更精確的模型,用來重新排序檢索結果。
它與嵌入模型的差別在於:
- 嵌入模型:分別把查詢和文件轉成向量,計算相似度。速度快,但精度有限。
- Reranker:把查詢和文件一起輸入模型,直接輸出相關性分數。速度慢,但精度高。
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
def rerank(query: str, documents: list, top_k: int = 3):
pairs = [(query, doc) for doc in documents]
scores = reranker.predict(pairs)
# 按分數排序
ranked = sorted(
zip(documents, scores),
key=lambda x: x[1],
reverse=True,
)
return [doc for doc, _ in ranked[:top_k]]
常見的 Reranker:
| 模型 | 特點 |
|---|---|
| BAAI/bge-reranker-v2-m3 | 開源,多語言,效果好 |
| Cohere Rerank | 商業,支援多語言 |
| cross-encoder/ms-marco-MiniLM-L-12-v2 | 開源,英文 |
步驟 4:生成(Generate)
把檢索到的 chunk 放進上下文,讓 LLM 根據它們生成回答。
RAG_PROMPT = """請根據以下參考資料回答問題。
參考資料:
{context}
問題:{query}
請遵守以下規則:
1. 只根據參考資料回答,不要使用外部知識。
2. 如果參考資料不足以回答,請明確說明。
3. 回答時請引用來源,例如 [文件1]。
回答:"""
def generate_answer(query: str, retrieved_chunks: list) -> str:
# 組裝上下文
context = "\n\n".join(
f"[文件{i+1}] {chunk}"
for i, chunk in enumerate(retrieved_chunks)
)
prompt = RAG_PROMPT.format(context=context, query=query)
return call_llm([
{"role": "user", "content": prompt}
])
五、完整 RAG 實作
把上面的片段整合起來:
import chromadb
from openai import OpenAI
client = OpenAI()
class RAGSystem:
def __init__(self, collection_name: str = "knowledge_base"):
self.chroma = chromadb.PersistentClient(path="./rag_db")
self.collection = self.chroma.get_or_create_collection(
name=collection_name
)
# ========== 索引階段 ==========
def index_documents(self, documents: list, chunk_size: int = 500, overlap: int = 50):
"""把文件切塊、嵌入、存入向量資料庫"""
all_chunks = []
all_metadatas = []
for doc_id, doc in enumerate(documents):
chunks = fixed_size_chunk(doc["text"], chunk_size, overlap)
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
"source": doc.get("source", f"doc_{doc_id}"),
"chunk_index": i,
})
# 批次嵌入
embeddings = []
for chunk in all_chunks:
emb = get_embedding(chunk)
embeddings.append(emb)
# 存入向量資料庫
self.collection.add(
documents=all_chunks,
embeddings=embeddings,
metadatas=all_metadatas,
ids=[f"chunk_{i}" for i in range(len(all_chunks))],
)
print(f"已索引 {len(all_chunks)} 個 chunk")
# ========== 檢索階段 ==========
def retrieve(self, query: str, n_results: int = 5) -> list:
"""檢索相關 chunk"""
results = self.collection.query(
query_texts=[query],
n_results=n_results,
)
return results["documents"][0] if results["documents"] else []
# ========== 生成階段 ==========
def answer(self, query: str, top_k: int = 3) -> str:
"""完整 RAG 流程"""
# 1. 檢索
chunks = self.retrieve(query, n_results=top_k * 2)
if not chunks:
return "抱歉,知識庫中沒有相關資訊。"
# 2. 重排
reranked = rerank(query, chunks, top_k=top_k)
# 3. 生成
return generate_answer(query, reranked)
# ========== 使用範例 ==========
rag = RAGSystem()
# 索引文件
documents = [
{"text": "本產品保固期為兩年,自購買日起算。...", "source": "warranty.txt"},
{"text": "若產品在保固期內故障,請聯繫客服...", "source": "support.txt"},
]
rag.index_documents(documents)
# 查詢
answer = rag.answer("這個產品的保固期是多久?")
print(answer)
六、進階 RAG 技術
標準 RAG 已經很強大,但仍有改進空間。以下是幾個進階技術:
1. Multi-Query RAG
用 LLM 把一個問題改寫成多個不同角度的查詢,分別檢索,再合併結果。
def multi_query_rag(query: str, rag: RAGSystem, n_queries: int = 3) -> str:
# 生成多個查詢
queries = rewrite_query(query)
# 分別檢索
all_chunks = []
for q in queries:
chunks = rag.retrieve(q, n_results=3)
all_chunks.extend(chunks)
# 去重
unique_chunks = list(dict.fromkeys(all_chunks))
# 重排與生成
reranked = rerank(query, unique_chunks, top_k=3)
return generate_answer(query, reranked)
2. Parent-Child RAG
檢索時用小 chunk 提高精度,生成時用大 chunk(父文件)提供更多上下文。
# 索引時:同時存小 chunk 和大 chunk
# 檢索時:用小 chunk 檢索,但回傳對應的大 chunk
def parent_child_retrieve(query: str, collection, n_results: int = 3):
# 用小 chunk 檢索
results = collection.query(
query_texts=[query],
n_results=n_results,
)
# 取得對應的父文件
parent_ids = [meta["parent_id"] for meta in results["metadatas"][0]]
parents = collection.get(ids=parent_ids)
return parents["documents"]
3. Graph RAG
用知識圖譜來組織文件之間的關係,適合需要多跳推理的問題。
問題:A 公司的執行長是誰?他畢業於哪所大學?
標準 RAG:可能只找到 A 公司執行長的資訊,找不到大學。
Graph RAG:A 公司 → 執行長 → 張三 → 畢業於 → 台大
4. Self-RAG
讓模型自己決定「是否需要檢索」、「檢索結果是否相關」、「回答是否有根據」。
def self_rag(query: str, rag: RAGSystem) -> str:
# 1. 判斷是否需要檢索
need_retrieval = call_llm([
{"role": "user", "content": f"問題「{query}」是否需要檢索外部知識?回答 yes 或 no。"}
])
if "no" in need_retrieval.lower():
return call_llm([{"role": "user", "content": query}])
# 2. 檢索
chunks = rag.retrieve(query)
# 3. 判斷每個 chunk 是否相關
relevant_chunks = []
for chunk in chunks:
is_relevant = call_llm([
{"role": "user", "content": f"以下資料是否與問題「{query}」相關?\n\n{chunk}\n\n回答 yes 或 no。"}
])
if "yes" in is_relevant.lower():
relevant_chunks.append(chunk)
# 4. 生成
if not relevant_chunks:
return "找不到相關資訊。"
return generate_answer(query, relevant_chunks)
5. Contextual Retrieval
在嵌入每個 chunk 之前,先用 LLM 為它加上一段上下文說明。
原始 chunk:
「本產品保固期為兩年。」
加上上下文後:
「這段文字來自產品保固條款。本產品保固期為兩年。」
這樣可以避免 chunk 脫離上下文後語意不清。
七、RAG 的評估
要怎麼知道 RAG 系統好不好?我們需要評估兩個階段:
檢索評估
- Recall@K:正確的 chunk 是否在前 K 個檢索結果中?
- Precision@K:前 K 個檢索結果中,有多少是相關的?
- MRR(Mean Reciprocal Rank):第一個相關結果的排名倒數平均。
生成評估
- Faithfulness:回答是否忠於檢索到的內容?
- Answer Relevancy:回答是否與問題相關?
- Context Precision:檢索到的內容是否精準?
- Context Recall:是否檢索到所有需要的資訊?
常用的評估框架:
- RAGAS:專門為 RAG 設計的評估框架。
- TruLens:提供 RAG 三元組評估。
- DeepEval:通用的 LLM 評估框架。
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
result = evaluate(
dataset=eval_dataset,
metrics=[faithfulness, answer_relevancy, context_precision],
)
print(result)
八、RAG 的最佳實踐
1. Chunking 是關鍵
- 不要用固定長度盲目切塊。
- 盡量保持語意完整。
- 保留 metadata。
- 考慮查詢類型來設計 chunk 大小。
2. 混合檢索優於單一檢索
- 向量檢索 + BM25 關鍵字檢索。
- 用 RRF 合併結果。
3. Reranker 能大幅提升精度
- 檢索階段多召回一些(如 top 20)。
- 用 Reranker 精選出 top 3-5。
4. Prompt 要明確
- 要求模型只根據參考資料回答。
- 要求模型引用來源。
- 要求模型在資料不足時明確說明。
5. 監控與迭代
- 記錄每次查詢與檢索結果。
- 定期評估檢索品質。
- 根據失敗案例調整 Chunking 與檢索策略。
6. 處理無答案的情況
- 不是所有問題都能從知識庫找到答案。
- 要求模型在資料不足時說「我不知道」,而不是亂編。
九、總結:RAG 讓 Agent 擁有專業知識
讓我們回顧這一篇的核心:
- 為什麼需要 RAG:LLM 知識有截止日期、無法存取私有資料、容易幻覺。
- RAG vs 微調:RAG 更輕量、可即時更新、可引用來源。
- 索引階段:載入 → 切塊 → 嵌入 → 存入向量資料庫。
- 檢索生成階段:查詢處理 → 檢索 → 重排 → 生成。
- Chunking 策略:固定長度、遞迴、語意、文件結構。
- 檢索方式:向量檢索、關鍵字檢索、混合檢索。
- Reranker:用 Cross-Encoder 重新排序,提升精度。
- 進階技術:Multi-Query、Parent-Child、Graph RAG、Self-RAG、Contextual Retrieval。
- 評估:Recall@K、Precision@K、Faithfulness、Answer Relevancy。
- 最佳實踐:好的 Chunking、混合檢索、Reranker、明確 Prompt、監控迭代。
RAG 是 Agent 從「通用助手」變成「領域專家」的關鍵技術。
有了 RAG,Agent 不再只能回答訓練資料中的問題,而是能根據最新的、私有的、專業的知識來回答。
不過,RAG 只是 Agent 的其中一個能力。
當多個 Agent 需要協作時,我們還需要一套機制來讓它們分工合作。
這就是下一篇要談的主題:多 Agent 協作。
下一篇預告
《多 Agent 協作:角色分工、辯論與 Swarm》
我們會解釋為什麼需要多個 Agent、如何設計角色與通訊協議、辯論式推理與投票機制,以及多 Agent 系統的常見失敗模式與應對策略。