前言

前幾天分別談了 Embedding 與向量資料庫的選型，今天要把兩者串接起來，實際把 Day 9 產出的 chunk 資料，轉換成向量並寫入資料庫，完成 Indexing 階段的最後一哩路。

完整流程回顧

原始文件 → 前處理 Pipeline（Day 5-9） → chunk 清單
                                            │
                                            ▼
                                  逐筆呼叫 Embedding API
                                            │
                                            ▼
                              寫入向量資料庫（含 content + embedding + metadata）
                                            │
                                            ▼
                                      建立索引，等待查詢

實作：批次寫入向量資料庫

import psycopg2
from psycopg2.extras import execute_values

def insert_chunks_to_db(chunks, conn):
    """
    chunks: [{"content": ..., "embedding": [...], "metadata": {...}}, ...]
    """
    cur = conn.cursor()
    data = [
        (c["content"], c["embedding"], psycopg2.extras.Json(c["metadata"]))
        for c in chunks
    ]
    execute_values(
        cur,
        "INSERT INTO chunks (content, embedding, metadata) VALUES%s",
        data,
        template="(%s,%s,%s)"
    )
    conn.commit()
    cur.close()

批次寫入（execute_values）相較於逐筆 INSERT，可以大幅減少資料庫往返次數，加快寫入速度。

整合前處理 + Embedding + 寫入的完整腳本

def build_index(doc_dir, conn):
    all_chunks = []
    for file_path in doc_dir.glob("**/*"):
        if file_path.suffix in [".pdf", ".html", ".md"]:
            chunks = preprocess_pipeline(file_path)  # Day 9 的 Pipeline
            all_chunks.extend(chunks)

    # 批次做 Embedding（可依 API 限制調整批次大小）
    batch_size = 50
    for i in range(0, len(all_chunks), batch_size):
        batch = all_chunks[i:i+batch_size]
        texts = [c["content"] for c in batch]
        embeddings = get_embeddings_batch(texts)  # 批次呼叫 Embedding API
        for c, emb in zip(batch, embeddings):
            c["embedding"] = emb

    insert_chunks_to_db(all_chunks, conn)
    print(f"共處理{len(all_chunks)} 個 chunk，已寫入資料庫")

處理大量文件時的注意事項

• API Rate Limit：Embedding API 通常有每分鐘請求數限制，大量文件建議加入延遲或使用官方的批次端點
• 記憶體用量：文件量大時，不要一次把所有 chunk 都讀進記憶體，可以分批處理
• 重複資料：文件如果更新過，避免重複寫入相同內容，建議先用 Day 9 提到的增量更新機制判斷是否需要重新處理

驗證索引是否建立成功

-- 檢查資料筆數
SELECT COUNT(*) FROM chunks;

-- 檢查是否有 embedding 為 NULL 的異常資料
SELECT COUNT(*) FROM chunks WHERE embedding IS NULL;

-- 簡單測試相似度搜尋是否能運作
SELECT content FROM chunks
ORDER BY embedding <=> (SELECT embedding FROM chunks LIMIT 1)
LIMIT 5;

小結

到這裡，Indexing 階段的完整流程——文件前處理、Chunking、Embedding、寫入向量資料庫——已經串接完成。這代表我們的知識庫已經「可以被查詢」了。明天開始，我們要進入 Retrieval 階段，先從最基礎的相似度搜尋原理談起。