Lab:RAG with canary + provenance
前三課都在拆規則,這一課把 retrieval、packing、canary check 接成一條真的能跑的最小管線。
Hook
到這裡, 你手上其實已經有三塊拼圖。 第一塊是 Lesson 02 的 provenance packing。 第二塊是 Lesson 03 的 canary output check。 第三塊是 Lesson 04 的 fail-closed operational 習慣。 如果這些東西一直分開看, 很容易覺得: 「概念懂了, 但 production 最小版到底長什麼樣子?」 所以這堂 lab 的目標很直接。 我們不接真 memhall。 不接真 vector DB。 不接真 LLM API。 只做一條最小 RAG retrieval pipeline。 它要證明六件事:
- store 裡可以先植入 canary
- retrieval 可以把 entry 包成帶 provenance 的 context
llm_derived與raw_source會被明確標 tier- fake LLM 一旦複製到 poison content,就會把 canary 帶出來
- output check 會命中 canary
- 命中後 flow 直接 fail-closed
換句話說, 這不是在展示 fancy 架構。 這是在做最小可執行證明。 只要這段 reference code 你能自己跑通, 你就已經把本課三個核心觀念接起來了。
Learn
這個 lab 在驗什麼
先把成功標準講清楚。 這堂課不是在證明:
- 你已經做出 production-grade retriever
- 你已經解完 ranking、dedup、rerank
- 你已經接上真實模型與真實 observability
它要驗的是比較小、 但非常關鍵的鏈條:
| 驗證點 | 你要看到什麼 |
|---|---|
| canary injection | poison entry 真的被植入 token |
| provenance packing | 每筆 retrieved entry 都有 source 與 tier |
| runtime leak | fake LLM 複誦 context 時會把 canary 帶出來 |
| fail-closed | 命中 canary 後直接停止 serve |
| clean control | 移除 poison entry 後同樣流程不會誤報 |
做到這五件事, 就已經足夠說明: 你的 context boundary 不是嘴上講講。 它真的在 runtime 有 enforcement。
我們只需要三個 function
這份 lab 的骨架故意很小。 只有三個核心 function:
def inject_canaries(store: dict) -> list[str]:
...
def pack_retrieved(query: str, store: dict) -> tuple[str, list[str]]:
...
def output_check(llm_output: str, canaries: list[str]) -> str | None:
...
其中:
inject_canaries(...)負責把 token 寫進指定 entrypack_retrieved(...)負責 retrieve 並把 provenance 一起包進 contextoutput_check(...)負責在輸出層做 deterministic 字串檢查
你可以把它想成: 最小 poison setup、 最小 retrieval contract、 加上最小 output gate。
store 結構刻意很土
這份 demo 不做 class hierarchy。
直接用 dict[str, dict]。
原因很簡單:
本課要看的是 boundary,
不是 object design。
我們只需要每筆 entry 至少有:
titlebodytier
然後對一筆 poison entry 額外加上:
canary_slot
只要資料結構夠平, 你等一下看 retrieval 與 packing 時就比較不會被語法分心。
canary token 為什麼做成 deterministic 生成
真實系統裡你可以真的隨機生成。 但在教學 demo 裡, 比較好的做法通常是 deterministic。 因為:
- 更容易重現
- 更容易自驗
- 出問題時更容易對照 entry id
所以這份 lab 會用 sha256(entry_id) 的前八碼,
拼成:
MK_CANARY_<digest>
它已經夠不像自然文字, 也夠穩定。
retrieve 的規則也故意很簡單
這份 demo 不做 embedding search。
只做很粗的 term match。
query 裡長度大於二的詞,
只要出現在 title + body 裡,
就算一分。
然後按分數高到低取前三筆。
這很陽春。
但足以讓 poison entry 因為提到 deploy 而被召回。
請注意,
這裡的目的不是模擬真實檢索品質。
而是保證示範中的 poison path 會被跑到。
packing 要把 provenance 放在最終字串裡
Lesson 02 已經說過,
provenance 不能只存在 store metadata。
因為下游 LLM 不會自己查 DB。
所以這份 lab 的 pack_retrieved(...)
會直接把每筆 entry 包成:
<context source="memhall:eX" tier="...">
這樣 fake LLM 複誦整段內容時, 你也能一眼看出哪筆資料進了 prompt。
output check 只回答一件事
output_check(...) 的設計刻意窄。
它不負責語義審查。 不負責 PII 分析。 不負責 judge 回答有沒有偏掉。 它只回答一件事: 這段輸出裡有沒有出現任何 canary。 窄, 正是它可靠的原因。
Reference code:最小 production RAG 骨架
下面這段 code 是完整可跑的。
只用 Python 標準函式庫。
它會跑兩輪:
第一輪保留 poison entry,
第二輪移除 poison entry 做對照組。
你應該會看到第一輪命中 canary 並印出 fail-closed: 503,
第二輪則印出 clean。
這段 code 的重點不在 retrieval 品質。
而在於你真的把 poison path 走完了。
同一段 pipeline,
只有是否包含 poison entry 這個差異,
結果就從 LEAKED 變成 clean。
這就是 boundary 實驗該長的樣子。
你應該看到的輸出
如果一切正常, 你至少會看到這種結構:
attack_ids = ['e1', 'e2', 'e5']
LEAKED: MK_CANARY_xxxxxxxx
fail-closed: 503
clean_ids = ['e1', 'e2']
clean
canary 後面的八碼會依 e5 的 digest 決定。
但整體流程不會變。
為什麼 inject_canaries(...) 要在 retrieve 前跑
因為我們要模擬的不是輸出後才補 canary。 而是 poison entry 本來就藏在 store 裡。 這比較接近真實情境。 當 retrieval 發生時, 它會把已被植入 token 的 entry 帶出來。 如果你把 canary 放在 retrieve 之後才動態加上, 那你測到的是 transport path。 不是 memory poisoning path。
為什麼 pack_retrieved(...) 回傳 retrieved_ids
因為一旦出事, 你需要快速知道是哪些 entry 進了 prompt。 只回傳 packed string 當然也能跑。 但可觀測性會差很多。 所以這份 lab 故意讓 function 回:
(context_str, retrieved_ids)
這樣之後不管你要 log、 trace、 或 debug, 都會比較容易。
為什麼 clean control 很重要
很多示範只做 attack case。
那樣其實不夠。
因為你無法分辨:
是偵測真的抓到了 poison,
還是你的檢測器本來就一直誤報。
所以這份 lab 刻意多跑一輪 clean control。
把 e5 拿掉後,
其他流程不變。
如果 output_check(...) 仍然命中,
那代表你的檢測器有問題。
如果它回到 clean,
你才比較能確定:
這次命中的確跟 poison entry 有關。
這一課要帶走的三個名字
如果只能記三個 function, 就記:
inject_canaries(...)pack_retrieved(...)output_check(...)
第一個負責放探針。 第二個負責把 provenance 帶進 prompt。 第三個負責在輸出層踩煞車。 這三個一起工作, 才是一條最小但完整的 context-boundary pipeline。
AI 協作:學了這個,跟 AI 怎麼配合?
這個技能在 AI 協作中的定位, 是讓你可以要求 AI 先幫你拼出最小 reference pipeline, 然後你再把真實 store、 真實 retriever、 真實 middleware 接上去。
你的人類優勢:
- 只有你知道哪個 query flow 屬於高風險,值得優先接 canary 與 fail-closed。
- 只有你能判斷某些
llm_derivedentry 在你的產品裡是否該完全排除,還是只能在內部研究模式保留。
可以這樣跟 AI 說:
請幫我做一個最小 RAG pipeline,包含 canary 注入、retrieved context provenance packing、以及輸出後的 deterministic leak check。我要一段可跑的 Python reference code,能示範 poison entry 存在時印出
LEAKED與fail-closed: 503,移除 poison entry 後印出clean。
Do
互動示範
挑戰任務
請修改 output_check(llm_output, canaries),除了完整命中外,也支援 partial match。規則:如果輸出裡出現某個 canary 的前 15 個字元,也算洩漏,並回傳完整 canary。測試資料請用 llm_output = "prefix leaked MK_CANARY_1234" 與 canaries = ["MK_CANARY_1234ABCD"]。最後印出回傳值。