[ai] RAG 架構實務

RAG 架構實務

RAG（Retrieval-Augmented Generation）是目前將 LLM 與企業知識庫結合的主流架構。本文從架構全景出發，逐步拆解每個環節的設計決策、技術選型與實務經驗。

架構概覽

flowchart LR
    subgraph 建立索引["Indexing Pipeline（離線）"]
        direction LR
        Docs["文件\nPDF / MD / HTML"] --> Parse["解析\nParser"]
        Parse --> Chunk["切分\nChunking"]
        Chunk --> Embed["嵌入\nEmbedding"]
        Embed --> VDB[("Vector DB\n向量資料庫")]
    end

    subgraph 查詢流程["Query Pipeline（線上）"]
        direction LR
        Query["使用者提問"] --> QEmbed["Query\nEmbedding"]
        QEmbed --> Search["相似度搜尋\nTop-K"]
        Search --> Context["相關 Chunks\n+ Metadata"]
        Context --> Prompt["Prompt 組裝\n系統指令 + 文件 + 問題"]
        Prompt --> LLM["LLM\n生成回答"]
        LLM --> Response["回答\n附來源引用"]
    end

    VDB -.->|"語意檢索"| Search

---

為什麼需要 RAG

LLM 的知識限制

1. Training Data Cutoff — 每個 LLM 都有知識截止日期，之後發生的事它一概不知。

2. 無法存取私有資料 — 內部文件、Confluence、Slack 對話、客戶資料，LLM 完全不知道。

3. Hallucination（幻覺） — LLM 不知道答案時，會自信地編造看起來合理但完全錯誤的回答。

用戶：「我們的退貨政策是什麼？」
LLM（沒有 RAG）：「一般來說，退貨期限為 30 天...」  ← 在猜
LLM（有 RAG）：「根據公司政策文件第 3.2 條，退貨期限為 14 個工作天...」  ← 有依據

4. Fine-tuning 昂貴且不靈活 — 成本高、資料更新需重新訓練、仍無法做到 source attribution。

RAG 解決的問題

RAG 的核心思路：不要把知識塞進模型，而是在回答前先查資料。

問題	RAG 如何解決
知識截止	隨時更新知識庫，不需重新訓練模型
私有資料	將內部文件索引到 Vector DB
幻覺	限制 LLM 只根據 retrieved context 回答
來源追溯	每個回答都可以標註資料來源
成本	比 fine-tuning 便宜數個數量級

一句話總結：RAG = 給 LLM 一個可以查詢的圖書館。

---

RAG 架構全景

RAG 系統分為 Indexing（建立索引） 和 Querying（查詢） 兩個流程：

=== Indexing Pipeline（離線）===

文件（PDF, DOCX, MD...）
    ▼
文件解析（Parser） → Chunking（切分段落） → Embedding（轉向量） → Vector DB（儲存）


=== Query Pipeline（線上）===

用戶查詢 → Query Embedding → 相似度搜尋 → 取回相關 chunks
                                                ▼
                    Prompt = 系統指令 + 相關 chunks + 用戶問題
                                                ▼
                                    LLM 生成回答（附來源）

---

Step 1: 文件處理（Document Processing）

垃圾進、垃圾出 — 文件解析不好，後面一切都沒意義。

支援的格式與解析策略

格式	解析工具	難度	注意事項
Markdown	直接讀取	低	保留標題結構作為 metadata
HTML	BeautifulSoup, Trafilatura	中	需去除 nav、footer 等噪音
PDF	PyMuPDF, Unstructured	高	表格、圖片、多欄排版是大挑戰
DOCX	python-docx, Unstructured	中	注意保留標題層級
CSV/Excel	pandas	中	需決定如何把表格轉成文字

from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader
 
# 載入整個目錄
loader = DirectoryLoader("./knowledge-base/", glob="**/*.md", loader_cls=UnstructuredMarkdownLoader)
documents = loader.load()

Metadata Extraction

Metadata 讓你在向量搜尋之外做精確過濾，是 RAG 品質的關鍵：

document = {
    "content": "第三季營收較去年同期成長 15%...",
    "metadata": {
        "source": "2024-Q3-financial-report.pdf",
        "page": 5,
        "section": "營收分析",
        "document_type": "financial_report",
        "quarter": "2024-Q3",
    }
}

Metadata 的用途：過濾搜尋範圍、來源追溯、權限控制、除錯追蹤。

---

Step 2: Chunking 策略

Chunking 是 RAG 系統中最關鍵的步驟。Bad chunking = bad RAG。

Chunking 方法比較

方法	描述	優點	缺點	適用場景
Fixed Size	每 N 個字元切一段	簡單、可預測	會切斷語意單元	快速原型
Sentence	按句子邊界切分	尊重語言結構	大小不均勻	短文件
Paragraph	按段落切分	自然的語意單元	段落可能太大或太小	結構良好的文件
Recursive	階層式切分	平衡語意與大小	實作較複雜	通用推薦
Semantic	根據主題變化切分	最佳語意品質	慢、需額外 ML 模型	高品質需求

實際範例

假設有以下文件：

# 公司差旅政策
 
## 交通
員工出差可搭乘高鐵或飛機。國內出差以高鐵為主，航程超過 4 小時可搭乘飛機。
搭乘飛機限經濟艙，主管級以上可搭乘商務艙。
 
## 住宿
住宿費用上限為每晚 NT$3,000。台北地區上限為 NT$4,000。
需於出差前 3 個工作天提出申請，經主管核准後方可預訂。

Fixed Size（chunk_size=100） — 會把「經濟艙」切成兩半，完全不管語意。

Recursive（推薦） — 按分隔符號層級切分，每個 chunk 都是完整語意段落：

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", " ", ""]
    # 先試段落，不行再換行，再不行用句號，最後才硬切
)

Semantic — 計算相鄰句子的 embedding 相似度，主題轉換時切分：

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
 
splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,
)

Chunk Size 考量

太小 (< 200 tokens)          適中 (500-1000 tokens)         太大 (> 2000 tokens)
├─────────────────────────────┼──────────────────────────────┤
失去上下文                    平衡點（sweet spot）            稀釋相關性

Overlap 的重要性：沒有 overlap 時語意會在 chunk 邊界被切斷。重疊 100-200 tokens 可確保連續性。

實務建議：

• 一般文件：chunk_size=800, overlap=200
• 技術文件：chunk_size=1000, overlap=200
• FAQ：chunk_size=300, overlap=50
• 法律文件：chunk_size=1500, overlap=300

---

Step 3: Embedding（向量嵌入）

什麼是 Embedding？

將文字轉為高維向量。語意相似的文字，向量也會相似。

"台北天氣"  → [0.12, -0.34, 0.56, ...]   (1536 維)
"臺北氣候"  → [0.11, -0.33, 0.55, ...]   ← cosine_similarity = 0.95
"公司財報"  → [-0.45, 0.67, -0.12, ...]   ← cosine_similarity = 0.12

Embedding 模型比較

模型	維度	品質	成本	中文支援
OpenAI text-embedding-3-small	1536	良好	$0.02/1M tokens	好
OpenAI text-embedding-3-large	3072	最佳	$0.13/1M tokens	好
Cohere embed-multilingual-v3	1024	良好	較低	好
BGE-M3 (local)	1024	優秀	免費	優秀
sentence-transformers (local)	768	良好	免費	需選對模型

使用範例

OpenAI Embedding

from openai import OpenAI
client = OpenAI()
 
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding
 
def get_embeddings_batch(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(input=texts, model="text-embedding-3-small")
    return [item.embedding for item in response.data]

本地 Embedding（BGE-M3，推薦中文場景）

from FlagEmbedding import BGEM3FlagModel
 
model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)
embeddings = model.encode(["什麼是 RAG 架構？", "今天天氣如何？"], batch_size=12, max_length=8192)
# embeddings["dense_vecs"]  → 用於向量搜尋
# embeddings["lexical_weights"]  → 用於關鍵字搜尋

Cloud vs Local

面向	Cloud (OpenAI, Cohere)	Local (BGE, Sentence-Transformers)
設定	API key 即可	需要 GPU、安裝模型
成本	按量計費	固定成本（硬體）
隱私	資料送出去	資料不出機房
維護	零維護	需要管理模型版本

建議：原型用 OpenAI，有隱私需求用 BGE-M3，中英混合用 Cohere multilingual。

---

Step 4: Vector Database（向量資料庫）

Vector DB 的核心功能

1. 儲存：存放數百萬個高維向量及其原文和 metadata
2. 索引：建立高效搜尋索引（HNSW、IVFFlat）
3. 搜尋：快速找到最相似的 K 個向量
4. 過濾：結合 metadata 做精確過濾

Vector DB 選型比較

資料庫	類型	部署方式	免費方案	最適合
pgvector	PostgreSQL 擴充	Self-hosted	免費	已用 PostgreSQL 的團隊
Pinecone	全託管	Cloud only	有	快速啟動、生產環境
Weaviate	開源	自架/Cloud	免費	多模態、hybrid search
Qdrant	開源	自架/Cloud	免費	高效能、進階過濾
Milvus	開源	自架/Cloud	免費	超大規模（億級）
ChromaDB	開源	本地	免費	原型開發、學習

跨參考：更多資料庫的選型分析見 資料庫全景

pgvector：最簡單的起步方式

如果你已經在用 PostgreSQL，pgvector 不需要引入新的基礎設施：

-- 安裝擴充 & 建表
CREATE EXTENSION IF NOT EXISTS vector;
 
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}',
    embedding vector(1536),
    created_at TIMESTAMP DEFAULT NOW()
);
 
-- 建立向量索引
CREATE INDEX idx_docs_embedding ON documents
USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
 
-- 相似度搜尋
SELECT content, metadata, 1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE metadata->>'type' = 'financial'
ORDER BY embedding <=> $1::vector
LIMIT 5;

Python 整合

import psycopg2
from psycopg2.extras import Json, execute_values
 
def store_documents(chunks: list[dict], embeddings: list[list[float]]):
    conn = psycopg2.connect("postgresql://user:pass@localhost:5432/ragdb")
    cur = conn.cursor()
    data = [(c["content"], Json(c["metadata"]), e) for c, e in zip(chunks, embeddings)]
    execute_values(cur, "INSERT INTO documents (content, metadata, embedding) VALUES %s",
                   data, template="(%s, %s, %s::vector)")
    conn.commit(); cur.close(); conn.close()
 
def search_similar(query_embedding, top_k=5, filters=None):
    conn = psycopg2.connect("postgresql://user:pass@localhost:5432/ragdb")
    cur = conn.cursor()
    sql = "SELECT content, metadata, 1 - (embedding <=> %s::vector) AS similarity FROM documents"
    params = [query_embedding]
    if filters:
        conditions = [f"metadata->>'{k}' = %s" for k in filters]
        sql += " WHERE " + " AND ".join(conditions)
        params.extend(filters.values())
    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params.extend([query_embedding, top_k])
    cur.execute(sql, params)
    results = [{"content": r[0], "metadata": r[1], "similarity": r[2]} for r in cur.fetchall()]
    cur.close(); conn.close()
    return results

ChromaDB：最快的原型開發

import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection("company_docs", metadata={"hnsw:space": "cosine"})
 
collection.add(
    documents=["文件內容 1", "文件內容 2"],
    metadatas=[{"source": "handbook.pdf"}, {"source": "policy.pdf"}],
    ids=["doc1", "doc2"],
)
 
results = collection.query(query_texts=["差旅費用規定"], n_results=3)

Pinecone：生產級託管方案

from pinecone import Pinecone, ServerlessSpec
pc = Pinecone(api_key="your-api-key")
pc.create_index(name="knowledge", dimension=1536, metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1"))
index = pc.Index("knowledge")
 
index.upsert(vectors=[{"id": "doc-001", "values": embedding, "metadata": {...}}])
results = index.query(vector=query_embedding, top_k=5,
                      filter={"department": {"$eq": "finance"}}, include_metadata=True)

---

Step 5: Retrieval 策略

Basic：Top-K + Similarity Threshold

def retrieve_with_threshold(query: str, top_k=10, threshold=0.7) -> list[dict]:
    query_embedding = get_embedding(query)
    results = search_similar(query_embedding, top_k=top_k)
    filtered = [r for r in results if r["similarity"] >= threshold]
    return filtered  # 空的話 → LLM 應回答「我不知道」

Advanced：Hybrid Search（向量 + BM25）

純向量搜尋有時漏掉精確關鍵字。Hybrid Search 融合兩種分數：

from rank_bm25 import BM25Okapi
import numpy as np
 
class HybridRetriever:
    def __init__(self, documents, embeddings):
        self.documents = documents
        self.embeddings = np.array(embeddings)
        self.bm25 = BM25Okapi([doc["content"].split() for doc in documents])
 
    def search(self, query: str, top_k=5, alpha=0.5):
        # alpha: 向量搜尋權重（0=純BM25, 1=純向量）
        query_emb = np.array(get_embedding(query))
        vec_scores = np.dot(self.embeddings, query_emb)
        vec_scores = (vec_scores - vec_scores.min()) / (vec_scores.max() - vec_scores.min())
        bm25_scores = self.bm25.get_scores(query.split())
        bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8)
        combined = alpha * vec_scores + (1 - alpha) * bm25_scores
        top_idx = np.argsort(combined)[::-1][:top_k]
        return [{**self.documents[i], "score": combined[i]} for i in top_idx]

Advanced：Re-ranking

先粗篩 Top-20，再用 cross-encoder 精排：

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
 
def retrieve_with_reranking(query, initial_k=20, final_k=5):
    candidates = basic_retrieve(query, top_k=initial_k)
    pairs = [(query, doc["content"]) for doc in candidates]
    scores = reranker.predict(pairs)
    for doc, score in zip(candidates, scores):
        doc["rerank_score"] = float(score)
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:final_k]

Advanced：MMR（Maximal Marginal Relevance）

避免取回的結果都在說同一件事，在相關性和多樣性之間取平衡。lambda 越高越重視相關性，越低越重視多樣性。

Metadata Filtering

# 用戶問：「2024 年第三季的財務報告中，營收成長率是多少？」
results = search_similar(
    query_embedding, top_k=5,
    filters={"type": "financial", "quarter": "2024-Q3"},
)

---

Step 6: Prompt Construction（提示詞建構）

基本 Template

def build_prompt(query: str, chunks: list[dict]) -> list[dict]:
    system = """你是一個基於知識庫回答問題的助手。請遵守以下規則：
1. 只根據提供的參考資料回答問題
2. 如果參考資料中沒有相關資訊，請明確說「根據現有資料，我無法回答這個問題」
3. 回答時請標註資料來源（文件名稱和頁碼）
4. 使用繁體中文回答"""
 
    context_parts = []
    for i, chunk in enumerate(chunks, 1):
        source = chunk["metadata"].get("source", "未知來源")
        page = chunk["metadata"].get("page", "")
        page_info = f"（第 {page} 頁）" if page else ""
        context_parts.append(f"[參考資料 {i}] 來源：{source}{page_info}\n{chunk['content']}")
 
    context = "\n\n---\n\n".join(context_parts)
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": f"參考資料：\n\n{context}\n\n---\n\n問題：{query}"},
    ]

結構為什麼重要

• System prompt 設定行為邊界（只用提供的資料回答）
• Retrieved chunks 提供事實依據
• 「不知道」處理 防止 hallucination
• Source attribution 讓用戶能驗證答案

處理「不知道」

def rag_answer(query: str) -> dict:
    results = retrieve_with_threshold(query, threshold=0.65)
    if not results:
        return {"answer": "根據現有知識庫，我無法找到相關資訊。", "sources": [], "confidence": "low"}
 
    messages = build_prompt(query, results)
    response = client.chat.completions.create(model="gpt-4o", messages=messages, temperature=0.1)
    return {
        "answer": response.choices[0].message.content,
        "sources": [r["metadata"] for r in results],
        "confidence": "high" if results[0]["similarity"] > 0.85 else "medium",
    }

---

RAG 的問題與限制

常見失敗模式

失敗模式	原因	解法
Retrieval Failure	Query 與 chunk 用詞差異大	Hybrid search、Query expansion
Generation Failure	LLM 忽略 context 用自己的知識	加強 prompt 約束、減少 context 量
Chunk 截斷	完整概念被切成兩個 chunks	增加 overlap、semantic chunking
Multi-hop Reasoning	需要多步推理跨文件	Agentic RAG、知識圖譜
資料過時	Vector DB 未更新	定時更新 pipeline、webhook 觸發

RAG 不適合的場景

場景	原因	替代方案
即時資料（股價、天氣）	RAG index 有延遲	API 直接查詢
複雜跨文件推理	單次 retrieval 不夠	Agentic RAG / Knowledge Graph
創意寫作	不需要事實根據	直接用 LLM
精確數值查詢	向量搜尋不精確	SQL 查詢

---

評估 RAG 系統

你不能改善你沒有測量的東西。RAG 評估分為 Retrieval 和 Generation 兩個維度。

Retrieval 評估

def evaluate_retrieval(queries, ground_truth, retriever, top_k=5):
    precisions, recalls, mrrs = [], [], []
    for query, relevant_ids in zip(queries, ground_truth):
        retrieved_ids = [r["id"] for r in retriever.search(query, top_k=top_k)]
        relevant_retrieved = set(retrieved_ids) & set(relevant_ids)
 
        precisions.append(len(relevant_retrieved) / top_k)  # Precision@K
        recalls.append(len(relevant_retrieved) / len(relevant_ids) if relevant_ids else 0)
 
        mrr = 0.0
        for rank, rid in enumerate(retrieved_ids, 1):
            if rid in relevant_ids:
                mrr = 1.0 / rank; break
        mrrs.append(mrr)
 
    return {"precision@k": sum(precisions)/len(precisions),
            "recall": sum(recalls)/len(recalls), "mrr": sum(mrrs)/len(mrrs)}

Generation 評估（RAGAS 框架）

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
 
eval_data = {
    "question": ["公司的差旅費用上限是多少？"],
    "answer": ["住宿費用上限為每晚 NT$3,000，台北地區為 NT$4,000。"],
    "contexts": [["住宿費用上限為每晚 NT$3,000。台北地區上限為 NT$4,000。"]],
    "ground_truth": ["住宿費用上限為每晚 NT$3,000，台北地區為 NT$4,000。"],
}
 
result = evaluate(Dataset.from_dict(eval_data),
                  metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
# {'faithfulness': 0.95, 'answer_relevancy': 0.92, 'context_precision': 0.88, 'context_recall': 0.90}

評估指標解讀

指標	含義	低分代表	改善方向
Faithfulness	回答是否忠於 context	LLM 在編造	加強 prompt 約束
Answer Relevancy	回答是否切題	回答離題	改善 retrieval + prompt
Context Precision	取回的 context 是否精準	太多不相關 chunk	調整 chunking、top_k
Context Recall	是否找到所有相關資料	遺漏重要資料	改善 embedding、hybrid search

---

實作建議

從簡單開始

Phase 1（1-2 週）：
├── pgvector（已有 PostgreSQL）或 ChromaDB（原型）
├── OpenAI text-embedding-3-small
├── Recursive chunking（chunk_size=800, overlap=200）
├── 基本 Top-K retrieval
└── 簡單的 prompt template

Phase 2（根據評估結果迭代）：
├── 加入 metadata filtering 和 hybrid search
├── 優化 chunking 策略、加入 re-ranking
└── 建立評估 pipeline

Phase 3（生產化）：
├── 監控/logging、定時更新 pipeline
├── 使用者回饋機制、A/B 測試
└── 成本優化

常見錯誤

1. 一開始就 over-engineer — 簡單 RAG 可處理 80% 場景，不需要上來就用 Milvus + 多路 retrieval
2. 忽略 chunking — 很多人花大量時間選 Vector DB，卻用最簡單的 fixed-size chunking。Chunking 才是品質瓶頸
3. 不做評估 — 即使手動標注 50 個 QA pairs，也比「感覺還行」有用得多
4. 忽略「不知道」 — 必須在 prompt 和 threshold 兩個層面都處理沒有相關資料的情況
5. 沒有 source attribution — 用戶無法驗證正確性，RAG 核心優勢就沒了

成本估算

# OpenAI 成本估算
total_chunks = 100_000  # 10,000 文件 x 10 chunks
avg_tokens = 500
 
# Embedding（一次性）
embedding_cost = (total_chunks * avg_tokens / 1_000_000) * 0.02  # ~$1.00
 
# 每次查詢
per_query = 0.00002 + 0.01  # embedding + LLM (GPT-4o) ≈ $0.01
 
# 每月 10,000 次查詢 ≈ $100

---

延伸閱讀

• AI 全景與核心概念 - LLM 基礎概念與生態系
• AI 工作流自動化 - 將 RAG 整合進自動化工作流
• 資料庫全景 - Vector DB 在資料庫全景中的定位

推薦資源

• LangChain RAG Tutorial - 最完整的 RAG 教學
• RAGAS Documentation - RAG 評估框架
• pgvector GitHub - PostgreSQL 向量擴充
• Chunking Strategies - Pinecone 的 chunking 策略指南