向量資料庫與相似度搜尋

前兩堂你準備好了兩樣材料：會把文字變成座標的 embedding 模型，和切好、帶著出處的文件塊。這堂課要把它們組起來，做出一個依照意思搜尋的小搜尋引擎。

我們會分兩步走：先用最樸素的方法親手寫一次，確定你知道裡面在做什麼；再換成專門的工具，讓它更實用。

整體流程：兩個階段

建索引像替圖書館的每本書編目，只要做一次；查詢則是每位讀者來問時做。

• 建索引（indexing）：把所有塊都算好 embedding，存起來。文件有新增或修改時才需要重做。
• 查詢（query）：把使用者的問題也變成 embedding，找出最接近的幾塊。

有一個重點一定要記得：查詢和建索引必須用同一個 embedding 模型。第一堂說過，不同模型是不同的地圖，混著用會完全找不到東西。

第一版：用 numpy 親手寫

建立 search_numpy.py，和上一堂的 chunking.py 放在同一個資料夾：

import numpy as np
from sentence_transformers import SentenceTransformer
from chunking import load_chunks

model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")

# 建索引：所有塊一次算好
chunks = load_chunks("docs")
vectors = model.encode([c["text"] for c in chunks], normalize_embeddings=True)
print(f"已建立索引：{len(chunks)} 塊")


def search(query: str, k: int = 3):
    q = model.encode(query, normalize_embeddings=True)
    scores = vectors @ q                # 問題和每一塊的相似度
    top = np.argsort(-scores)[:k]       # 分數由高到低排序，取前 k 個
    return [(float(scores[i]), chunks[i]) for i in top]


while True:
    query = input("\n問題（直接按 Enter 結束）：").strip()
    if not query:
        break
    for score, chunk in search(query):
        preview = chunk["text"].replace("\n", " ")[:40]
        print(f"{score:.2f}  {chunk['id']}  {preview}…")

執行後試著問「計程車錢可以拿回來嗎？」：

python search_numpy.py

k 是「要拿回幾塊」。前幾名分數高、後面明顯掉下來，是很健康的搜尋結果。

整個搜尋只有三行：把問題變成向量、和所有塊相乘得到分數、排序取前 k 個。這就是相似度搜尋的全部原理。

k 要設多少？

• 太小（例如 1）：答案剛好分散在兩塊時，會漏掉一半。
• 太大（例如 20）：送給模型的內容變多，又慢又貴，不相關的內容也可能干擾回答。

一般從 3 到 5 開始試。第五堂課會教你用測試題找出適合你資料的數字。

為什麼還需要向量資料庫？

numpy 版本很好懂，但有兩個問題：

• 每次執行都要重算所有 embedding。三份手冊沒感覺，三千份文件就要等很久。
• 資料都在記憶體裡，程式一關就不見了，出處等資訊也要自己管理。

向量資料庫（vector database）就是專門解決這些事的工具：把向量、原文、出處一起存在硬碟上，查詢時直接幫你找出最近的幾筆。資料量很大時，它還會用特殊的索引方法加速，不必每次都和每一筆比對。

常見的選擇有 Chroma、Qdrant、pgvector（PostgreSQL 的擴充功能）、Pinecone 等。我們用 Chroma，因為它安裝簡單，資料就存在專案資料夾裡，不用另外架伺服器。

pip install chromadb

第二版：換成 Chroma

建立 store.py。之後的課程都會從這裡匯入 search：

import chromadb
from sentence_transformers import SentenceTransformer
from chunking import load_chunks

EMBED_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"

embedder = SentenceTransformer(EMBED_MODEL)
db = chromadb.PersistentClient(path="chroma_db")  # 資料存在 chroma_db 資料夾
collection = db.get_or_create_collection(name="handbook")


def embed(texts: list[str]) -> list[list[float]]:
    return embedder.encode(texts, normalize_embeddings=True).tolist()


def build_index(folder: str = "docs") -> int:
    chunks = load_chunks(folder)
    collection.upsert(  # upsert：有同樣 id 就更新，沒有就新增
        ids=[c["id"] for c in chunks],
        documents=[c["text"] for c in chunks],
        metadatas=[{"source": c["source"], "title": c["title"]} for c in chunks],
        embeddings=embed([c["text"] for c in chunks]),
    )
    return len(chunks)


def search(query: str, k: int = 3) -> list[dict]:
    result = collection.query(query_embeddings=embed([query]), n_results=k)
    hits = []
    for i in range(len(result["ids"][0])):
        distance = result["distances"][0][i]
        hits.append({
            "id": result["ids"][0][i],
            "text": result["documents"][0][i],
            "source": result["metadatas"][0][i]["source"],
            "score": 1 - distance / 2,  # 把距離換回餘弦相似度，見下方說明
        })
    return hits


if __name__ == "__main__":
    print("已建立索引：", build_index(), "塊")
    while True:
        query = input("\n問題（直接按 Enter 結束）：").strip()
        if not query:
            break
        for hit in search(query):
            preview = hit["text"].replace("\n", " ")[:40]
            print(f"{hit['score']:.2f}  {hit['id']}  {preview}…")

執行：

python store.py

結果應該和 numpy 版本差不多。不同的是，專案裡多了一個 chroma_db 資料夾，索引已經存起來了，之後其他程式匯入 search 時可以直接查詢。

幾個值得注意的地方

• query 回傳的是「清單的清單」：因為 Chroma 允許一次查詢好幾個問題，所以結果要用 [0] 取出第一個問題的答案。
• Chroma 給的是「距離」，不是相似度：距離越小越像。它預設用的是平方歐氏距離，因為我們的向量都標準化過，距離和餘弦相似度剛好可以互換：相似度 ＝ 1 減去「距離除以 2」。換回相似度，數字就和第一堂學的一致，比較好理解。
• metadatas 就是出處：下一堂讓模型附上引用時，就靠這裡的 source。