iT邦幫忙

2025 iThome 鐵人賽

DAY 9
0
DevOps

30 天帶你實戰 LLMOps:從 RAG 到觀測與部署系列 第 9

Day09 - 向量化 (Vectorize)與索引(Index)建立

  • 分享至 

  • xImage
  •  

🔹 前言

昨天(Day 8)我們完成了兩件重要的事:

  1. 文件清洗 (Cleaning) → 把雜訊、廣告、過長段落處理乾淨,確保知識來源乾淨。
  2. 文件切片 (Chunking) → 把超長的文本切成合適大小的片段,並處理 overlap,避免語意斷裂。

到這裡,我們已經把「原始文件」轉換成一堆乾淨、可用的小片段。
但是這些片段依然只是「文字」,電腦並不理解它們的語意。

今天(Day 9)我們要邁向下一步:

把文字轉換成向量(Vectorize),並建立索引(Index),讓檢索能夠快速、語意準確地進行。

https://ithelp.ithome.com.tw/upload/images/20250923/20120069QopAM67NRF.png


🔹 為什麼需要向量化?

在 RAG 系統中,檢索的關鍵在於「查詢」和「知識」之間的語意對齊。

  • 傳統的關鍵字檢索:只能比對字面文字,無法理解語意。
  • 向量化 (Embedding):把文字轉換成高維度的數字向量,讓電腦能透過「距離」來判斷語意的接近程度。

舉個例子:

  • 「總公司在哪裡?」 vs. 「你們的 Headquarter 地址?」

關鍵字幾乎不同,但在向量空間裡會非常接近,代表能檢索到同一段知識。


🔹 向量化流程

https://ithelp.ithome.com.tw/upload/images/20250923/20120069uvfxCHjeAF.png

  1. 選擇 Embedding 模型

    • OpenAI:text-embedding-3-small / large (簡單好用,支援多國語言)
    • HuggingFace:sentence-transformers/all-MiniLM-L6-v2 (開源,可離線部署)
    • 本地模型:成本高,但可針對特定領域客製(需 GPU 支援,常用於內網環境)
  2. 用 OpenAI API 把文字向量化

詳見 GitHub Repo - 01_embed_quickcheck.py

def main():
    load_dotenv()
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("沒有找到 OPENAI_API_KEY,請在 .env 設定!")

    # 使用環境變數建立 Client(建議統一風格:隱式讀取即可)
    client = OpenAI()

    text = "RAG pipeline 需要清洗與切片"
    resp = client.embeddings.create(model="text-embedding-3-small", input=text)
    vec = resp.data[0].embedding
    print(f"✅ 取得 embedding 成功,維度長度:{len(vec)}")

執行結果:

❯ python 01_embed_quickcheck.py
✅ 取得 embedding 成功,維度長度:1536
  1. 儲存到向量資料庫 (Vector DB)

Day04 - 向量資料庫選型 (Weaviate, Pinecone, FAISS, Milvus) - 以維運成本做決策 我們比較了 Weaviate, Pinecone, FAISS, Milvus 等向量資料庫,今天就直接把向量化後的文字存進去。

  • 小規模 Demo → FAISS(純 Python、可本地跑)
  • 需要雲端服務 → Pinecone(托管型,維護成本低)
  • 想要自架 → Milvus / Weaviate(功能完整,支援擴展)

💡 如果你只是要跑小型測試,用 FAISS 就夠;但在生產環境,通常會選擇雲端向量 DB 來處理擴展性、備份與監控問題。


🔹 向量索引 (Indexing)

當資料量越來越大,檢索速度就會成為瓶頸。

向量索引(Indexing) 能在高維空間建立「捷徑」,讓查詢能在毫秒級完成。在 Day04 - 向量資料庫選型 (Weaviate, Pinecone, FAISS, Milvus) - 以維運成本做決策 我們比較過「資料庫選型」;今天則進一步深入「索引演算法」。兩者加在一起,才能構成完整的檢索方案。

常見技術

方法 原理 適合場景 優點 缺點 常見搭配 實務案例
Flat 暴力搜尋,精準但慢 小型 demo / baseline 準確 資料量大時慢 - FAISS IndexFlatL2- 真實標準 / 標準答案的基準線測試 高精度需求:醫療研究文件、法律文件比對
HNSW (Hierarchical Navigable Small World Graph) 基於圖的高效率近似最近鄰檢索 百萬級、低延遲需求 查詢快速 記憶體佔用高,建立時間較長 - Weaviate、Milvus 預設索引 - Qdrant 預設演算法 聊天機器人、即時問答
IVF (Inverted File Index) 分桶 + 掃桶加速搜尋 上億級,折衷速度/精度 速度快 需訓練、調整參數(精準度取決於分桶策略) - FAISS IndexIVF - Milvus 支援 IVF+PQ 大規模文件檢索、雲端服務
PQ (Product Quantization) 壓縮向量以降低記憶體需求 超大資料,節省成本 節省記憶體 精準度下降 - 雲端商用方案(如 Pinecone、Qdrant)底層演算法多半用 HNSW+ IVF/PQ 混合- FAISS IndexIVFPQ 搜尋引擎、電商推薦系統

🔹 範例:用 FAISS 建立索引

這邊的程式碼有先簡化過,我特別把「我要報銷」拿來跟每個文件片段逐一比較,這樣更能直觀理解所謂「語意相似的距離」是什麼。

然而實務上在檢索的時候並不會這麼做,而是會自動比較 Query 向量所有文件向量 計算距離以及相似度,再依相似度排序(Top-K)。我把貼近實務的程式碼貼在 GitHub Repo 裡的 03_search_compare_l2_cosine.py ,有興趣深入了解的話可以再去研究看看。

# 02_faiss_minimal_flat.py

TOP_K = 5
EMBED_MODEL = "text-embedding-3-small"

DOCS = [
    "加班申請需事先提出,加班工時可折換補休",
    "出差申請需填寫出差單,並附上行程與預算",
    "報銷規則需要提供發票,金額超過 1000 需經理簽核",
    "員工請假需提前一天提出,病假需附上診斷證明",
    "差旅住宿費上限為每晚 3000 元",
    "年度績效考核結果將影響年終獎金比例",
    "離職需提前一個月提出申請",
    "會議室使用需事先預約,不可長期佔用",
    "公司總部位於台北市信義區",
    "飲料與零食可由團隊經費報支",
    "伺服器維護時間為每週日凌晨 2 點到 4 點",
    "資料庫備份需至少保存 90 天",
]

def pretty_print_results(query: str, docs: list[str], indices: np.ndarray, dists: np.ndarray, k: int = TOP_K) -> None:
    print("\n" + "═" * 60)
    print(f"【查詢】{query}")
    print(f"【Top {k} 結果|距離越小越相關】")
    print("─" * 60)
    for rank, (i, d) in enumerate(zip(indices[:k], dists[:k]), start=1):
        print(f"{rank:>2}. 距離={d:.4f} | #{i:02d} | {docs[i]}")
    print("═" * 60)

def main() -> None:
    # 讓 FAISS 在 macOS/Arm 上更穩定(非必要,但建議保留)
    faiss.omp_set_num_threads(1)

    # 讀取 API Key
    load_dotenv()
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("沒有找到 OPENAI_API_KEY,請在 .env 設定!")

    client = OpenAI(api_key=api_key)

    # 顯示文件清單
    print("【文件清單】")
    for i, d in enumerate(DOCS):
        print(f"{i:02d}: {d}")

    # 向量化(示範簡單逐筆;若大量文件,建議改成 batch 版本)
    print("\n正在產生文件向量(embedding)...")
    embs = []
    for d in DOCS:
        emb = client.embeddings.create(model=EMBED_MODEL, input=d).data[0].embedding
        embs.append(emb)
    embs = np.array(embs, dtype="float32")

    # 顯示向量維度
    dim = embs.shape[1]
    print(f"Embedding 維度:{dim}(模型:{EMBED_MODEL})")
    print("第一個文件向量前 8 維:", np.round(embs[0][:8], 4))

    # 建立 FAISS Flat (L2) 索引
    index = faiss.IndexFlatL2(dim)
    index.add(embs)

    # 讀取查詢字串(可由命令列帶入)
    query = sys.argv[1] if len(sys.argv) > 1 else "我要報銷 2000 元"

    # 產生查詢向量並檢索
    q = client.embeddings.create(model=EMBED_MODEL, input=query).data[0].embedding
    q = np.array([q], dtype="float32")

    D, I = index.search(q, k=min(TOP_K, len(DOCS)))
    dists, idxs = D[0], I[0]

    # 顯示結果
    pretty_print_results(query, DOCS, idxs, dists, k=min(TOP_K, len(DOCS)))

    # 參考:與「Naive 全量比較」對照(驗證正確性)
    # (Flat 索引與全量 L2 計算理論上結果應一致)
    naive = np.array([norm(q[0] - v) for v in embs], dtype="float32")
    naive_order = naive.argsort()[: min(TOP_K, len(DOCS))]
    if not np.array_equal(naive_order, idxs[: len(naive_order)]):
        print("\n⚠️ 提示:FAISS 與 Naive 排序不同,請檢查向量或索引設定。")
    else:
        print("\n✅ 驗證:FAISS 與 Naive(L2) 排序一致。")

執行結果:

❯ python 02_faiss_minimal_flat.py
【文件清單】
00: 加班申請需事先提出,加班工時可折換補休
01: 出差申請需填寫出差單,並附上行程與預算
02: 報銷規則需要提供發票,金額超過 1000 需經理簽核
03: 員工請假需提前一天提出,病假需附上診斷證明
04: 差旅住宿費上限為每晚 3000 元
05: 年度績效考核結果將影響年終獎金比例
06: 離職需提前一個月提出申請
07: 會議室使用需事先預約,不可長期佔用
08: 公司總部位於台北市信義區
09: 飲料與零食可由團隊經費報支
10: 伺服器維護時間為每週日凌晨 2 點到 4 點
11: 資料庫備份需至少保存 90 天

正在產生文件向量(embedding)...
Embedding 維度:1536(模型:text-embedding-3-small)
第一個文件向量前 8 維: [-0.0004  0.0519  0.0299  0.0002 -0.029  -0.001   0.004   0.0532]

════════════════════════════════════════════════════════════
【查詢】我要報銷 2000 元
【Top 5 結果|距離越小越相關】
────────────────────────────────────────────────────────────
 1. 距離=0.9170 | #02 | 報銷規則需要提供發票,金額超過 1000 需經理簽核
 2. 距離=1.2500 | #09 | 飲料與零食可由團隊經費報支
 3. 距離=1.2730 | #06 | 離職需提前一個月提出申請
 4. 距離=1.3131 | #04 | 差旅住宿費上限為每晚 3000 元
 5. 距離=1.3954 | #11 | 資料庫備份需至少保存 90 天
════════════════════════════════════════════════════════════

✅ 驗證:FAISS 與 Naive(L2) 排序一致。

實務上的文件量通常動輒上千上萬條,所以並不會一筆一筆慢慢處理,而是會批次處理(Batch Processing)進行以節省 API 呼叫次數、時間以及 Token 用量:

def batch_embedding(client: OpenAI, texts: list[str], batch_size: int = BATCH_SIZE) -> np.ndarray:
    """批次 embedding,避免 API 頻繁呼叫"""
    all_embeddings: list[np.ndarray] = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        resp = client.embeddings.create(model=EMBED_MODEL, input=batch)
        embs = [d.embedding for d in resp.data]
        all_embeddings.extend(embs)
    return np.array(all_embeddings, dtype="float32")

範例:02b_faiss_flat_batch.py

📝 生活化比喻:圖書館的目錄卡片

可以用「圖書館」來比喻:

  • 原始文件:就像一堆沒有整理的書
  • 清洗 & Chunking:把書拆成段落,寫成小卡片
  • 向量化:替每張卡片加上「語意條碼」
  • 索引建立:把卡片放進目錄櫃,依照條碼分類。
  • 檢索:讀者提問時,能馬上翻到最相關的書頁。

https://ithelp.ithome.com.tw/upload/images/20250923/20120069dcrq3dYdz6.png


🔹 生產環境要注意什麼?

除了基本的「向量化 + 索引」之外,在真實系統中還會遇到:

  • 模型選擇:精準度(與維度相關)、成本(越高維度越貴)、語言支援(small vs large vs multilingual)
  • 批次處理:避免 API 頻繁呼叫導致費用遽增,生產環境會需要做 batch embedding(一次處理 50~100 筆)
  • 正規化:很多人忘記做 L2 normalization,導致檢索分數不穩定。
  • 成本與效能:需要監控 embedding 成本、紀錄延遲(p95)的指標。
  • 快取與冷啟動:常見查詢和向量可以快取,新系統要設計 fallback 機制(例如混合關鍵字)
  • 版本與一致性:文件更新後,向量與索引需要同步更新,否則回答會失真。

📌 這些議題我會在 Day19(可觀測性)、Day21(快取機制)、Day22(版本治理)、Day27(FAQ Bot 驗收) 再深入討論。


🔹 小結

在 Day 4 我們比較過不同的 向量資料庫(FAISS、Pinecone、Milvus、Weaviate)。今天則是實際動手,從「乾淨的文件片段」更進一步,完成了 向量化索引建立

  • 向量化 (Embedding):把文字轉換成數字向量,讓電腦能用「距離」來判斷語意相似度。
  • 索引 (Indexing):在龐大的高維空間裡建立捷徑,讓檢索能在毫秒級完成。

有了向量與索引,我們的知識庫正式成為「語意可檢索的資料庫」。不過,光是檢索到候選文件還不夠,排序策略會直接影響答案品質

明天(Day 10),我們會進一步探討 檢索策略設計 (Retriever Design),看看如何把「問 → 找 → 回答」 這個流程設計得更聰明,並串成完整的 RAG Pipeline。

📚 延伸閱讀


上一篇
Day08 - 文件清洗 (Cleaning) 與切片策略 (Chunking Strategies)
系列文
30 天帶你實戰 LLMOps:從 RAG 到觀測與部署9
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言