向量資料庫:選型與實作

安裝依賴套件、啟動 Qdrant

pip install qdrant-client openai python-dotenv

用 Docker 在本機啟動 Qdrant,資料持久化到 ./qdrant_data:

docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -v $(pwd)/qdrant_data:/qdrant/storage \
  qdrant/qdrant

啟動後用瀏覽器開 http://localhost:6333/dashboard 可以看到 Qdrant 的管理介面——Collection 列表、向量數量都在這裡查。

建立 .env:

OPENAI_API_KEY=sk-...

建立 Collection

Collection 相當於關聯式資料庫的 table,建立時必須指定兩個關鍵參數:向量維度和距離度量。

# create_collection.py
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

qdrant = QdrantClient(host="localhost", port=6333)

COLLECTION_NAME = "my_docs"
VECTOR_SIZE = 1536  # text-embedding-3-small 輸出維度

# 如果 collection 已存在就刪掉重建(開發階段方便重跑)
if qdrant.collection_exists(COLLECTION_NAME):
    qdrant.delete_collection(COLLECTION_NAME)

qdrant.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=VectorParams(
        size=VECTOR_SIZE,
        distance=Distance.COSINE,  # embedding 模型用 cosine,dot product 要正規化向量
    ),
)

print(f"Collection '{COLLECTION_NAME}' 建立完成")

距離度量的選法:OpenAI 和 Voyage 的 embedding 輸出向量已正規化,用 COSINE 或 DOT 效果相同——用 COSINE 最保險,不會因為向量大小差異影響相似度計算。

準備 100 份文件並批次 Embed + 寫入

這一步是整個 RAG pipeline 裡最耗時的——Embed 要呼叫 API、費用也在這裡發生。我們用 OpenAI text-embedding-3-small(每百萬 token $0.02,是 2026 年最經濟實用的選擇)。

先準備資料結構(用你自己的文件替換這個範例):

# docs 是你的 100 份文件,每份是一個 dict
# 這裡用假資料示範結構
docs = [
    {
        "id": i,
        "content": f"這是第 {i+1} 份文件的正文內容,談到某個主題的具體說明...",
        "source": f"doc_{i+1:03d}.txt",
        "category": "技術文件" if i % 2 == 0 else "使用說明",
    }
    for i in range(100)
]

然後批次取得 embedding 並寫入 Qdrant:

# ingest.py
import os
from dotenv import load_dotenv
import openai
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct

load_dotenv()

openai_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
qdrant = QdrantClient(host="localhost", port=6333)

COLLECTION_NAME = "my_docs"
EMBED_MODEL = "text-embedding-3-small"
BATCH_SIZE = 50  # OpenAI 單次最多 2048 筆,50 一批安全又夠快

def embed_batch(texts: list[str]) -> list[list[float]]:
    """一次送多段文字,回傳對應的向量列表"""
    response = openai_client.embeddings.create(
        model=EMBED_MODEL,
        input=texts,
    )
    # response.data 是有序列表,index 對應輸入 index
    return [item.embedding for item in response.data]

# 把 docs 切成每批 BATCH_SIZE 筆
all_points = []
for batch_start in range(0, len(docs), BATCH_SIZE):
    batch = docs[batch_start : batch_start + BATCH_SIZE]
    texts = [doc["content"] for doc in batch]

    print(f"Embedding {batch_start + 1}–{batch_start + len(batch)} / {len(docs)} ...")
    vectors = embed_batch(texts)

    for doc, vector in zip(batch, vectors):
        all_points.append(
            PointStruct(
                id=doc["id"],        # 整數 ID,Qdrant 也接受 UUID 字串
                vector=vector,
                payload={            # payload 存任意 metadata,之後用來過濾
                    "content": doc["content"],
                    "source": doc["source"],
                    "category": doc["category"],
                },
            )
        )

# 批次寫入 Qdrant
for batch_start in range(0, len(all_points), BATCH_SIZE):
    batch = all_points[batch_start : batch_start + BATCH_SIZE]
    qdrant.upsert(collection_name=COLLECTION_NAME, points=batch)
    print(f"已寫入 {batch_start + len(batch)}/{len(all_points)} 筆")

print("入庫完成!")
info = qdrant.get_collection(COLLECTION_NAME)
print(f"Collection 總向量數: {info.vectors_count}")

成功執行後你會看到:

Embedding 1–50 / 100 ...
Embedding 51–100 / 100 ...
已寫入 50/100 筆
已寫入 100/100 筆
入庫完成!
Collection 總向量數: 100

費用估算:100 份文件、每份平均 500 中文字 ≈ 800 token,總計約 8 萬 token。用 text-embedding-3-small 費用是 $0.02/MTok × 0.08 MTok = 不到 $0.002 美元。可以放心實驗。

語義查詢

查詢的邏輯:把使用者的問題 embed 成向量,然後在 Collection 裡找最相近的前 N 筆。

# query.py
import os
from dotenv import load_dotenv
import openai
from qdrant_client import QdrantClient

load_dotenv()
openai_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
qdrant = QdrantClient(host="localhost", port=6333)

COLLECTION_NAME = "my_docs"
EMBED_MODEL = "text-embedding-3-small"

def embed_query(text: str) -> list[float]:
    response = openai_client.embeddings.create(model=EMBED_MODEL, input=text)
    return response.data[0].embedding

query_text = "如何設定系統環境變數?"
query_vector = embed_query(query_text)

results = qdrant.query_points(
    collection_name=COLLECTION_NAME,
    query=query_vector,
    limit=5,              # 取前 5 筆
    with_payload=True,    # 把 metadata 也帶回來
).points

for i, r in enumerate(results, 1):
    print(f"#{i} [相似度={r.score:.4f}] 來源:{r.payload['source']}")
    print(f"    {r.payload['content'][:120]}...")
    print()

r.score 是 cosine 相似度,1.0 = 完全相同,實際上 0.75 以上通常已是高度相關。

加上 metadata 過濾

純語義搜尋有時太廣:你希望只在「技術文件」這個分類裡搜,或只搜某個使用者上傳的文件。這時要用 metadata 過濾。

重要前置動作:要讓過濾走索引而不是掃描,先幫要過濾的欄位建 payload index:

# 只需建一次,建完 Qdrant 會自動維護
from qdrant_client.models import PayloadSchemaType

qdrant.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="category",
    field_schema=PayloadSchemaType.KEYWORD,  # 字串完全匹配用 KEYWORD
)

建好後就可以在查詢時帶過濾條件:

from qdrant_client.models import Filter, FieldCondition, MatchValue

results_filtered = qdrant.query_points(
    collection_name=COLLECTION_NAME,
    query=query_vector,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="category",
                match=MatchValue(value="技術文件"),
            )
        ]
    ),
    limit=5,
    with_payload=True,
).points

print(f"過濾後找到 {len(results_filtered)} 筆(僅限「技術文件」分類)")

Filter 支援 must(AND)、should(OR)、must_not(NOT)的組合,也支援數值範圍過濾(例如 created_at >= 某個時間戳記)——之後第 5 課 hybrid search 會用到。