RAG 系統實戰指南：從零建構企業知識庫
RAG（Retrieval-Augmented Generation）是目前企業導入 LLM 最實用的架構，能讓 AI 基於企業內部知識回答問題，避免幻覺產生。本文將從原理到實作，完整解析如何建構生產級 RAG 系統。
什麼是 RAG？
RAG 結合了資訊檢索與文字生成：
用戶問題 → 向量檢索相關文件 → 結合文件內容生成回答
相比微調（Fine-tuning），RAG 有以下優勢：
| 比較項目 | RAG | Fine-tuning |
|---|---|---|
| 知識更新 | 即時 | 需重新訓練 |
| 成本 | 低 | 高 |
| 可解釋性 | 可追溯來源 | 黑盒子 |
| 幻覺問題 | 較少 | 可能加重 |
系統架構
完整的 RAG 系統包含以下元件：
┌─────────────────────────────────────────────────────────────┐
│ RAG 系統架構 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────┐ │
│ │ 資料來源 │ │
│ │ PDF/Word/Web │ │
│ └───────┬───────┘ │
│ │ │
│ ▼ │
│ ┌───────────────┐ ┌───────────────┐ │
│ │ 文件處理 │────▶│ Embedding │ │
│ │ Chunking │ │ 向量化 │ │
│ └───────────────┘ └───────┬───────┘ │
│ │ │
│ ▼ │
│ ┌───────────────┐ │
│ │ 向量資料庫 │ │
│ │ Pinecone │ │
│ └───────┬───────┘ │
│ │ │
│ ┌───────────────┐ │ │
│ │ 用戶問題 │◀──────────┘ │
│ └───────┬───────┘ 相似度檢索 │
│ │ │
│ ▼ │
│ ┌───────────────┐ ┌───────────────┐ │
│ │ Prompt │────▶│ LLM │ │
│ │ 組合 │ │ 生成回答 │ │
│ └───────────────┘ └───────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
文件處理
1. 文件解析
支援多種格式的文件解析：
from langchain_community.document_loaders import (
PyPDFLoader,
Docx2txtLoader,
UnstructuredHTMLLoader,
)
from pathlib import Path
class DocumentProcessor:
def __init__(self):
self.loaders = {
'.pdf': PyPDFLoader,
'.docx': Docx2txtLoader,
'.html': UnstructuredHTMLLoader,
}
def load_document(self, file_path: str) -> list:
"""載入文件並返回文件物件列表"""
suffix = Path(file_path).suffix.lower()
if suffix not in self.loaders:
raise ValueError(f"不支援的文件格式: {suffix}")
loader = self.loaders[suffix](file_path)
documents = loader.load()
# 添加 metadata
for doc in documents:
doc.metadata['source'] = file_path
doc.metadata['file_type'] = suffix
return documents
2. 文件切割（Chunking）
切割策略直接影響檢索品質：
from langchain.text_splitter import RecursiveCharacterTextSplitter
class ChunkingStrategy:
"""文件切割策略"""
@staticmethod
def recursive_split(documents: list, chunk_size: int = 1000,
chunk_overlap: int = 200) -> list:
"""遞迴字元切割 - 通用策略"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "，", " ", ""],
)
return splitter.split_documents(documents)
@staticmethod
def semantic_split(documents: list, embeddings) -> list:
"""語義切割 - 基於內容相似度"""
from langchain_experimental.text_splitter import SemanticChunker
splitter = SemanticChunker(
embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95,
)
return splitter.split_documents(documents)
切割參數選擇
| 文件類型 | chunk_size | chunk_overlap | 說明 |
|---|---|---|---|
| 技術文件 | 1000-1500 | 200-300 | 保留完整段落 |
| 法律合約 | 500-800 | 100-150 | 精確檢索 |
| 客服 FAQ | 300-500 | 50-100 | 問答配對 |
| 長篇報告 | 1500-2000 | 300-400 | 保留上下文 |
向量化（Embedding）
選擇 Embedding 模型
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
# OpenAI Embedding（準確度高，需 API 費用）
openai_embeddings = OpenAIEmbeddings(
model="text-embedding-3-small", # 或 text-embedding-3-large
)
# 本地 Embedding（免費，適合敏感資料）
local_embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
model_kwargs={'device': 'cuda'}, # 使用 GPU
)
Embedding 模型比較
| 模型 | 維度 | 中文支援 | 成本 |
|---|---|---|---|
| text-embedding-3-small | 1536 | 良好 | $0.02/1M tokens |
| text-embedding-3-large | 3072 | 良好 | $0.13/1M tokens |
| multilingual-e5-large | 1024 | 優秀 | 免費（本地） |
| bge-large-zh | 1024 | 優秀 | 免費（本地） |
向量資料庫
Pinecone 實作
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore
class VectorStoreManager:
def __init__(self, api_key: str, index_name: str):
self.pc = Pinecone(api_key=api_key)
self.index_name = index_name
def create_index(self, dimension: int = 1536):
"""建立向量索引"""
if self.index_name not in self.pc.list_indexes().names():
self.pc.create_index(
name=self.index_name,
dimension=dimension,
metric="cosine",
spec=ServerlessSpec(
cloud="aws",
region="us-east-1"
)
)
def get_vectorstore(self, embeddings) -> PineconeVectorStore:
"""取得向量資料庫實例"""
return PineconeVectorStore(
index=self.pc.Index(self.index_name),
embedding=embeddings,
text_key="text",
)
def add_documents(self, vectorstore, documents: list):
"""新增文件到向量資料庫"""
vectorstore.add_documents(documents)
本地替代方案
from langchain_community.vectorstores import Chroma
# Chroma（適合開發與小規模部署）
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db",
)
檢索策略
1. 基礎相似度檢索
def basic_retrieval(vectorstore, query: str, k: int = 5):
"""基礎向量相似度檢索"""
return vectorstore.similarity_search(query, k=k)
2. 混合檢索（Hybrid Search）
結合向量檢索與關鍵字檢索：
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
def hybrid_retrieval(documents: list, vectorstore, query: str):
"""混合檢索：向量 + BM25"""
# BM25 關鍵字檢索
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# 向量檢索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 混合（權重可調整）
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6], # BM25 40%, Vector 60%
)
return ensemble_retriever.invoke(query)
3. 重排序（Reranking）
使用 Cross-encoder 重新排序檢索結果：
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
def reranking_retrieval(vectorstore, query: str):
"""使用 Cohere Rerank 重排序"""
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
reranker = CohereRerank(
model="rerank-multilingual-v3.0",
top_n=5,
)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=base_retriever,
)
return compression_retriever.invoke(query)
回答生成
Prompt 設計
from langchain_core.prompts import ChatPromptTemplate
RAG_PROMPT = ChatPromptTemplate.from_template("""
你是一個專業的企業知識庫助手。請根據以下提供的資料回答用戶問題。
規則：
1. 只根據提供的資料回答，不要編造資訊
2. 如果資料中沒有相關內容，請明確告知
3. 回答要簡潔專業，必要時列點說明
4. 在回答最後標註資料來源
參考資料：
{context}
用戶問題：{question}
請回答：
""")
完整 RAG Chain
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
def create_rag_chain(vectorstore, model_name: str = "gpt-4"):
"""建立 RAG Chain"""
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
llm = ChatOpenAI(model=model_name, temperature=0)
def format_docs(docs):
return "\n\n".join([
f"[來源: {doc.metadata.get('source', '未知')}]\n{doc.page_content}"
for doc in docs
])
rag_chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough()
}
| RAG_PROMPT
| llm
| StrOutputParser()
)
return rag_chain
# 使用方式
chain = create_rag_chain(vectorstore)
answer = chain.invoke("公司的請假規定是什麼？")
進階優化
1. 查詢改寫
REWRITE_PROMPT = ChatPromptTemplate.from_template("""
將以下用戶問題改寫成更適合檢索的形式。
保持原意，但使用更精確的關鍵字。
原始問題：{question}
改寫後的問題：
""")
def query_rewrite(llm, question: str) -> str:
chain = REWRITE_PROMPT | llm | StrOutputParser()
return chain.invoke({"question": question})
2. 多查詢檢索
from langchain.retrievers.multi_query import MultiQueryRetriever
def multi_query_retrieval(vectorstore, llm, question: str):
"""從多個角度檢索"""
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm,
)
return retriever.invoke(question)
3. 來源引用
from langchain_core.runnables import RunnableParallel
def create_chain_with_sources(vectorstore, llm):
"""建立帶來源引用的 Chain"""
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
def format_docs_with_sources(docs):
formatted = []
sources = []
for i, doc in enumerate(docs):
formatted.append(f"[{i+1}] {doc.page_content}")
sources.append(doc.metadata.get('source', '未知'))
return {
"context": "\n\n".join(formatted),
"sources": sources
}
chain = RunnableParallel(
answer=(
{"context": retriever | format_docs_with_sources,
"question": RunnablePassthrough()}
| RAG_PROMPT
| llm
| StrOutputParser()
),
sources=retriever | (lambda docs: [d.metadata.get('source') for d in docs])
)
return chain
評估與監控
檢索品質評估
def evaluate_retrieval(vectorstore, test_queries: list):
"""評估檢索品質"""
results = []
for query_data in test_queries:
query = query_data['query']
expected_doc = query_data['expected_source']
retrieved = vectorstore.similarity_search(query, k=5)
retrieved_sources = [d.metadata.get('source') for d in retrieved]
# 計算 Hit@5
hit = expected_doc in retrieved_sources
rank = retrieved_sources.index(expected_doc) + 1 if hit else -1
results.append({
'query': query,
'hit': hit,
'rank': rank,
})
# 計算指標
hit_rate = sum(r['hit'] for r in results) / len(results)
mrr = sum(1/r['rank'] for r in results if r['hit']) / len(results)
return {'hit_rate': hit_rate, 'mrr': mrr}
結論
建構生產級 RAG 系統需要注意：
- 文件處理：選擇適當的切割策略
- Embedding：根據語言和成本選擇模型
- 檢索策略：混合檢索 + 重排序提升準確度
- Prompt 設計：引導模型正確使用檢索結果
- 持續優化：建立評估機制，迭代改進
如果你正在規劃企業知識庫或 AI 助手專案，歡迎聯繫我們討論。