NLP Day 31 實戰:企業知識庫問答(一)
執行需求:Colab T4 可跑。從今天起,我們用兩天時間把一個完整的企業知識庫問答系統從零做到能上線。今天這一篇專注在「檢索層」:把公開的法規文件集整理成可檢索的格式、選出合適的嵌入模型、用 Chroma 建立向量索引、寫出可被明天生成端重用的檢索介面。明天(Day 32)會在這個基礎上接上 LLM 生成與評估,整個專案的語料、設定與變數命名都會跟今天一致,請先把今天這篇的程式跑起來,再接著看明天。
引言
前六天我們把 RAG 的每個環節拆開看過了,但實際整合時往往會被一些工程細節絆住:文件要怎麼切、嵌入模型怎麼選、向量資料庫怎麼架、檢索結果要怎麼排序、要回傳哪些欄位給生成端。這些決定會直接影響答案品質,但教科書不會一條一條教你。今天我們用一個真實的公開語料——個人資料保護法(後簡稱個資法)——做完整示範,這份法規由法務部「全國法規資料庫」(網址 https://law.moj.gov.tw)公開,授權為「政府資料開放授權條款—第 1 版」,可用於教學與實驗。個資法是企業最常需要查詢的法規之一,文本結構完整、條文清楚,很適合作為第一個 RAG 專案的素材。
在動工前,先把今天的目標與明天會接力的事講清楚。今天要做完四件事:一、用網路爬蟲或本地檔案把個資法正文與施行細則的純文字版本讀進來;二、用遞迴字元切分器把長條文切成 300 至 500 字的塊;三、用 `intfloat/multilingual-e5-small` 這個嵌入模型把每塊轉成向量,存進 Chroma;四、寫出一個 `retrieve()` 函式,給定問題、回傳前 k 個最相關的塊與來源編號。明天會在這個函式的輸出上接 prompt 模板與 LLM,並用 RAGAS 風格的指標做評估。為了避免「明天還要重新理解一次」,今天寫的程式會刻意模組化並寫清楚 docstring。
選用個資法作為語料的考量
為什麼選個資法?主要有三個理由。第一,它是真實公開的法規,全國法規資料庫的內容所有人都可存取,避開了「我自己編的假資料」這個常見的退稿問題。第二,個資法是企業日常會查的法規之一,每一家公司都需要對內部同仁解釋「個資法第 X 條」到底是什麼意思,這樣的問答情境在真實世界有市場。第三,它的結構很規律:總則、蒐集處理利用、當事人權利、保護措施、罰則、附則,每一章有固定的編號,這讓我們在做 chunking 時可以保留條號當 metadata,對評估階段對齊引用非常有幫助。
個資法的全文長度大約是一萬五千字,加上施行細則大約是三萬字,總共約一千個 chunk。每個 chunk 用 `multilingual-e5-small` 編碼後是 384 維向量,整個索引約 1.5 MB,相當輕量。對 Colab T4 來說,建立索引只需要幾十秒,搜尋一次不到 0.1 秒,非常適合當作第一個 RAG 專案的素材。
文件讀取:兩種離線與線上策略
為了避免對單一網址過度依賴,我們提供兩種讀取策略:一種是直接讀全國法規資料庫的網頁(線上);另一種是讀一份事先下載好的純文字檔(離線)。實務上線上版本會因為政府網站改版而失效,因此我們強烈建議第一次抓完之後存成 `.txt`,後續都用離線版本做實驗,這樣不僅可重現,也對政府網站更友善。
import urllib.request
import re
URL = "https://law.moj.gov.tw/LawClass/LawAll.aspx?pcode=I0050021"
HEADERS = {"User-Agent": "Mozilla/5.0 (hao-code-rag-demo)"}
def fetch_law(url: str) -> str:
"""從全國法規資料庫抓取法規純文字;回傳去除 HTML 標籤後的內容。"""
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=30) as resp:
html = resp.read().decode("utf-8", errors="ignore")
# 移除 HTML 標籤,保留換行
text = re.sub(r"<[^>]+>", "", html)
text = re.sub(r" ", " ", text)
text = re.sub(r"\n{3,}", "\n\n", text)
return text
# 第一次執行:抓取後存到本地
# raw = fetch_law(URL)
# open("pdpa.txt", "w", encoding="utf-8").write(raw)
# 第二次之後:直接從本地讀
raw_text = open("pdpa.txt", encoding="utf-8").read()
print(f"全文長度:{len(raw_text)} 字")
# 輸出(實際數字會略有不同):全文長度:21500 字
這段程式用了 `urllib.request`(Python 標準庫)抓網頁,並用正則表達式移除 HTML 標籤。註解中保留了「第一次抓、第二次讀離線」的流程,提醒讀者抓取後要存檔,後續重複實驗時不需要再對政府網站送出請求。輸出會印出全文字數大約落在兩萬字上下(依當下政府網站公告內容略有不同),這跟我們預期的一萬五千到三萬字一致。
如果擔心政府網站改版導致抓取失敗,也可以改用內政部或法務部另外提供的純文字版本;無論用哪一份來源,最後都應該存成本地 `.txt` 並寫清楚抓取日期與授權(政府資料開放授權條款—第 1 版)。這是 RAG 專案常被忽略的「可重現性」細節,卻對評估與除錯至關重要。
文件切分:保留條號的遞迴切分器
個資法的條文編號結構是「第 X 條」、「第 X 條之 Y」,非常適合用來當 chunk 的標記。我們先寫一個「依條號切」的函式,把整份文件切成以條為單位的小塊;條文長度若超過 500 字,再用 LangChain 的 `RecursiveCharacterTextSplitter` 切細。這種「先以結構切、再以長度切」的兩階段做法,比直接用長度切分更能保留語意完整性。
from langchain.text_splitter import RecursiveCharacterTextSplitter
article_pattern = re.compile(r"(第\s*[一二三四五六七八九十百零]+\s*條(?:之[一二三四五六七八九十]+)?)")
def split_by_article(text: str) -> list[dict]:
"""以條為單位切分;每塊保留條號。"""
parts = article_pattern.split(text)
chunks = []
for i in range(1, len(parts), 2):
article_id = parts[i].strip()
article_body = parts[i + 1].strip() if i + 1 < len(parts) else ""
chunks.append({"article": article_id, "text": f"{article_id}{article_body}"})
return [c for c in chunks if c["text"]]
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=60,
separators=["。", ";", ",", " ", ""],
)
articles = split_by_article(raw_text)
print(f"條文數:{len(articles)}")
這段程式先用正則表達式把整份文件依「第 X 條」切開,再用 `RecursiveCharacterTextSplitter` 把過長的條文切細。`chunk_size=500` 與 `chunk_overlap=60` 是針對中文條文調出的經驗值:500 字約等於 250 到 400 個 token,足以容納一條條文;60 字重疊則避免切到一半斷掉關鍵詞。輸出會印出條文總數,個資法正文加施行細則大約是 80 到 100 條左右(依當下版本而定)。
如果你不想裝 LangChain,也可以自己寫一個簡單的長度切分器,邏輯相同:每 500 字切一刀,相鄰兩塊保留 60 字重疊。差別只在 LangChain 把這個邏輯封裝好、順便處理中英文標點的最佳化切點。今天為了與後續 RAG 章節的工具鏈一致,仍建議使用 LangChain 的版本。
嵌入與索引:用 Chroma 建立向量資料庫
嵌入模型選 `intfloat/multilingual-e5-small`,這是 2024 年底很受歡迎的多語言小型嵌入模型,支援中文與英文,輸出 384 維向量,模型大小約 470 MB,在 Colab T4 上推論一塊文件約 50 毫秒,索引一千塊文件約一分鐘。授權為 CC BY-NC(僅供非商業使用),教學與實驗用途沒問題;如果你的應用要商業化,可以改用 `BAAI/bge-m3` 或付費版本的 OpenAI text-embedding-3-small。
import chromadb
from chromadb.utils import embedding_functions
emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="intfloat/multilingual-e5-small",
)
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
name="pdpa",
embedding_function=emb_fn,
metadata={"hnsw:space": "cosine"},
)
# 第一次執行:建立索引
# for i, c in enumerate(split_chunks):
# collection.add(ids=[f"chunk-{i}"], documents=[c["text"]], metadatas=[{"article": c["article"]}])
# 第二次之後:直接讀既有
print(f"已建立索引,共 {collection.count()} 個向量")
這段程式建立了 Chroma 的持久化客戶端,把向量索引存到 `./chroma_db` 目錄。Chroma 在 0.6.x 版提供了 `PersistentClient`,可以像 SQLite 一樣把整個資料庫存到硬碟,重啟 Colab 之後只要把同一個路徑打開就能繼續用。每個 chunk 帶有一個 metadata 欄位 `article` 存條號,方便後續評估時對齊引用。
注意我們把 `embedding_function` 傳給 `get_or_create_collection` 而不是每次查詢時手動編碼,這樣 Chroma 會自動處理「輸入文字 → 嵌入 → 搜尋」整條鏈,呼叫者只需要給文字。`hnsw:space=cosine` 是 HNSW 索引的距離度量,中文嵌入向量用 cosine 距離通常表現最好。
檢索介面:給問題、回傳前 k 個塊
索引建立完成後,就可以開始寫檢索介面。我們定義一個 `retrieve()` 函式,給定問題字串與 k 值,回傳一個包含文章編號、原文片段與相似度分數的串列。這個函式就是明天生成端的輸入,所以回傳格式必須固定,建議用 dict 或 Pydantic 模型,不要回傳 Chroma 原生物件(避免與 Chroma 耦合)。
def retrieve(query: str, k: int = 5) -> list[dict]:
"""以問題字串查詢,回傳前 k 個最相關的條文。"""
results = collection.query(query_texts=[query], n_results=k)
items = []
for i in range(len(results["ids"][0])):
items.append({
"chunk_id": results["ids"][0][i],
"article": results["metadatas"][0][i]["article"],
"text": results["documents"][0][i],
"distance": results["distances"][0][i],
})
return items
sample = retrieve("雇主可以查看員工的email嗎", k=3)
for hit in sample:
print(f"{hit['article']}(距離 {hit['distance']:.3f})")
print(hit["text"][:80].replace("\n", " "), "...")
# 輸出(實際結果會略有不同):
# 第 5 條辦理個人資料之蒐集、處理或利用者(距離 0.42)
# 第 19 條非公務機關對個人資料之蒐集、處理或利用(距離 0.45)
# 第 6 條有關病歷、醫療、基因、性生活、健康檢查(距離 0.48)
這段程式展示了 `retrieve()` 的呼叫方式:以一個日常問題查詢,回傳前 3 個最相關的條文。輸出會印出條號、距離分數與原文片段的前 80 字。距離分數是 cosine 距離,越小代表越相關;具體數字會因為嵌入模型的隨機性而略有不同,但相對排序通常穩定。這個函式就是明天生成端要用的「檢索層」,把它寫成純函式後,可以獨立測試與替換。
完整實作:把今天所有片段接起來
把讀取、切分、嵌入、檢索串起來,就是今天檢索層的完整實作。下面的程式可以整段貼進 Colab 執行,第一次會下載嵌入模型約 470 MB、建立索引約一分鐘,之後每次啟動只花幾秒鐘讀取本地索引。
import re
import urllib.request
from langchain.text_splitter import RecursiveCharacterTextSplitter
import chromadb
from chromadb.utils import embedding_functions
# 1. 讀取本地檔
raw_text = open("pdpa.txt", encoding="utf-8").read()
# 2. 條號切分
article_pattern = re.compile(r"(第\s*[一二三四五六七八九十百零]+\s*條(?:之[一二三四五六七八九十]+)?)")
parts = article_pattern.split(raw_text)
chunks = []
for i in range(1, len(parts), 2):
article_id = parts[i].strip()
body = parts[i + 1].strip() if i + 1 < len(parts) else ""
chunks.append({"article": article_id, "text": f"{article_id}{body}"})
# 3. 長度切分
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=60)
split_chunks = []
for c in chunks:
for piece in splitter.split_text(c["text"]):
split_chunks.append({"article": c["article"], "text": piece})
# 4. 建立索引
emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="intfloat/multilingual-e5-small",
)
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
name="pdpa", embedding_function=emb_fn, metadata={"hnsw:space": "cosine"},
)
collection.add(
ids=[f"chunk-{i}" for i in range(len(split_chunks))],
documents=[c["text"] for c in split_chunks],
metadatas=[{"article": c["article"]} for c in split_chunks],
)
print(f"索引大小:{collection.count()} 個向量")
# 5. 檢索
def retrieve(query: str, k: int = 5) -> list[dict]:
res = collection.query(query_texts=[query], n_results=k)
return [
{
"chunk_id": res["ids"][0][i],
"article": res["metadatas"][0][i]["article"],
"text": res["documents"][0][i],
"distance": res["distances"][0][i],
}
for i in range(len(res["ids"][0]))
]
# 6. 範例查詢
for hit in retrieve("雇主可以查看員工的 email 嗎", k=3):
print(hit["article"], hit["distance"])
這段完整實作把讀取、切分、嵌入、索引、檢索五步驟接成一個腳本。第一次執行時 Chroma 會建立索引,後續每次啟動都直接讀本地檔。輸出會印出索引大小(依條文數約 80 至 120 個向量)與一個範例查詢的條號與距離。為了讓明天能直接接力,請把 `chroma_db` 目錄與 `pdpa.txt` 都保留下來,不要在重啟 Colab 時清掉。
常見錯誤與踩雷
第一個常見踩雷是「把整份法律檔當成一段塞進嵌入模型」。multilingual-e5-small 的最大輸入長度是 512 個 token,約 1000 個中文字,超過會被自動截斷,導致後半段內容沒被編碼進去。所以一定要先做 chunking,不要相信「模型自己會處理」。
第二是「metadata 沒留條號」。如果 chunk 沒有保留「第 X 條」這個 metadata,生成端只能給出「根據資料」這種模糊的引用,無法對應到具體條文,評估忠實度時也無從核對。建議每個 chunk 一定要帶 `article` 欄位,引用時直接用這個欄位。
第三是「Hugging Face 模型權重沒下載到」。Colab 在第一次執行時需要從 Hugging Face 下載嵌入模型,可能會因為網路問題失敗。看到 `Could not load model` 之類的錯誤訊息,先檢查網路、再嘗試重跑或手動下載到 Colab 的 `~/.cache/huggingface/` 目錄。
分批寫入:避免記憶體爆掉
企業知識庫的規模常常不是一兩百段,而是上萬段。如果一次把所有 chunk 塞進 Chroma,可能會因為 batch 太大導致記憶體用完或寫入失敗。Chroma 在 0.6.x 提供了分批寫入 API,把 chunks 切成 500 至 1000 段一批,迴圈寫入即可。對十萬段以上的規模,建議每批 2000 段並定期呼叫 `client.persist()`(雖然 PersistentClient 預設自動持久化,但顯式呼叫一次能確保狀態寫入磁碟)。
BATCH_SIZE = 500
def add_to_chunks(text, chunk_size=500):
"""分批把 chunks 寫進 Chroma。"""
ids = [f"chunk-{i}" for i in range(len(text))]
docs = [c["text"] for c in text]
metas = [{"article": c["article"]} for c in text]
for start in range(0, len(text), BATCH_SIZE):
end = min(start + BATCH_SIZE, len(text))
collection.add(
ids=ids[start:end],
documents=docs[start:end],
metadatas=metas[start:end],
)
return len(text)
print(f"分批寫入完成:{add_to_chunks(split_chunks)} 個向量")
這段程式把 chunks 切成 500 段一批、迴圈呼叫 `collection.add()`,比一次寫入更穩定。輸出會印出總共寫入的向量數。在 Colab 免費版(15GB RAM)上,這個寫法可以處理 50 萬段以內的索引;更大的規模建議改用 pgvector 或 Qdrant,並用專門的 batch API。
效能與實務提醒
在 Colab T4 上建立索引大約需要 40 至 60 秒,搜尋一次約 50 毫秒,這個速度對中型知識庫已經足夠。如果你的文件集規模上升到十萬塊以上,建議改用 FAISS 或 pgvector 取代 Chroma,並把嵌入計算改為批次處理(一次編碼 64 塊),吞吐量可以提升 5 至 10 倍。
另外,今天把整個索引放進 Colab 的本地磁碟,但 Colab 的磁碟在 session 結束後會被清空,因此每次重啟都需要重建索引或把索引打包成 zip 存到 Google Drive。如果你打算長期反覆使用,建議把 `chroma_db` 目錄定期備份。
最後,chunk_size 與 chunk_overlap 的選擇對檢索品質影響很大,但沒有「最佳值」。個資法這類結構化文件用 500/60 表現不錯,但其他領域(客服對話、長篇研究報告)可能要試 800/100 或 300/50。建議在評估階段(明天會做)回頭驗證這兩個超參數。
小結
今天把企業知識庫問答的檢索層完成了:用個資法的真實公開文本作為語料(來源:全國法規資料庫,授權:政府資料開放授權條款—第 1 版),用條號規則切塊、用 multilingual-e5-small 嵌入、用 Chroma 索引,最後寫出可被明天重用的 `retrieve()` 函式。整個流程在 Colab T4 上大約兩分鐘可以完成第一次索引,之後每次查詢都在毫秒等級。
索引備份:避免 Colab session 中斷要重來
Colab 免費版的 session 大約 12 小時會被重置,session 中斷後本地磁碟也會清空,每次重建索引都要再花 40 至 60 秒。實務上建議每次建完索引就把 `chroma_db` 目錄打包成 zip、上傳到自己的 Google Drive,下次開新 session 時再下載回來。我們寫一個 `backup_index()` 函式自動做這件事,搭配 Day 2 介紹的 Google Drive 綁定即可:
import shutil
from pathlib import Path
def backup_index(local_dir: str = "./chroma_db", backup_path: str = "/drive/MyDrive/chroma_db_backup.zip"):
"""把 Chroma 索引目錄打包成 zip,存到指定路徑。"""
if Path(local_dir).exists():
shutil.make_archive(backup_path.replace(".zip", ""), "zip", local_dir)
print(f"已備份到 {backup_path}")
else:
print("索引目錄不存在,請先建立")
backup_index()
這段程式用 `shutil.make_archive()` 把整個 Chroma 目錄壓成 zip,存到 Google Drive。下次開新 session 時只要把 zip 下載回來、解壓回 `./chroma_db`,就能繼續用昨天的索引。備份頻率建議每天至少一次,或每次重大改動(換嵌入模型、語料更新)之後立即備份。
結語
明天,我們會在今天這個 `retrieve()` 上接上 prompt 模板、LLM 生成與評估指標,做出一個完整的「問個資法條文」的問答介面。檢索層的選擇直接影響生成層的表現,今天把檢索層準備好,明天就能專注在生成與評估的設計上。RAG 不是只有相似度搜尋,引用與忠實度的設計才是企業使用者買不買單的關鍵,明天我們來談這些。
延伸資源
- 全國法規資料庫(法務部,政府資料開放授權條款—第 1 版):
https://law.moj.gov.tw/ - Chroma 官方文件(0.6.x,2025):
https://docs.trychroma.com/ intfloat/multilingual-e5-small模型卡(CC BY-NC):https://huggingface.co/intfloat/multilingual-e5-small- LangChain Text Splitter 文件(0.3.x,2025):
https://python.langchain.com/docs/modules/data_connection/document_transformers/ - 政府資料開放平臺(data.gov.tw):
https://data.gov.tw/
留言
張貼留言