🔹 前言

昨天（Day 12）我們談到了 知識庫資料管理：

• 多種來源整合 → 把 PDF、Web、API 等不同來源轉換成統一格式。
• 資料版本控制 → 用 DVC / LakeFS 等工具，確保知識庫可追溯、可回滾。

但是，光有版本控制還不夠。 在真實世界裡，知識會隨著時間而變動 (Drift)，如果我們的系統沒有及時更新，就會回答錯誤的資訊。所以，今天我們的重點就是：

👉 如何發現變動以及更新的策略介紹

🔹 什麼是 Data Drift？

在機器學習裡，Data Drift 指的是 資料分布隨時間改變。知識不像硬碟檔案一成不變，它會隨時間更新。如果我們不跟上，就好像還在用十年前的百科全書查資料——結果自然會出錯。

在 RAG 系統裡， Data Drift 分為三種：

1. 概念漂移 (Concept Drift)

   • 問題的含義隨著時間改變了。
   • 例子：
     • 2022 年問「Twitter CEO 是誰？」答案是 Parag Agrawal。
     • 2025 年答案變成 Elon Musk。

2. 資料漂移 (Data Drift)

   • 新文件出現，舊文件過時。
   • 例子：FAQ 文件改版、新法規公布。

3. 檢索漂移 (Retrieval Drift)

   • 文件結構變了，導致相同 query 檢索到不同內容。
   • 例子：
     • 官網改版，FAQ「退款政策」原本在第 5 題，後來改到第 8 題，舊的檢索方式就抓不到了。
     • 某些網頁路徑失效。

🔹 為什麼知識更新很重要？

• 正確性：確保模型回答的是「最新的事實」。
• 信任度：使用者若發現答案過時，會懷疑系統可靠性。
• 合規性：在醫療、金融、法規等場景，過時的資訊可能帶來法律風險。

🔹 知識更新策略

1. 定期重建（Batch Update）

• 做法：每天 / 每週重新爬資料，重新建索引。
• 優點：實作簡單，保證最新。
• 缺點：成本高、效能差。

2. 增量更新（Incremental Update）

• 做法：只更新新增 / 修改過的文件。
• 優點：效率高。
• 缺點：需要設計「變更偵測」機制。

3. 熱點更新（On-demand Update）

• 做法：遇到查詢時，若發現知識庫沒有答案，就即時去外部抓資料。
• 優點：確保關鍵查詢總是最新。
• 缺點：延遲高，依賴外部 API。

實際例子：

完整可執行程式碼 已經放在 GitHub。

🔹 Demo 1 : 偵測知識需要被更新的方式

1. 基於 Metadata

• 比較檔案大小、hash、修改時間。
• 適合檔案型來源（PDF、Markdown、程式碼、報表、數據檔）。

import os, hashlib

def get_file_hash(path):
    hash_md5 = hashlib.md5()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):  # 分塊計算避免一次吃光記憶體
            hash_md5.update(chunk)
    return hash_md5.hexdigest()

file_path = "faq.pdf"
print(os.path.getmtime(file_path), get_file_hash(file_path))

執行結果：

# 把 Day12 的 worker_manual.pdf 移過來 Day13 用
❯ python metadata_comparison.py
📥 第一次建立 metadata 紀錄。

# 檔案沒有變動的情況下再執行一次
❯ python metadata_comparison.py
📂 檔案: worker_manual.pdf
  - 之前修改時間: 2025-09-27 11:17:25
  - 之前 Hash: 5106549b250c4c06a9bd1e59ab950e8a
  - 目前修改時間: 2025-09-27 11:17:25
  - 目前 Hash: 5106549b250c4c06a9bd1e59ab950e8a
✅ 檔案內容無變更。

# 修改 worker_manual.pdf 後儲存，再執行一次
❯ python metadata_comparison.py
📂 檔案: worker_manual.pdf
  - 之前修改時間: 2025-09-27 11:17:25
  - 之前 Hash: 5106549b250c4c06a9bd1e59ab950e8a
  - 目前修改時間: 2025-09-27 11:25:57
  - 目前 Hash: 0323bd12d945343f73035ab4503b2e3d
⚠️ 檔案內容已變更，需要更新知識庫！

2. 基於版本號

• 文件本身有版本號 / 發布日期（法規、API Schema）。
• 直接比對 metadata 決定是否更新。

# version_check.py
import yaml

def get_version(path: str) -> str:
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    return data.get("version", "0.0.0")

def compare_versions(old_v: str, new_v: str) -> bool:
    """若版本號升級，回傳 True"""
    def parse(v): return list(map(int, v.split(".")))
    return parse(new_v) > parse(old_v)

old_file = "faq_v1.yaml"
new_file = "faq_v2.yaml"

old_version = get_version(old_file)
new_version = get_version(new_file)

print(f"舊版: {old_version}, 新版: {new_version}")
if compare_versions(old_version, new_version):
    print("⚠️ 偵測到新版本，需要更新知識庫！")
else:
    print("✅ 沒有新版本，無需更新。")

這邊可以準備兩個檔案：

# 範例輸入檔（faq_v1.yaml）
version: 1.0.0
faq:
  - q: 退貨政策
    a: 商品需在 7 天內退回

# 範例輸入檔（faq_v2.yaml）
version: 1.1.0
faq:
  - q: 退貨政策
    a: 商品需在 14 天內退回
  - q: 是否支援線上客服？
    a: 是的，請至官網點選聊天室

執行結果：

❯ python version_check.py
舊版: 1.0.0, 新版: 1.1.0
⚠️ 偵測到新版本，需要更新知識庫！

3. 基於差異檢測

• 對比舊版與新版的內容差異。
• 常見方法：Diff 工具、Embedding 相似度比較。
• 較適合 語意層面需要追蹤更新 的場合，比方說法律條文、財務報表、合約的版本比對。

# embedding_comparison.py
from openai import OpenAI
from dotenv import load_dotenv
import numpy as np
import os

# 載入 .env 檔案
load_dotenv()

# 讀取 OPENAI_API_KEY
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("沒有找到 OPENAI_API_KEY，請檢查境變數！")

# 初始化 OpenAI client
client = OpenAI()

def embedding(text: str):
    """取得文字的向量表示"""
    return client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    ).data[0].embedding

def cosine_similarity(v1, v2):
    """計算兩個向量的餘弦相似度"""
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

# 舊版 FAQ vs 新版 FAQ
old_text = "退貨政策：商品需在 7 天內退回，且保持完整包裝。"
new_text = "退貨政策：商品需在 14 天內退回，並保持原始包裝未損壞。"

vec_old = embedding(old_text)
vec_new = embedding(new_text)

similarity = cosine_similarity(vec_old, vec_new)
print(f"語意相似度: {similarity:.4f}")

# 設定門檻（可依需求調整，例如 0.95）
if similarity < 0.95:
    print("⚠️ 偵測到重要差異，需要更新知識庫！")
else:
    print("✅ 差異不大，可以不用更新。")

執行結果：

❯ python embedding_comparison.py
語意相似度: 0.9096
⚠️ 偵測到重要差異，需要更新知識庫！

這邊列出了適合 / 不適合差異檢測的各種場景：

完整可執行程式碼 已經放在 GitHub。

🔹 Demo 2 : 檔案改動後更新索引

目標

模擬一個知識庫更新流程：

1. 偵測檔案是否更新
2. 若有變動 → 重新做 embedding & 更新索引
3. 若無變動 → 跳過

註：以下 pipeline.py 的向量化採用 本地 hashing trick（無需 API Key），
方便 Demo 與 CI/CD 測試，不會額外消耗 OpenAI Token。
真實場景可改成 OpenAI Embeddings 或自家模型 (詳見 README.md)。

# pipeline.py
from __future__ import annotations
import os
from steps.detect import detect_change, save_hash
from steps.embed import embed_texts
from steps.update_index import write_index

SRC_FILE = "data/faq.txt"
META_HASH = "artifacts/source.hash"
INDEX_OUT = "artifacts/vector_index.json"

def simple_chunk(text: str, max_chars: int = 400) -> list[str]:
    # 極簡切片：依段落與長度切
    parts = []
    for para in text.splitlines():
        if not para.strip():
            continue
        buf = para.strip()
        while len(buf) > max_chars:
            parts.append(buf[:max_chars])
            buf = buf[max_chars:]
        if buf:
            parts.append(buf)
    return parts or ["(空文件)"]

def main() -> None:
    print("🔍 檢查檔案是否變動…")
    res = detect_change(SRC_FILE, META_HASH)
    if not res.changed:
        print("✅ 無變動，跳過更新")
        return

    print("⚠️ 偵測到變動，開始更新索引…")
    with open(SRC_FILE, "r", encoding="utf-8") as f:
        text = f.read()

    chunks = simple_chunk(text)
    print(f"✂️ 切出 {len(chunks)} 個片段")

    vectors = embed_texts(chunks)
    print(f"🧠 產生 {len(vectors)} 筆向量")

    records = [
        {"id": i, "text": chunks[i], "vector": vectors[i]}
        for i in range(len(chunks))
    ]
    write_index(INDEX_OUT, records)
    save_hash(META_HASH, res.new_hash)
    print(f"📦 已寫入索引：{INDEX_OUT}")
    print("🎉 更新完成！")

if __name__ == "__main__":
    # 確保資料夾存在
    os.makedirs("data", exist_ok=True)
    os.makedirs("artifacts", exist_ok=True)
    if not os.path.exists(SRC_FILE):
        with open(SRC_FILE, "w", encoding="utf-8") as f:
            f.write("公司員工手冊 v1.0：\n第二章 加班與補休：加班需事前申請，工時可折換補休。\n")
    main()

執行結果：

# 第一次跑
❯ python pipeline.py
🔍 檢查檔案是否變動…
⚠️ 偵測到變動，開始更新索引…
✂️ 切出 2 個片段
🧠 產生 2 筆向量
📦 已寫入索引：artifacts/vector_index.json
🎉 更新完成！

# 修改 data/faq.txt 後跑第二次
❯ python pipeline.py
🔍 檢查檔案是否變動…
⚠️ 偵測到變動，開始更新索引…
✂️ 切出 3 個片段
🧠 產生 3 筆向量
📦 已寫入索引：artifacts/vector_index.json
🎉 更新完成！

# 都不修改檔案，直接跑第三次
❯ python pipeline.py
🔍 檢查檔案是否變動…
✅ 無變動，跳過更新

今天我們展示了如何偵測版本過舊以及簡單的更新示範，而在 Day23 - 再訓練與知識迭代 (Retraining & Continual Learning) 我們會示範增量更新實際的作法。

🔹 常見陷阱 vs 最佳實務

🔹 生活化比喻

想像一座圖書館，知識庫就像它的館藏。

• 資料版本控制：圖書館會保存不同版本的書籍，例如第一版、第二版，讓人能回頭查舊資料。
• Data Drift / 知識更新：館藏需要不斷更新，不能一直只擺 1999 年的百科全書。

如果圖書館不更新，你在 2025 年查「世界人口」，卻還讀到「2000 年人口 60 億」，這樣的答案就失去參考價值。

同樣地，三種 Drift 也能在圖書館裡找到對應：

1. 概念漂移 (Concept Drift)：字典裡「手機」的定義，從過去的大哥大變成今天的智慧型手機。
2. 資料漂移 (Data Drift)：教科書出了新版，舊版內容就不再適用。
3. 檢索漂移 (Retrieval Drift)：
   • 圖書館調整了館藏編碼，原本在「005.3」的書，現在換到「007.1」。
   • 如果目錄不更新，你就會找不到書。

🔹 小結

Day 13 的重點：

• Data Drift 是知識隨時間變動的必然現象。
• 常見的知識更新策略： 定期重建 / 增量更新 / 熱點更新
• 更新偵測方式：Metadata、版本號、差異比對。
• 目標是：保持知識庫永遠「新鮮」，回答始終正確。

明天（Day 14），我們將進入 Pipeline 自動化，看看如何讓這些「文件清洗 → 向量化 → 索引更新」的流程完全自動化，減少人力負擔。

📚 引用 / 延伸閱讀

• Data Drift vs Concept Drift
• PureStorage - 什麼是資料飄移？ 模型漂移神秘化
• What is data drift in ML, and how to detect and handle it
• Understanding Data Drift and Model Drift: Drift Detection in Python