Kerwin AI Information

標籤: Context Governance

  • 從 Naive RAG 升級到 Context Governance

    從 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 裡建立最小可用的審計表格