從 Naive RAG 升級到 Context Governance

📌 本文重點

• Naive RAG 上線後會遇到治理與時效問題
• 將 RAG 拆成索引 / 治理 / 拼接三層
• 用 metadata 做權限、版本、有效期與追溯
• 打通 log 與 trace，讓回答可審計可監控

在專案一開始，我們多半用一個 向量庫 + chunk + 單次檢索 的 Naive RAG 搭起 Demo，看起來效果不錯。但一旦上到真正的多人、多租戶、活資料（會更新、會過期）場景，很快就會踩到幾個痛點：

• 回答引用了 已刪除或過期的文件（時效問題）
• 同一問題在不同時間、不同使用者之間 答案不一致（治理問題）
• 回答內容 無法清楚追溯來源，審計與合規做不到（可追溯性問題）

Context Governance / Context Engineering 要解決的，就是：不要只問「怎麼多抓幾個 chunk」；要開始問「LLM 在推理時到底可以看見什麼」，並把這件事做成一個可控、可監控的架構層。

---

重點說明：從檢索到治理的三層拆分

1. 檢索層拆成三層：索引 / 治理 / 拼接

典型 Naive RAG：

1. embed(text_chunks) → 向量庫
2. query → 相似度搜尋 → top_k chunks → 拼成 prompt

升級成 Context Governance 之後，建議拆成三層：

1. 索引層（Index Layer）
2. 向量索引：文件內容的 semantic embedding

3. 結構化索引：metadata / 權限 / version / valid_from / valid_to / tenant_id 等，存在關聯式 DB 或專門的索引（如 OpenSearch filter，Postgres JSONB）

4. 治理層（Governance Layer）

5. 權限過濾：per-tenant、per-user、per-role
6. 時效與版本控制：只允許在有效期內、最新版本的文件進入 context

7. 來源標註：在 context 裡保留 doc_id / version / source_system / updated_at，為後續 trace 和審計打基礎

8. 拼接層（Context Assembly Layer）

9. 根據任務（QA / summarization / agent 工單處理）選擇不同 prompt template
10. 根據 context policy（例如：最多 N 個來源、每個來源最多 M 個 chunk、必須包含最新公告等）來決定如何拼接

這樣做的核心好處：

• 可控：你能明確說出「這個回答看到了哪些資料」
• 可審計：任何回答都能追溯到具體版本的文件
• 可演進：未來換向量庫、換模型、換圖資料結構，只要維持治理層的 contract 即可

💡 關鍵： 把 RAG 拆成索引、治理、拼接三層，可以在不動底層基礎設施的前提下，持續優化與替換任一層實作。

2. 何時只用 Vector，何時要 GraphRAG / Schema

向量搜尋擅長回答：「跟這段話最像的是什麼？」，但在以下情境常常不夠：

• 跨多個實體、關係的複雜查詢（例如：
• 「這客戶過去一年所有 ticket 中，和某功能相關的主要痛點是什麼？」）
• 知識本身有明確的 schema / 關係（產品 → 版本 → bug → 修補公告）
• 需要 聚合 / 推理 而不是單純摘錄段落

這時考慮引入：

• GraphRAG / 知識圖譜：用圖結構顯式表示實體關係，向量只負責節點 / 邊的語義近似
• 業務 schema：在 DB 或 graph DB 中建模 Customer -[reported]-> Ticket -[relates_to]-> Feature 等結構，RAG 只從圖或 DB 中抓「已整理好」的資料進 context

實務建議：

• FAQ、文件查詢、規範查詢：優先 純 vector + metadata filter
• 關係推理、跨文件關聯、流程追蹤：優先 GraphRAG / schema + vector（vector 幫你找到節點，圖幫你沿著關係走）

3. 在 RAG pipeline 中引入治理：我們需要哪些欄位？

最小可用治理模型，可以從以下 metadata 開始：

• tenant_id：租戶隔離
• visibility / allowed_roles：權限
• version / is_latest：版本
• valid_from / valid_to：有效期
• source / doc_id / chunk_id：來源追蹤

治理層的業務邏輯就是：

• 只選 is_latest = true 且 now BETWEEN valid_from AND valid_to 的文件
• 只選 tenant_id 與當前請求相符，且角色有權限的文件
• 把上述 metadata 一起帶入 context，並在回答後產生 answer trace

💡 關鍵： 把 tenant_id、版本與有效期變成硬性 filter，可以大幅降低資料外洩與用到過期文件的風險。

---

實作範例：Python / TypeScript 實戰片段

以下不是完整專案，而是幾個關鍵點：版本與有效期控制、per-tenant 權限過濾、answer trace、監控與日誌。

1. Python：檢索層 + 治理層

假設向量存在 pgvector，metadata 在同一張表的 JSONB 裡：

# pseudo: Python + asyncpg + pgvector
from datetime import datetime
from typing import List, Dict

async def search_governed_chunks(
    db, *, query_embedding: list[float], tenant_id: str, user_roles: list[str],
    top_k: int = 8
) -> List[Dict]:
    now = datetime.utcnow()

    sql = """
    SELECT
      id, doc_id, chunk_id, content, metadata,
      1 - (embedding <=> $1::vector) AS score
    FROM documents
    WHERE
      metadata->>'tenant_id' = $2
      AND (metadata->>'is_latest')::boolean = true
      AND (metadata->>'valid_from')::timestamptz <= $3
      AND (metadata->>'valid_to')::timestamptz >= $3
      AND (
        metadata->>'visibility' = 'public'
        OR EXISTS (
          SELECT 1 FROM jsonb_array_elements_text(metadata->'allowed_roles') r
          WHERE r = ANY($4::text[])
        )
      )
    ORDER BY embedding <=> $1::vector
    LIMIT $5
    """

    rows = await db.fetch(sql, query_embedding, tenant_id, now, user_roles, top_k)
    return [
        {
            "doc_id": r["doc_id"],
            "chunk_id": r["chunk_id"],
            "content": r["content"],
            "metadata": r["metadata"],
            "score": r["score"],
        }
        for r in rows
    ]

上面這段把 向量相似度 + 權限 + 版本 + 有效期 一次做掉，這就是「治理層」的核心入口。

2. TypeScript：拼接層 + answer trace

假設你用 Node.js + OpenAI API：

// pseudo: TypeScript + OpenAI
import { OpenAI } from "openai";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

type RetrievedChunk = {
  content: string;
  metadata: {
    doc_id: string;
    chunk_id: string;
    version: string;
    source: string;
    updated_at: string;
  };
};

function buildContext(chunks: RetrievedChunk[]): {
  contextText: string;
  trace: any[];
} {
  const trace = chunks.map((c) => c.metadata);

  const contextText = chunks
    .map(
      (c, idx) =>
        `[[SOURCE ${idx + 1}]]\n` +
        `doc_id=${c.metadata.doc_id} version=${c.metadata.version} ` +
        `source=${c.metadata.source} updated_at=${c.metadata.updated_at}\n` +
        `${c.content}\n`
    )
    .join("\n\n");

  return { contextText, trace };
}

export async function answerWithTrace(
  question: string,
  chunks: RetrievedChunk[],
) {
  const { contextText, trace } = buildContext(chunks);

  const prompt = `你是企業知識庫助理，只能根據給定的資料來源回答。\n
如果無法從資料中找到答案，請明確回答「目前資料不足以回答」。\n
=== 已檢索資料 ===\n${contextText}\n
=== 問題 ===\n${question}`;

  const completion = await openai.chat.completions.create({
    model: "gpt-4.1-mini",
    messages: [{ role: "user", content: prompt }],
  });

  const answer = completion.choices[0]?.message?.content ?? "";

  return { answer, trace };
}

這裡的重點：

• context 內嵌了 doc_id / version / source / updated_at，LLM 的回答自然可以參照這些標籤
• 回傳結果中附帶 trace，可以直接存進 answers 表，實現問答可追溯

3. 監控與日誌設計（Python 片段）

最低限度，你應該記：

• 請求 id、tenant_id、user_id
• query、top_k、實際使用的 chunk 數
• 每個 chunk 的 doc_id / version / score
• LLM model / latency / token 數

import logging, time

logger = logging.getLogger("rag")

async def rag_pipeline(request_id: str, tenant_id: str, user_id: str, query: str):
    t0 = time.time()
    # 1. embed
    embedding = await embed_query(query)
    # 2. retrieve + governance
    chunks = await search_governed_chunks(
        db, query_embedding=embedding, tenant_id=tenant_id,
        user_roles=["user"], top_k=8
    )
    # 3. assemble + call LLM
    answer, trace = await call_llm_with_trace(query, chunks)

    latency = time.time() - t0

    logger.info(
        "RAG_REQUEST", extra={
            "request_id": request_id,
            "tenant_id": tenant_id,
            "user_id": user_id,
            "query": query,
            "retrieved_count": len(chunks),
            "trace": trace,
            "latency_ms": int(latency * 1000),
        }
    )

    return answer

有了這層 log，你可以：

• 回頭調查「為何某次回答錯」——看他用了哪個版本的哪份文件
• 做品質分析——哪些 doc/source 經常被引用，哪些很少被用到

💡 關鍵： 系統性記錄 request → used_docs 與 latency，能同時支撐故障排查與長期品質優化。

---

建議與注意事項：常見坑與最佳實踐 Checklist

常見坑

1. 只相信 embedding 相似度
2. 沒有權限 / 版本 / 時效 filter，導致過時、無權資料進 context
3. 不處理刪除 / 更新
4. 檔案在主系統被刪掉或更新，向量庫沒同步，LLM 繼續引用舊版本
5. 忽略權限與租戶隔離
6. 向量庫是共享的，但檢索時沒過濾 tenant_id 或 roles，直接變資料外洩風險
7. 多索引源混用不標註
8. 搜尋結果來自「公告 / 內部 wiki / 工單系統」，但沒有來源標籤，LLM 很難判斷權威度，使用者也無從分辨

可落地的最佳實踐 Checklist

• 索引層
• • [ ] 每個 chunk 至少包含：tenant_id, doc_id, chunk_id, version, is_latest, valid_from, valid_to, source
  • [ ] 有明確的 upsert pipeline，文件更新時同步更新 metadata 與向量
• 治理層
• • [ ] 檢索 API 強制 filter tenant_id + is_latest + 有效期
  • [ ] 權限使用 白名單策略（角色不在 allowed_roles 就直接排除）
  • [ ] 支援 per-tenant 可配置的 context policy（top_k 上限、允許的 source 列表）
• 拼接層
• • [ ] 不同任務（QA / summarization / agent）使用不同 prompt template
  • [ ] 在 context 中保留清楚的 來源標註，並鼓勵 LLM 在回答中引用
• 監控與審計
• • [ ] 每個回答都存 request → used_docs (doc_id, version) 的 trace
  • [ ] 有簡單 dashboard 監控：平均 retrieved_count / top_k 命中度 / 各 source 的引用率

做到以上，你的系統就不是「會講話的向量搜尋」，而是有 治理能力 的企業級 RAG：回答可控、資料可審計、行為可監控，也比較能撐過 Demo 以後的那一階段——真正上線的環境。

🚀 你現在可以做的事

• 審視現有 RAG pipeline，補上 tenant_id、版本、有效期等必要 metadata 欄位
• 在搜尋 SQL 或檢索層實作權限 + 時效 + is_latest 的強制 filter
• 為每次回答記錄 request → used_docs trace，並在 log 或 DB 裡建立最小可用的審計表格