前言
前幾天分別談了 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 階段，先從最基礎的相似度搜尋原理談起。