Python RAG 應用開發實戰：從原理到生產部署

AI与大数据

什麼是 RAG？

RAG（Retrieval-Augmented Generation，檢索增強生成）是一種將外部知識檢索與大型語言模型生成相結合的架構模式。它解決了大模型固有的三大痛點：知識過時、幻覺生成、私有資料無法觸達。

為什麼 2026 年 RAG 如此重要？

大模型雖然能力強大，但存在根本性限制：

問題 說明 RAG 如何解決
知識截止 訓練資料有截止日期，無法獲取最新資訊 即時檢索最新文件
幻覺問題 模型可能編造看似合理但錯誤的內容 基於檢索到的事實生成回答
私有資料 企業內部文件模型從未見過 檢索企業知識庫後生成
成本問題 微調大模型成本極高 僅需維護向量索引
可追溯性 大模型輸出無法溯源 每個回答都有文件引用

RAG 的基本工作流程

使用者提問 → Query 向量化 → 向量資料庫檢索 → 檢索結果 + Prompt → LLM 生成回答
    │                                              │
    │              ┌───────────────────┐           │
    └─────────────→│  Embedding Model  │──────────→│
                   └───────────────────┘           │
                                                   ↓
                                          ┌──────────────┐
                                          │  LLM (GPT等) │
                                          └──────────────┘

向量嵌入（Vector Embeddings）基礎

向量嵌入是將文字轉換為高維數值向量的過程，使得語義相近的文字在向量空間中距離更近。

嵌入模型選型

模型 維度 最大長度 特點 適用場景
text-embedding-3-large 3072 8191 tokens OpenAI 最新，效能最強 高精度英文檢索
text-embedding-3-small 1536 8191 tokens 性價比高 通用英文場景
bge-large-zh-v1.5 1024 512 tokens 中文效果最佳 中文文件檢索
bge-m3 1024 8192 tokens 多語言，支援稠密+稀疏 多語言混合場景
gte-Qwen2-7B-instruct 3584 32768 tokens 超長上下文，開源最強 長文件、複雜語義
Cohere embed-v4 1024 128k tokens 多模態支援 圖文混合檢索

使用 OpenAI Embedding

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="RAG 是檢索增強生成的縮寫",
    dimensions=1536
)

embedding = response.data[0].embedding
print(f"向量維度: {len(embedding)}")  # 1536

使用開源 BGE 模型

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

embeddings = model.encode(
    ["RAG 是檢索增強生成", "向量資料庫用於儲存嵌入"],
    normalize_embeddings=True
)

print(f"向量維度: {embeddings.shape}")  # (2, 1024)

💡 使用 Base64 編解碼 工具可以檢視和除錯嵌入向量的傳輸編碼。

文件分塊策略

分塊（Chunking）是 RAG 系統中最關鍵的前處理步驟，直接影響檢索品質。

分塊方法對比

方法 原理 優點 缺點 推薦場景
固定大小分塊 按字元/token 數切分 實作簡單 可能截斷語義 日誌、結構化文字
遞迴字元分塊 按分隔符層級遞迴切分 保持段落完整 需調參 通用文件
語義分塊 按嵌入向量相似度切分 語義完整性最好 計算成本高 高品質問答
文件結構分塊 按 Markdown/HTML 標題切分 尊重文件結構 依賴文件格式 結構化文件
句子視窗 以句子為單位，保留上下文視窗 檢索精準+上下文豐富 實作複雜 精細問答

LangChain 遞迴字元分塊

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
    length_function=len
)

chunks = splitter.split_text(long_document)
print(f"分塊數量: {len(chunks)}")

語義分塊（進階）

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

splitter = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=75
)

chunks = splitter.split_text(long_document)

分塊參數調優建議

# 經驗法則：chunk_size 與 overlap 的關係
# overlap 通常為 chunk_size 的 10%-20%

# 短文件（FAQ、知識卡片）
chunk_size = 200
chunk_overlap = 20

# 中等文件（技術文件、部落格）
chunk_size = 500
chunk_overlap = 50

# 長文件（論文、法律文書）
chunk_size = 1000
chunk_overlap = 100

向量資料庫選型

主流向量資料庫對比

資料庫 類型 維度支援 持久化 過濾 分散式 適用場景
Chroma 嵌入式 任意 ✅ ✅ 基礎 ❌ 原型開發、小規模
FAISS 記憶體庫 任意 ⚠️ 手動 ❌ ❌ 高效能單機檢索
Pinecone 雲端服務 任意 ✅ ✅ 完整 ✅ 生產環境、免運維
Milvus 獨立服務 任意 ✅ ✅ 完整 ✅ 大規模企業級
Qdrant 獨立服務 任意 ✅ ✅ 完整 ✅ Rust 高效能
Weaviate 獨立服務 任意 ✅ ✅ GraphQL ✅ 混合搜尋
pgvector PG 擴充套件 ≤2000 ✅ ✅ SQL ✅ 已有 PostgreSQL

使用 Chroma（快速上手）

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)

collection.add(
    documents=["RAG 是檢索增強生成", "向量資料庫儲存嵌入向量"],
    metadatas=[{"source": "doc1"}, {"source": "doc2"}],
    ids=["id1", "id2"]
)

results = collection.query(
    query_texts=["什麼是 RAG?"],
    n_results=3
)

print(results["documents"])

使用 FAISS（高效能）

import faiss
import numpy as np

dimension = 1024
index = faiss.IndexFlatIP(dimension)

embeddings = np.random.rand(1000, dimension).astype("float32")
faiss.normalize_L2(embeddings)
index.add(embeddings)

query = np.random.rand(1, dimension).astype("float32")
faiss.normalize_L2(query)

distances, indices = index.search(query, k=5)
print(f"Top-5 索引: {indices}")
print(f"Top-5 相似度: {distances}")

使用 Pinecone（生產級）

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")
index_name = "rag-knowledge"

if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

index = pc.Index(index_name)

index.upsert(vectors=[
    {"id": "doc1", "values": [0.1] * 1536, "metadata": {"source": "wiki"}},
    {"id": "doc2", "values": [0.2] * 1536, "metadata": {"source": "blog"}}
])

results = index.query(
    vector=[0.15] * 1536,
    top_k=5,
    filter={"source": {"$eq": "wiki"}}
)

建構 RAG Pipeline

使用 LangChain 建構完整 RAG

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import TextLoader

loader = TextLoader("knowledge.txt", encoding="utf-8")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db"
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "什麼是 RAG?"})
print(result["result"])
print(f"來源文件: {[doc.metadata for doc in result['source_documents']]}")

使用 LlamaIndex 建構 RAG

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI as LlamaOpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = LlamaOpenAI(model="gpt-4o", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

documents = SimpleDirectoryReader("./data").load_data()

index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(
    similarity_top_k=4,
    response_mode="tree_summarize"
)

response = query_engine.query("RAG 的核心優勢是什麼？")
print(response)
print(f"來源節點: {[n.metadata for n in response.source_nodes]}")

檢索優化策略

結合稠密檢索（語義）和稀疏檢索（關鍵詞），效果顯著優於單一方法：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

bm25_retriever = BM25Retriever.from_documents(chunks, k=5)
vector_retriever = Chroma.from_documents(
    chunks, OpenAIEmbeddings()
).as_retriever(k=5)

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

results = ensemble_retriever.invoke("RAG 優化方法")

2. 重排序（Reranking）

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large")

query = "如何優化 RAG 檢索效果？"
candidates = ["候選文件1...", "候選文件2...", "候選文件3..."]

pairs = [[query, doc] for doc in candidates]
scores = reranker.predict(pairs)

ranked = sorted(zip(scores, candidates), reverse=True)
print(f"重排序結果: {ranked}")

3. 查詢改寫（Query Rewriting）

from langchain.prompts import ChatPromptTemplate

rewrite_template = ChatPromptTemplate.from_messages([
    ("system", "你是一個查詢改寫助手。將使用者的模糊問題改寫為更精確的檢索查詢。"),
    ("human", "原始問題: {question}\n請生成3個不同角度的檢索查詢：")
])

rewrite_chain = rewrite_template | ChatOpenAI(model="gpt-4o", temperature=0)

rewritten = rewrite_chain.invoke({"question": "RAG 怎麼用？"})
print(rewritten.content)

4. 效能基準參考

配置 檢索方式 資料集 Recall@5 MRR 延遲(ms)
基礎 RAG 稠密檢索 MS MARCO 0.72 0.58 120
+ BM25 混合 混合檢索 MS MARCO 0.81 0.67 180
+ Reranker 混合+重排 MS MARCO 0.89 0.78 350
+ Query Rewrite 全優化 MS MARCO 0.92 0.83 420
基礎 RAG 稠密檢索 中文 CMedQA 0.65 0.51 150
+ BGE 重排 稠密+重排 中文 CMedQA 0.84 0.73 400

常見錯誤與除錯

1. 向量維度不匹配

# ❌ 錯誤：嵌入模型和資料庫維度不一致
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  # 1536維
# 但向量資料庫建立時用了 1024 維

# ✅ 正確：確保維度一致
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 向量資料庫也使用 1536 維

2. 分塊過大導致檢索噪聲

# ❌ 錯誤：chunk_size 過大，一個區塊包含多個主題
splitter = RecursiveCharacterTextSplitter(chunk_size=5000)

# ✅ 正確：合理的 chunk_size
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)

3. 忽略元資料過濾

# ❌ 錯誤：全庫檢索，結果包含無關文件
results = vectorstore.similarity_search("Python 教程", k=5)

# ✅ 正確：利用元資料過濾
results = vectorstore.similarity_search(
    "Python 教程",
    k=5,
    filter={"category": "programming", "language": "zh"}
)

4. 嵌入模型與查詢語言不匹配

# ❌ 錯誤：中文文件用英文嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  # 英文為主

# ✅ 正確：中文文件用中文優化模型
from sentence_transformers import SentenceTransformer
embeddings = SentenceTransformer("BAAI/bge-large-zh-v1.5")

5. 除錯技巧

import json

def debug_retrieval(query, retriever, top_k=5):
    docs = retriever.invoke(query)
    for i, doc in enumerate(docs):
        print(f"--- 結果 {i+1} ---")
        print(f"內容: {doc.page_content[:200]}")
        print(f"元資料: {json.dumps(doc.metadata, ensure_ascii=False)}")
    return docs

# 使用 [JSON 格式化](/zh-TW/json/format) 工具檢視元資料結構
debug_retrieval("什麼是向量資料庫？", vectorstore.as_retriever(k=5))

生產部署建議

1. 架構設計

                    ┌─────────────┐
                    │  API 閘道器  │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
        ┌─────┴─────┐ ┌───┴───┐ ┌─────┴─────┐
        │  文件入庫  │ │檢索服務│ │  LLM 服務  │
        │  Pipeline  │ │       │ │           │
        └─────┬─────┘ └───┬───┘ └─────┬─────┘
              │            │            │
        ┌─────┴─────┐ ┌───┴───┐ ┌─────┴─────┐
        │ 訊息佇列   │ │向量DB │ │  模型服務  │
        │ (Celery)  │ │(Milvus)│ │ (vLLM等)  │
        └───────────┘ └───────┘ └───────────┘

2. 快取策略

import hashlib
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_embedding(text_hash: str, model: str):
    return embeddings_model.embed_query(text)

def get_embedding_with_cache(text: str, model: str = "text-embedding-3-small"):
    text_hash = hashlib.md5(text.encode()).hexdigest()
    return cached_embedding(text_hash, model)

3. 非同步文件入庫

from celery import Celery

app = Celery("rag_worker", broker="redis://localhost:6379/0")

@app.task
def ingest_document(file_path: str):
    loader = TextLoader(file_path, encoding="utf-8")
    documents = loader.load()
    chunks = splitter.split_documents(documents)
    vectorstore.add_documents(chunks)
    return {"status": "success", "chunks": len(chunks)}

4. 監控指標

import time
from dataclasses import dataclass

@dataclass
class RAGMetrics:
    retrieval_latency_ms: float
    llm_latency_ms: float
    total_latency_ms: float
    num_chunks_retrieved: int
    num_source_docs: int
    query_tokens: int
    response_tokens: int

def measure_rag_performance(query: str, qa_chain):
    start = time.time()
    result = qa_chain.invoke({"query": query})
    total = (time.time() - start) * 1000

    metrics = RAGMetrics(
        retrieval_latency_ms=0,
        llm_latency_ms=0,
        total_latency_ms=total,
        num_chunks_retrieved=len(result.get("source_documents", [])),
        num_source_docs=len(set(
            d.metadata.get("source", "")
            for d in result.get("source_documents", [])
        )),
        query_tokens=len(query),
        response_tokens=len(result["result"])
    )
    return result, metrics

常見問題 FAQ

Q1: RAG 和微調（Fine-tuning）怎麼選？

維度 RAG Fine-tuning
知識更新 即時更新文件即可 需要重新訓練
成本 低（僅向量索引） 高（GPU 訓練）
可解釋性 高（可追溯來源） 低（黑盒）
風格定製 弱 強
推薦策略 事實性問答首選 風格/格式定製首選

Q2: chunk_size 設多大合適？

通常 300-800 字元效果最佳。具體取決於文件類型：FAQ 用 200，技術文件用 500，法律文書用 1000。務必用實際資料做 A/B 測試。

Q3: 向量資料庫選哪個？

  • 原型/MVP：Chroma（零配置，嵌入式）
  • 單機高效能：FAISS + 自建過濾
  • 生產免運維：Pinecone
  • 大規模企業級：Milvus / Qdrant
  • 已有 PG 基礎設施：pgvector

Q4: 如何評估 RAG 系統品質？

使用 RAGAS 框架評估四個核心指標：

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
)

results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision]
)
print(results)

Q5: 中文 RAG 有什麼特別注意事項？

  1. 選擇中文優化的嵌入模型（BGE 系列、M3E 等）
  2. 分塊時使用中文分隔符（。！？；）
  3. 混合檢索對中文效果提升更明顯
  4. 注意中英文混合文件的處理
  5. 查詢改寫對中文口語化問題幫助很大

相關工具

總結

RAG 是目前最實用的大模型應用架構，它以較低的成本解決了大模型知識過時、幻覺和私有資料存取三大痛點。建構高品質 RAG 系統的關鍵在於：選擇合適的嵌入模型、精心設計分塊策略、選對向量資料庫、以及持續優化檢索效果。從 Chroma 原型起步，逐步引入混合檢索和重排序，最終演進到生產級架構——這是 RAG 落地的最佳路徑。

本站提供瀏覽器本地工具，免註冊即可試用 →

#RAG#Python#AI#大模型#向量检索#教程