LangChain 進階主題：自訂 RAG Pipeline

簡介

在資訊爆炸的時代，檢索增強生成（Retrieval‑Augmented Generation，簡稱 RAG） 已成為讓大型語言模型（LLM）產出更精確、具參考依據答案的關鍵技術。LangChain 作為目前最受歡迎的 LLM 應用框架，提供了完整的 RAG 抽象層與多樣化的組件，讓開發者能快速建置搜尋、檢索、摘要與生成的端到端流程。

然而，面對不同的業務需求（例如：企業內部文件搜尋、法律條文比對、醫學文獻摘要），預設的 RAG pipeline 常常無法直接滿足。此時，我們需要自行組合檢索器、向量資料庫、文件切分器、重排模型等元件，打造符合特定需求的自訂 RAG Pipeline。本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，完整闡述如何在 LangChain 中打造屬於自己的 RAG 流程，幫助你在實務專案中發揮最大效能。

---

核心概念

1. RAG 基本流程

步驟	目的	LangChain 相關類別
文件切分	把長篇文字拆成適合向量化的片段	RecursiveCharacterTextSplitter、TokenTextSplitter
向量化	把文字轉成向量，供相似度搜尋使用	Embeddings（如 OpenAIEmbeddings）
向量儲存	儲存與索引向量，支援快速相似度檢索	FAISS, Pinecone, Chroma
檢索	依使用者查詢找出最相關的片段	VectorStoreRetriever、BM25Retriever
重排（Rerank）	進一步挑選或排序檢索結果，提高相關性	CohereRerank, LLMRerank
生成	把檢索結果與原始問題送入 LLM，產出完整答案	LLMChain, ConversationalRetrievalChain

重點：RAG 並非單一模型，而是 多個模組的協同工作。自訂 pipeline 的核心在於 自由組合 這些模組，並根據資料特性與效能需求調整每個環節的參數。

2. 為什麼要自訂？

• 資料領域特殊：醫學、金融、法律等領域有專有術語與嚴謹的引用規則，通用向量模型往往表現不佳。
• 效能需求：即時聊天需要毫秒級回應；離線報告則可以接受較長的檢索時間。
• 成本控制：不同向量資料庫與重排模型的計費方式差異巨大，根據使用量自行選擇能有效降低成本。
• 安全合規：企業內部資料不允許外部傳輸，需要自行部署離線向量資料庫與檢索服務。

3. LangChain 中的自訂組件

LangChain 以 抽象基底類別（BaseRetriever、BaseReranker、BasePromptTemplate）為核心，允許開發者 繼承並覆寫 需要的行為。例如：

from langchain.schema import BaseRetriever
from typing import List

class MyHybridRetriever(BaseRetriever):
    """同時結合向量檢索與關鍵字檢索的混合檢索器"""

    def __init__(self, vector_store, bm25):
        self.vector_store = vector_store
        self.bm25 = bm25

    def get_relevant_documents(self, query: str) -> List[Document]:
        # 向量檢索前 3 名
        vec_docs = self.vector_store.similarity_search(query, k=3)
        # BM25 前 2 名
        bm25_docs = self.bm25.get_relevant_documents(query)[:2]
        # 合併並去重
        return list({doc.metadata["source"]: doc for doc in vec_docs + bm25_docs}.values())

上述範例示範了 混合檢索（Hybrid Retrieval）的概念，接下來會在完整 pipeline 中看到它的實際應用。

4. 完整自訂 RAG Pipeline 的架構圖

User Query
   │
   ▼
PromptTemplate (可自訂 System Prompt)
   │
   ▼
HybridRetriever (Vector + BM25) ──► Reranker (Cohere)
   │                                 │
   ▼                                 ▼
Top‑k Documents (重新排序後)   ──► LLM (OpenAI / Claude / Gemini)
   │
   ▼
Answer (含來源引用)

---

程式碼範例

以下範例以 Python 為主，因為 LangChain 的官方 SDK 以 Python 為主要語言。每段程式碼均附上說明，方便你直接套用或改寫。

4.1 建立文件切分與向量化流程

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
import os

# 1️⃣ 讀取本機資料夾內的 .txt 檔案
loader = DirectoryLoader(
    path="data/company_policy",
    glob="*.txt",
    loader_cls=TextLoader,
)

documents = loader.load()                       # List[Document]

# 2️⃣ 文字切分（每段 500 tokens，重疊 50 tokens）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
)

chunks = splitter.split_documents(documents)    # List[Document]

# 3️⃣ 向量化並存入 FAISS（本地向量資料庫）
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
vector_store = FAISS.from_documents(chunks, embeddings)

# 將索引儲存至磁碟，方便下次直接載入
vector_store.save_local("faiss_index")

技巧：切分時的 chunk_overlap 能確保關鍵資訊不會因跨段而遺失，對於法律條文或程式碼說明特別重要。

4.2 建立混合檢索器（Vector + BM25）

from langchain.retrievers import BM25Retriever
from langchain.vectorstores import FAISS
from langchain.schema import Document
from typing import List

# 重新載入 FAISS 索引
vector_store = FAISS.load_local("faiss_index", embeddings)

# BM25 檢索器（基於原始文字）
bm25 = BM25Retriever.from_documents(chunks)

class HybridRetriever:
    """簡易混合檢索：向量前 3 + BM25 前 2"""

    def __init__(self, vector_store, bm25, top_k_vec=3, top_k_bm25=2):
        self.vector_store = vector_store
        self.bm25 = bm25
        self.top_k_vec = top_k_vec
        self.top_k_bm25 = top_k_bm25

    def get_relevant_documents(self, query: str) -> List[Document]:
        vec_docs = self.vector_store.similarity_search(query, k=self.top_k_vec)
        bm25_docs = self.bm25.get_relevant_documents(query)[:self.top_k_bm25]
        # 合併去重
        merged = {doc.metadata["source"]: doc for doc in vec_docs + bm25_docs}
        return list(merged.values())

hybrid_retriever = HybridRetriever(vector_store, bm25)

說明：混合檢索能彌補向量模型對於精確關鍵字匹配的不足，特別適用於「代號」或「專有詞」較多的文件。

4.3 加入重排模型（Rerank）

LangChain 已支援多種第三方重排服務，下面以 Cohere 的 rerank 為例。

from langchain.llms import Cohere
from langchain.retrievers import CohereRerank

# 初始化 Cohere Rerank（需要 API_KEY）
cohere_rerank = CohereRerank(
    model="rerank-english-v2.0",
    top_n=3,                     # 只保留最相關的 3 個結果
    api_key=os.getenv("COHERE_API_KEY")
)

def rerank_documents(query: str, docs: List[Document]) -> List[Document]:
    return cohere_rerank.compress_documents(query, docs)

小提醒：大多數重排服務都有 每月免費配額，但一次請求的 top_n 不宜過高，避免額外成本。

4.4 組合完整的 RetrievalQA Chain

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 1️⃣ LLM（OpenAI GPT‑4）
llm = OpenAI(model_name="gpt-4o-mini", temperature=0)

# 2️⃣ Prompt（自訂 System Prompt，強調引用來源）
prompt = PromptTemplate.from_template(
    """你是一位企業政策助理，請根據以下檔案內容回答使用者問題。
    請在回答中**引用來源檔名**，若無法在檔案中找到答案，請說明「資料不足」。
    
    問題: {question}
    相關文件:
    {context}
    """
)

# 3️⃣ 建立 RetrievalQA
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",                 # 直接把所有檔案塞入 Prompt
    retriever=hybrid_retriever,         # 前一步的混合檢索器
    return_source_documents=True,
    chain_type_kwargs={"prompt": prompt},
)

def answer_query(query: str):
    # 取得檢索文件
    docs = hybrid_retriever.get_relevant_documents(query)
    # 重排
    reranked = rerank_documents(query, docs)
    # 呼叫 QA Chain
    result = qa_chain({"question": query, "input_documents": reranked})
    return result["answer"], [d.metadata["source"] for d in result["source_documents"]]

# 範例呼叫
answer, sources = answer_query("請說明公司遠端工作政策的加班補償規定")
print("Answer:", answer)
print("Sources:", sources)

重點：chain_type="stuff" 代表把所有檢索結果一次性塞入 LLM，適合文件量不大（< 10 篇）。若文件量大，請改用 map_reduce 或 refine 方式分批處理。

4.5 使用自訂 PromptTemplate 產生更具結構的回應

structured_prompt = PromptTemplate.from_template(
    """以下是一段公司政策文件的摘錄，請根據使用者的問題，以 JSON 格式回傳答案與引用檔案。
    
    問題: {question}
    文件內容:
    {context}
    
    回應範例:
    {{
        "answer": "...",
        "sources": ["policy_2023.txt", "overtime_guideline.md"]
    }}
    """
)

qa_chain_structured = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=hybrid_retriever,
    chain_type_kwargs={"prompt": structured_prompt},
)

def answer_json(query: str):
    docs = hybrid_retriever.get_relevant_documents(query)
    reranked = rerank_documents(query, docs)
    result = qa_chain_structured({"question": query, "input_documents": reranked})
    # 直接把 LLM 產出的 JSON 文字轉成 dict
    import json
    return json.loads(result["answer"])

print(answer_json("公司出差報銷的上限是多少？"))

好處：使用 JSON 回傳可以讓前端或其他系統直接解析，提升自動化流程的可用性。

4.6 部署為 FastAPI 端點（可直接給前端呼叫）

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Company Policy RAG Service")

class QueryRequest(BaseModel):
    question: str

class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/qa", response_model=AnswerResponse)
async def ask_question(req: QueryRequest):
    try:
        answer, sources = answer_query(req.question)
        return AnswerResponse(answer=answer, sources=sources)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 若在本機執行
# uvicorn main:app --reload

部署提示：在正式環境建議將向量索引與模型服務放在 同一個 VPC，避免跨區域的網路延遲。

---

常見陷阱與最佳實踐

陷阱	說明	解決方案 / Best Practice
向量維度不一致	不同 Embedding 模型產出的向量長度不同，導致 FAISS 初始化失敗。	統一使用同一套 Embeddings，或在切換模型時重新建索引。
文件切分過小	切分粒度太細會產生大量碎片，增加檢索成本且降低語意完整性。	根據 token 數（而非字元）設定 chunk_size，建議 500‑1500 tokens。
檢索結果缺乏來源	LLM 回答時未附上檔案來源，無法驗證答案。	在 Prompt 中強制要求 引用來源，或使用 structured JSON 回傳。
重排成本失控	Rerank 服務每次請求都有費用，若 top_n 設太高會快速超支。	只對 前 5‑10 個檢索結果重排，並使用 批次重排（一次傳多筆）降低請求次數。
向量資料庫未持久化	每次重啟程式都重新建立索引，耗時且資料遺失。	使用 save_local / persist_directory，或部署雲端向量服務（Pinecone、Weaviate）。
LLM 輸出過長	生成的文字過長會被截斷，導致答案不完整。	設定 max_tokens 或在 Prompt 中加入 「請在 200 字以內回答」 的限制。
安全性洩漏	把機密文件直接送給外部 LLM 服務，可能違反合規。	使用 本地部署的 LLM（如 Llama‑2、Mistral）或將檢索結果先做 脫敏 處理。

建議的開發流程

1. 資料前處理：先完成文件清理、切分、向量化，確保每一步都有測試腳本。
2. 單元測試：對 HybridRetriever、rerank_documents 分別寫測試，確保輸入輸出符合預期。
3. 效能測試：使用 Benchmark（如 timeit）測量檢索 + 重排的總延遲，根據需求調整 top_k。
4. 成本監控：在 Cloud Provider 設定 API 使用上限，避免突發的高額帳單。
5. 監控與日志：將每次查詢、檢索結果、LLM 回應寫入 ElasticSearch 或 Loki，方便事後分析。

---

實際應用場景

場景	為什麼需要自訂 RAG	可能的組件配置
企業內部政策助理	資料分散在多個部門、PDF、Word，且有保密需求。	本地 FAISS + BM25 + Cohere Rerank + 本地部署的 Llama‑2。
法律文件檢索	法條、判例具有高度精確的條款號碼，必須完整引用。	RecursiveCharacterTextSplitter（保留條款編號）+ OpenAI Embeddings + HybridRetriever + LLMRerank（使用 GPT‑4）+ JSON Prompt。
醫學文獻摘要系統	文獻量大、專業術語多，需要高召回率且避免錯誤資訊。	TokenTextSplitter（以 token 為基準）+ BioBERT Embeddings + Pinecone 向量庫 + CohereRerank + OpenAI（作摘要生成）。
客服聊天機器人	需要即時回應且結合 FAQ、產品手冊。	BM25Retriever（快速關鍵字匹配）+ FAISS（補強語意）+ LLMChain（即時生成）+ FastAPI 作為微服務。
程式碼搜尋與說明	程式碼片段與說明文檔需要保持對應關係。	CodeSplitter（根據 function/class 切分）+ OpenAIEmbeddings（code‑aware）+ Chroma 向量庫 + LLMRerank（利用 GPT‑4 重新排序）+ PromptTemplate（要求回傳程式碼與說明）。

關鍵：每個場景的「資料特性」與「回應需求」決定了向量模型、檢索方式、重排模型與 Prompt 的選擇，這正是自訂 RAG Pipeline 的價值所在。

---

總結

• RAG 是 LLM 與外部知識結合的核心技術，LangChain 為我們提供了 模組化、可插拔 的架構。
• 透過 自訂切分、向量化、混合檢索、重排與 Prompt，我們可以針對不同業務需求打造高效、合規、成本可控的解決方案。
• 本文提供了 完整的程式碼範例（從資料前處理到 FastAPI 部署），以及 常見陷阱與最佳實踐，幫助你在實務專案中快速落地。
• 未來可以根據需求探索 多模態檢索（圖像、音訊）或 動態檢索（即時資料流）等進階方向，讓 RAG 的應用更廣、更深。

祝你在 LangChain 的自訂 RAG 之路上，以資料為根、以模型為翼，打造出令人驚豔的 AI 助手！ 🚀