RAG 完整解析:檢索、重排、生成

從原理、Chunking 策略到 Retriever、Reranker 與進階 RAG 技術,理解 Agent 最強大的武器。

RAG 完整解析:檢索、重排、生成

上一篇我們給 Agent 加上了記憶,讓它能記住使用者的偏好與過去的互動。

但記憶解決的是「關於使用者」的資訊,如果 Agent 需要回答的是專業知識呢?
例如:公司內部的產品文件、最新的法律條文、某篇論文的細節。
這些知識不可能全部塞進模型的參數裡,也不可能全部放進上下文窗口。

這一篇,我們要談的是 Agent 最強大的武器:RAG(Retrieval-Augmented Generation,檢索增強生成)
我們會從原理、完整流程、Chunking 策略、Retriever 與 Reranker,一路談到進階 RAG 技術與實作。

Query → Retrieve → Rerank → Generate

一、為什麼需要 RAG?

LLM 有三個限制:

1. 知識有截止日期

模型的知識來自訓練資料,它不知道昨天發布的新聞、上個月更新的法規。

2. 無法存取私有資料

它不知道你公司的內部文件、你的產品手冊、你的客戶資料。

3. 容易產生幻覺

當模型不確定答案時,它可能生成一段聽起來很合理但完全錯誤的內容。

傳統的解法是微調(Fine-Tuning):用你自己的資料繼續訓練模型。
但微調有幾個問題:

  • 成本高:需要大量算力與標註資料。
  • 更新慢:每次資料變動都要重新訓練。
  • 無法引用來源:模型不會告訴你答案來自哪份文件。
  • 可能災難性遺忘:微調後可能失去原有的通用能力。

RAG 提供了一個更輕量、更靈活的替代方案:

不改變模型參數,而是在生成之前,先從外部知識庫檢索相關資訊,再把這些資訊放進上下文,讓模型根據它們生成回答。

這樣做的好處:

  • 知識可即時更新:只要更新知識庫,不需要重新訓練。
  • 可引用來源:每條回答都能追溯到具體文件。
  • 成本低:不需要 GPU 訓練,只需要向量檢索。
  • 減少幻覺:模型有具體的參考資料,不用憑空猜測。

RAG (Retrieve + Generate) vs Fine-Tuning (Retrain)

二、RAG 的完整流程

一個標準的 RAG 系統,可以拆成兩個階段:

階段一:索引(Indexing)

這是離線階段,在建立知識庫時執行一次。

原始文件
    ↓ 1. 載入(Load)
    ↓ 2. 切塊(Chunking)
    ↓ 3. 嵌入(Embedding)
    ↓ 4. 存入向量資料庫(Store)
向量資料庫

階段二:檢索與生成(Retrieval & Generation)

這是線上階段,每次使用者提問時執行。

使用者查詢
    ↓ 1. 查詢處理(Query Processing)
    ↓ 2. 檢索(Retrieve)
    ↓ 3. 重排(Rerank)
    ↓ 4. 生成(Generate)
最終回答

完整流程如下圖:

Documents → Chunks → Embeddings → Vector DB → Retrieve → 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-small1536OpenAI,便宜、快速
text-embedding-3-large3072OpenAI,效果更好
BGE-M31024開源,支援多語言
E5-large1024開源,英文效果好
Cohere Embed v31024商業,多語言

選擇嵌入模型時要考慮:

  • 語言:是否支援中文?
  • 維度:維度越高,表達能力越強,但儲存與檢索成本也越高。
  • 成本: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 系統的常見失敗模式與應對策略。