EN
07 / 07第 5 章：建立基本 RAG 流程 · 5 分鐘 · 2026-10-05

5.7 動手實踐：端到端 RAG 系統

一個約一百行的完整 RAG 程式，從文件到答案，可以直接複製執行。

RAG

下面把載入文件、分塊、建立索引、檢索、組提示詞和呼叫模型接成一個可執行的檔案。範例刻意維持單純，方便逐步查看每個中間結果。

先說執行狀況

  • 載入、分塊、嵌入、存入、檢索、組提示詞：我實際跑過，下面的輸出是真的。
  • 呼叫 LLM（generate 的 API 那一段）：我沒有執行，因為沒有設定金鑰。沒有金鑰時，程式會印出「送給模型的提示詞」就結束。有金鑰的話請自己測試。
  • 範例文件的內容都是虛構的。

完整程式

存成 rag_demo.py：

"""最小的端到端 RAG：載入 → 分塊 → 嵌入 → 存入 → 檢索 → 組提示詞 → 生成"""
import os

import chromadb
from sentence_transformers import SentenceTransformer

# ---------- 0. 設定 ----------
EMBED_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"
TOP_K = 3
MAX_DISTANCE = 0.5  # 餘弦距離超過這個值，就當作「找不到相關資料」
LLM_MODEL = "claude-sonnet-5-5"  # 換成你要用的模型

# ---------- 1. 載入：兩份範例文件（內容是虛構的） ----------
DOCS = {
    "員工手冊.md": """# 員工手冊

## 特別休假

員工在同一公司服務滿六個月未滿一年者，每年有三天特別休假。滿一年未滿兩年者，有七天。滿兩年未滿三年者，有十天。

特別休假的日期，由員工和公司協商安排。年度結束時沒有休完的天數，公司應該折算成工資發給員工。

## 加班規定

加班必須事先經主管同意。平日加班前兩小時，以平日每小時工資的 1.34 倍計算。第三小時起，以 1.67 倍計算。
""",
    "報帳規定.md": """# 報帳規定

## 差旅費

出差的住宿費，每晚上限新台幣 2,500 元。交通費以實支實付為原則，搭乘高鐵須附上票根。

## 報帳期限

出差結束後 14 天內必須完成報帳。逾期未報帳者，須先向主管說明原因。
""",
}


# ---------- 2. 分塊：依標題切，並記下標題路徑 ----------
def heading_chunks(text):
    chunks, path, body = [], [], []

    def flush():
        content = "\n".join(body).strip()
        if content:
            chunks.append({"headings": " > ".join(path), "text": content})
        body.clear()

    for line in text.splitlines():
        if line.startswith("#"):
            flush()
            level = len(line) - len(line.lstrip("#"))
            path[:] = path[: level - 1] + [line.lstrip("# ").strip()]
        else:
            body.append(line)
    flush()
    return chunks


# ---------- 3. 建立索引 ----------
embedder = SentenceTransformer(EMBED_MODEL)
collection = chromadb.Client().create_collection("knowledge-base", metadata={"hnsw:space": "cosine"})

ids, texts, metas, to_embed = [], [], [], []
for source, content in DOCS.items():
    for i, chunk in enumerate(heading_chunks(content)):
        ids.append(f"{source}-{i}")
        texts.append(chunk["text"])
        metas.append({"source": source, "headings": chunk["headings"]})
        to_embed.append(f"{chunk['headings']}\n{chunk['text']}")  # 標題放進內容再嵌入

collection.add(ids=ids, embeddings=embedder.encode(to_embed).tolist(), documents=texts, metadatas=metas)
print(f"索引完成：{len(texts)} 塊\n")


# ---------- 4. 檢索器 ----------
def retrieve(question, k=TOP_K):
    result = collection.query(query_embeddings=embedder.encode([question]).tolist(), n_results=k)
    hits = []
    for text, meta, dist in zip(result["documents"][0], result["metadatas"][0], result["distances"][0]):
        if dist <= MAX_DISTANCE:
            hits.append({"text": text, "distance": dist, **meta})
    return hits


# ---------- 5. 提示詞 ----------
SYSTEM = """你是公司的行政助理。請只根據「參考資料」回答問題。
規則：
1. 資料裡沒有答案時，回答「資料中沒有提到」，不要猜。
2. 用到哪一段資料，就在句子後面標上編號，例如 [1]。
3. 用繁體中文，簡短回答。"""


def build_prompt(question, hits):
    blocks = [f"[{n}]（{h['source']} ＞ {h['headings']}）\n{h['text']}" for n, h in enumerate(hits, 1)]
    return "# 參考資料\n" + "\n\n".join(blocks) + f"\n\n# 問題\n{question}"


# ---------- 6. 生成器：有 API 金鑰就呼叫真的模型，沒有就只示範到提示詞 ----------
def generate(prompt):
    if os.environ.get("ANTHROPIC_API_KEY"):
        import anthropic

        reply = anthropic.Anthropic().messages.create(
            model=LLM_MODEL,
            max_tokens=500,
            system=SYSTEM,
            messages=[{"role": "user", "content": prompt}],
        )
        return reply.content[0].text
    return "（沒有設定 API 金鑰，這裡不呼叫模型，只顯示送給模型的提示詞）"


# ---------- 7. 串起來 ----------
def ask(question):
    print("=" * 50)
    print("問題：", question)
    hits = retrieve(question)
    if not hits:
        print("檢索：沒有夠相關的資料 → 直接回答「資料中沒有提到」，不呼叫模型")
        return
    for n, h in enumerate(hits, 1):
        print(f"檢索 [{n}] 距離 {h['distance']:.3f}｜{h['source']} ＞ {h['headings']}")
    prompt = build_prompt(question, hits)
    print("\n--- 送給模型的提示詞 ---")
    print(prompt)
    print("\n--- 回答 ---")
    print(generate(prompt))


if __name__ == "__main__":
    ask("沒休完的特休假會怎麼處理？")
    ask("出差住宿費有上限嗎？")
    ask("公司有提供員工旅遊補助嗎？")

執行

uv run --with chromadb --with sentence-transformers rag_demo.py

或先用 pip install chromadb sentence-transformers 裝好套件，再執行 python rag_demo.py。要接真的模型，另外安裝 anthropic 並設定 ANTHROPIC_API_KEY。

執行結果

沒有設定金鑰時，我實際得到的輸出（節錄）：

索引完成：4 塊

==================================================
問題： 沒休完的特休假會怎麼處理？
檢索 [1] 距離 0.422｜員工手冊.md ＞ 員工手冊 > 特別休假
...
--- 回答 ---
（沒有設定 API 金鑰，這裡不呼叫模型，只顯示送給模型的提示詞）
==================================================
問題： 出差住宿費有上限嗎？
檢索 [1] 距離 0.270｜報帳規定.md ＞ 報帳規定 > 差旅費
...
==================================================
問題： 公司有提供員工旅遊補助嗎？
檢索：沒有夠相關的資料 → 直接回答「資料中沒有提到」，不呼叫模型

完整的逐題分析在 5.6。

自己試試看

  1. 把 DOCS 換成你自己的文件，換成真實的問題，看檢索找不找得到。
  2. 把 MAX_DISTANCE 調成 0.3 或 0.8，觀察哪些問題被擋掉、哪些被放行。
  3. 把 TOP_K 調成 1 和 5，看提示詞怎麼變。
  4. 拿掉「標題放進內容再嵌入」那一行，改成只嵌入內文，看距離怎麼變。
  5. 把問題故意問得很口語、很簡短，例如「年假咧？」，看還找不找得到。
  6. 有 API 金鑰的話，接上模型，檢查答案有沒有標出編號，編號對不對。
  7. 把 chromadb.Client() 改成 chromadb.PersistentClient(path="./db")，讓索引保存下來。

這個範例缺什麼

它能跑，但離正式的系統還很遠：

  • 知識庫只有 4 塊，真實的資料量大很多。
  • 門檻值只用 3 個問題湊出來，沒有系統性的驗證。
  • 沒有處理多輪對話、錯誤重試、權限。
  • 完全沒有衡量「答案好不好」。
← 全部文章