如何在正式環境加入 Temporal RAG

這篇教你在既有 RAG 中加入時間感知重排層，讓新版本、有效期間內的事件與最新資料優先被 LLM 使用。

這篇給正在做內容會隨時間變動的 RAG 系統開發者，像是文件、政策、教學、警報與客服知識庫。照著做完，你會得到一個可上線的時間層，能先排除過期資料、提高有效事件權重，並在不重建 retriever 的前提下偏好較新的版本。

你也會建立一條清楚的分數路徑，介於向量搜尋與 LLM 之間，並能驗證舊文件不再壓過新文件。

開始之前

• Python 3.11+
• Node 20+，僅當你的應用外殼或 API 層使用 Node 時需要
• 可用的 RAG 堆疊，搭配 Pinecone、Weaviate、pgvector 或 FAISS 其中之一
• 文件中已有 created_at、updated_at，以及可選的 expires_at 欄位
• LLM API key，例如 OpenAI 或 Anthropic
• 可存取參考實作的 GitHub repo，以及原始說明文的 Towards Data Science

Step 1: 標記時間敏感文件

這一步的產出是文件清單，能把永不過期的事實、已被新版本取代的內容、以及有時間窗口的事件分開。時間層只有在知道文件屬於哪一類時，才會正確工作。

先替每筆來源資料加上類型與有效性欄位。實用的結構如下：

kind: STATIC | VERSIONED | EVENT
valid_from: ISO-8601 timestamp
valid_to: ISO-8601 timestamp or null
expires_at: ISO-8601 timestamp or null
supersedes_id: optional document id

驗收：你應該能指向任一文件，並說出它是永恆事實、已被新版本取代，或只在某段時間內有效。

Step 2: 把新鮮度欄位寫進索引

這一步的產出是可供重排使用的索引記錄。向量資料庫仍然負責語意搜尋，但它必須把 metadata 一起回傳，讓時間層能做判斷。

在資料匯入或更新時，把時間欄位寫進同一筆嵌入資料。若你已經有 pipeline，只要補欄位，不必更換 embedding model。

doc = {
  "id": "policy_v2",
  "text": "API rate limits are now 200 requests per minute",
  "kind": "VERSIONED",
  "created_at": "2026-01-10T09:00:00Z",
  "updated_at": "2026-03-01T12:00:00Z",
  "valid_from": "2026-03-01T12:00:00Z",
  "valid_to": null,
  "supersedes_id": "policy_v1"
}

驗收：你應該看到每個被取回的 chunk 都帶著 metadata，而不只是文字與相似度分數。

Step 3: 在排序前先判定有效性

這一步的產出是前置判定器，會先移除過期事實，並把正在生效的時間事件標記出來，然後才交給 LLM。這是關鍵安全步驟，因為過期內容不該只是降權，而是要直接移除。

你可以用三種狀態：EXPIRED、VALID、TEMPORAL。依照原始實作模式，只有 EVENT 文件才會變成 TEMPORAL。VERSIONED 文件在未被取代前都維持 VALID，舊版一旦被 supersede 就排除。

def classify(doc, now):
    if doc.get("expires_at") and doc["expires_at"] < now:
        return "EXPIRED"
    if doc["kind"] == "EVENT" and doc.get("valid_from") <= now <= doc.get("valid_to"):
        return "TEMPORAL"
    return "VALID"

驗收：你應該看到過期項目從候選集合中移除，而活躍警報被標成 TEMPORAL，不只是排得比較後面。

Step 4: 用時間衰減重排候選項

這一步的產出是結合語意相似度與新鮮度的重排器。目標不是讓時間壓過相關性，而是在多個近似匹配時，讓較新的內容更有優勢。

實作上可以先正規化 cosine similarity，再依年齡套用指數衰減，並對有效事件加上額外提升。原始文章採用混合分數，讓最新且相關的文件勝出，但不會把所有新文件都推到最前面。

decay = 0.5 ** (age_in_days / half_life_days)
final_score = (0.7 * vector_score) + (0.3 * decay)
if state == "TEMPORAL":
    final_score *= 1.2
if state == "EXPIRED":
    final_score = 0.0

驗收：你應該看到較新的政策壓過語意相近但較舊的政策，而在事件有效期間內，直播告警會跳到靜態文件前面。

Step 5: 把時間層接在 retriever 與 LLM 之間

這一步的產出是可上線的資料流，保留原本 retriever 不動，只在最後一段改寫候選順序。這樣風險低，也容易塞進既有系統。

做法是先取 top-k 語意結果，再分類、重排，最後只把最終 context 傳給 LLM。這會保留你目前的 retriever，同時在最後一個責任點修正時間盲點。

candidates = retriever.search(query, top_k=20)
scored = [score_candidate(c, now) for c in candidates]
ranked = sorted(scored, key=lambda x: x.final_score, reverse=True)
context = [item.doc for item in ranked if item.state != "EXPIRED"]
answer = llm.generate(query, context)

驗收：你應該看到 prompt context 內是最新有效版本，而不是語意最像但最舊的 chunk。

Step 6: 測試舊答案失敗情境

這一步的產出是一組可重複執行的測試，專門驗證最重要的時間錯誤。重點不是一般 retrieval 測試，而是被取代的政策、已過期的教學，以及應該壓過其他資料的即時事件。

先建立至少三種 fixtures：一個被 supersede 的文件、一個過期文件、以及一個正在生效的事件。接著同時檢查排序結果與排除行為。

assert "policy_v1" not in final_context
assert final_rank[0].id == "announcement_today"
assert any(item.state == "TEMPORAL" for item in ranked)

驗收：你應該看到舊版本被移除、活躍通知被提到前面，且最新有效文件排在前一版之前。

常見錯誤

• 把所有文件都套用同一個新鮮度規則。修法：分開 STATIC、VERSIONED、EVENT，讓每種文件有自己的處理方式。
• 只把過期事實降權，沒有真正移除。修法：在 rerank 或組 prompt 前先過濾 EXPIRED 項目。
• 把每個較新的文件都當成緊急事件。修法：只對真正有時間窗口的事件套用 TEMPORAL 提升，不要套用到一般版本更新。

接下來可以看什麼

當這套流程能跑之後，可以再加上來源別 half-life、排序決策的 audit log，以及對使用者可讀的解釋，說明為什麼某份文件勝過另一份。接著就能加入即時刷新、政策專屬過期規則，還有衡量舊答案下降幅度的評估集。