第 7 課 · 殺手鐧:讓它記得上週、只報 diff

第 7 課 · 殺手鐧:讓它記得上週、只報 diff

回到第 1 課那份週報,那句最不起眼、卻最值錢的話:

「上週還是 $12。」

這一課,我們要造出能說出這句話的那台機器

在此之前,我們的 graph 有個致命的失憶症:它會抓料(第 3 課)、會平行跑很多競品又去重(第 4 課)、會串進控制面(第 5 課)、會停下來讓真人審一眼(第 6 課)——但每次跑完,它就忘得一乾二淨。這週跑完,下週再跑,它是張白紙,根本不知道「上週 Acme 賣多少」。既然不知道上週,它就永遠說不出「漲了」。

diff(只報變化)是這個產品的靈魂,而 diff 的前提,是記得上週。 這一課就是補上這個記憶,並把它變成護城河。

這一課我們回答四個問題:

  1. thread 與 checkpointer 到底是什麼——為什麼有了它,同一個競品的追蹤線可以「上週跑完的 state 這週還在」,而一次性 LLM 呼叫根本做不到?
  2. checkpointer 跟 Store 怎麼分工——競品情報週報各自該用哪個、放什麼?
  3. 「只報 diff」怎麼實作——這段純邏輯怎麼寫、怎麼離線測?它跟第 4 課的**去重(dedup)**差在哪?
  4. 為什麼「跨執行狀態+只報 diff」是護城河,是一次性 prompt/toy 抄不走的核心價值?

💡 先講清楚本課的邊界:checkpointer 內部怎麼把 state 存下來、怎麼做時光回溯,那是「自己造編排引擎」的題目,屬於姊妹系列 ai-pipeline-from-scratch。這一課只教怎麼用 LangGraph.js 現成的 checkpointer 原語出貨。一句話分工:手刻懂原理,框架去出貨。


一、失憶的顧問:為什麼一次性呼叫給不出 diff

先用一個比喻把問題釘死。

你有兩種方式盯競品。第一種:每週花錢,重新雇一個從沒見過你公司的顧問。他很聰明,一天內就能把 Acme 的官網、定價、changelog 全看一遍,交給你一份 40 頁的「Acme 現況報告」。問題是——他不知道上週發生什麼,所以他只能把「現在的全部」倒給你,沒辦法告訴你「哪裡變了」。想知道變化?你得自己翻出上週那份報告,一頁一頁比。

第二種:雇一個跟了你一年的分析師。他一進門就說:「欸,就一件事——Acme 的 Pro 從 $12 漲到 $15 了,這是這週唯一的新聞,其他都沒動。」

第二個人值錢的地方,不是他更聰明,是他記得。同樣一顆聰明的腦,加上「記得上週」,價值差一個量級。

一次性 LLM 呼叫,就是那個每週重雇的顧問。這不是它笨,是它的唯一記憶就是這一次呼叫的 context——呼叫一結束,context 蒸發,下次呼叫又是一張白紙。

「那我把上週的資料塞進 prompt 不就好了?」

可以,但你有沒有發現——這麼做的瞬間,「記得上週」這件累活,就從機器身上掉回你身上了。你得自己在某個地方存住每個競品的上週狀態、下週開跑前自己撈出來、拼進 prompt、跑完再把這週的存回去,而且要撐過機器重啟、撐過三個月後你早忘了當初怎麼存的。換句話說:沒有 checkpointer,你不是不能記得上週——你是得自己親手變成一個 checkpointer。

而「親手變成一個 checkpointer」——把 state 持久化、跨執行撈回、按追蹤線分開存——這正是 LangGraph.js 已經幫你做好的事。你當然可以自己刻一套(那是 ai-pipeline-from-scratch 的樂趣),但我們這門課要出貨,所以直接用現成的。

💡 其實你第 6 課已經偷用過它了。 那課我們用 interrupt() / resume 做 HITL:graph 停在審核關卡、等真人按下核准、再從斷點恢復。能停住不丟失、能原地恢復,靠的就是一個記憶體版的 checkpointer——暫停時把 state 存住、resume 時撈回來。當時我們「只用不解釋」。這一課補上正式定義,並把同一個原語從「暫停幾分鐘等人」升級成「記得整整一週」。停 5 分鐘等人審,和「停」一整週等下次排程,對 checkpointer 來說是同一件事:state 存住、下次撈回。


二、thread 與 checkpointer:正式定義

兩個詞,一次講清楚。

thread(一條追蹤線):一條跨越多次執行的追蹤線。在我們的產品裡,一個競品就是一條 thread——「acme-corp 這個客戶正在追的 Acme Planner」是一條 thread,「同一個客戶追的 BoltBoard」是另一條。每條 thread 有一個穩定、唯一的 thread_id(例如 "acme-corp::acme-planner"),它是「哪條追蹤線」的名牌。

checkpointer(跨執行狀態):一個掛在 graph 上的持久化層,它把單一 thread 的 state 存下來,跨執行還在。它的工作節奏很單純:

  • graph 開跑前,checkpointer 按 thread_id撈回這條 thread 上次跑完的 state
  • graph 每跑完一步,checkpointer把最新的 state 存回去(所以中途掛了也能從斷點續,不用整條重跑)。

把兩者擺一起,「記得上週」就有著落了:每個競品一條 thread,checkpointer 幫每條 thread 存住它自己的 state。這週開跑,先撈回上週那條 thread 的 state(裡面有「上週 Acme 是 $12」),比對完再把「這週是 $15」存回去,留給下週。

⚠️ 別把 checkpointer 想成「你要自己去讀寫的一張資料表」。它的重點是跟 graph 的生命週期綁在一起、自動觸發:你只要在編譯 graph 時把它掛上、在呼叫時給一個 thread_id,撈回與存回就自動發生,而且用的是你第 2 課已經學過的 state/reducer 語意。你當然「可以」自己在節點裡手動讀寫資料庫達到類似效果——但那就是在重刻一個 checkpointer,而重刻的原理留給 ai-pipeline-from-scratch。

動手:兩次執行,同一條 thread

來把它跑起來。我們刻意讓程式碼離線可跑、不需要任何金鑰——用記憶體版的 MemorySaver 當 checkpointer,用 stub 假資料頂替「抓取」(真實抓取是第 3 課的事,這裡不是重點)。焦點只有一個:證明「上週的 state 這週還在」。

先把 graph 需要的 state 定好。這條 thread 要記住四樣東西:

// graph.ts
import {
  Annotation,
  StateGraph,
  START,
  END,
  type LangGraphRunnableConfig,
  type BaseCheckpointSaver,
} from "@langchain/langgraph";
import { computeDiff, type Snapshot, type Change } from "./diff";

// 這條 thread 的共享 state
const StateAnnotation = Annotation.Root({
  competitor: Annotation<string>,
  // 這次執行抓到的:每次跑都被覆蓋
  current: Annotation<Snapshot | null>({
    reducer: (_prev, next) => next,
    default: () => null,
  }),
  // 上週存下來的:靠 checkpointer 跨執行留著
  lastSnapshot: Annotation<Snapshot | null>({
    reducer: (_prev, next) => next,
    default: () => null,
  }),
  // 這週的 diff 結果
  changes: Annotation<Change[]>({
    reducer: (_prev, next) => next,
    default: () => [],
  }),
});
type State = typeof StateAnnotation.State;

💡 reducer: (_prev, next) => next 意思是「這個欄位每次執行直接覆蓋、不累加」。state/reducer 的完整定義在第 2 課,這裡只是用它。真正的主角是 lastSnapshot——它就是那個「跨執行還在」的欄位

接著是抓取的 stub 和兩個節點。抓取節點在真實產品裡會呼叫第 3 課的工具+LLM 把競品現況抽成結構化的 Snapshot;這裡用固定 fixture 頂替,好讓 demo 每次跑結果一樣:

// graph.ts(續)

// 離線 stub:真實版=第 3 課的抓取工具 + LLM 抽成結構化 snapshot
const FIXTURES: Record<string, Record<number, Snapshot>> = {
  "Acme Planner": {
    26: { price: 12, features: ["看板", "甘特圖"], headline: "團隊協作", fetchedAt: "2026-06-29T03:00:00Z" },
    27: { price: 15, features: ["看板", "甘特圖"], headline: "團隊協作", fetchedAt: "2026-07-06T03:00:00Z" },
  },
};

async function fetchSnapshot(competitor: string, week: number): Promise<Snapshot> {
  const snap = FIXTURES[competitor]?.[week];
  if (!snap) throw new Error(`stub 沒有 ${competitor}${week} 週的資料`);
  return snap;
}

// node 1:抓這週的狀態
async function fetchNode(state: State, config: LangGraphRunnableConfig): Promise<Partial<State>> {
  // 「第幾週」只是 demo 用來挑固定資料的旋鈕,屬於測試關注點,不是產品 state,
  // 所以放在 config、不放進 state。真實版這裡直接抓「現在」。
  const week = config.configurable?.asOfWeek as number;
  const current = await fetchSnapshot(state.competitor, week);
  return { current };
}

// node 2:比上週、算 diff,並把「這週」滾成「下週的上週」
async function diffNode(state: State): Promise<Partial<State>> {
  const changes = computeDiff(state.lastSnapshot, state.current); // 純邏輯,見下一節
  return { changes, lastSnapshot: state.current };
}

// 建圖 + 掛 checkpointer
export function buildGraph(checkpointer: BaseCheckpointSaver) {
  return new StateGraph(StateAnnotation)
    .addNode("fetch", fetchNode)
    .addNode("diff", diffNode)
    .addEdge(START, "fetch")
    .addEdge("fetch", "diff")
    .addEdge("diff", END)
    .compile({ checkpointer }); // ← 就這一行,graph 從此有記憶
}

注意 diffNode 最後那句 lastSnapshot: state.current——這是讓它記得上週的關鍵動作:把「這週抓到的」寫進 lastSnapshot,checkpointer 就會把它存住,變成「下週的上週」。 一週一週往前滾。

現在跑給你看。重點在第二次呼叫:它是一次全新的 invoke,我們沒手動把上週資料傳進去,但它記得。

// demo.ts —— 直接 `npx tsx demo.ts` 就能跑(免金鑰、免資料庫)
import { MemorySaver } from "@langchain/langgraph";
import { buildGraph } from "./graph";

async function main() {
  const graph = buildGraph(new MemorySaver());
  const threadId = "acme-corp::acme-planner"; // 這條 thread = 這個帳號追的 Acme Planner

  // 第 26 週:第一次見到 Acme,沒有上週可比
  const w26 = await graph.invoke(
    { competitor: "Acme Planner" },
    { configurable: { thread_id: threadId, asOfWeek: 26 } },
  );
  console.log("第 26 週 changes:", w26.changes);
  // → [](建基準,沒有「變化」可談)

  // 第 27 週:同一個 thread_id 再跑一次。這是全新的 invoke,
  // 我們什麼上週資料都沒手動傳——但它記得上週是 $12。
  const w27 = await graph.invoke(
    { competitor: "Acme Planner" },
    { configurable: { thread_id: threadId, asOfWeek: 27 } },
  );
  console.log("第 27 週 changes:", w27.changes);
  // → [{ field: "price", severity: "red",
  //      summary: "Pro 方案從 $12/座 變成 $15/座(+25%)" }]

  // 對照組:換一條從沒跑過的 thread,同樣餵第 27 週的資料
  const cold = await graph.invoke(
    { competitor: "Acme Planner" },
    { configurable: { thread_id: "acme-corp::acme-planner::COLD", asOfWeek: 27 } },
  );
  console.log("冷啟動 changes:", cold.changes);
  // → [] 同樣的輸入,但這條 thread 沒有「上週」,就給不出 diff

  // 掀開看 checkpointer 幫這條 thread 存了什麼
  const persisted = await graph.getState({ configurable: { thread_id: threadId } });
  console.log("acme-planner 現在存著的 lastSnapshot:", persisted.values.lastSnapshot);
  // → { price: 15, features: [...], headline: "團隊協作", fetchedAt: "2026-07-06..." }
}

main();

停下來品一下這三段輸出,整課的重點都在這裡:

  • 第 27 週那次呼叫,是徹頭徹尾獨立的一次執行。 換到真實世界,它是下週一凌晨三點被排程叫醒的、跟第 26 週那次隔了七天、中間程序可能重啟過八百次。我們沒手動餵任何上週資料。它照樣算得出「$12 → $15」——因為 checkpointer 按 thread_id 把上週的 lastSnapshot 撈了回來。這就是一次性呼叫做不到的事。
  • 冷啟動那條 thread,輸入一模一樣,卻給不出 diff。 這是最乾淨的對照實驗:diff 算得出來,不是因為資料,是因為記憶。同樣的第 27 週資料,有上週記憶的 thread 說「漲了」,沒記憶的 thread 只能聳肩。
  • getState 讓你直接掀開看 checkpointer 到底存了什麼——lastSnapshot 真的在裡面。它不是玄學,就是一份被持久化的 state,你隨時查得到。
flowchart LR subgraph W26["第 26 週執行"] A26["fetch → current = $12"] --> B26["diff:上週=null → 無 diff"] B26 --> C26["寫回 lastSnapshot = $12"] end subgraph W27["第 27 週執行(七天後、全新一次 invoke)"] A27["fetch → current = $15"] --> B27["diff:上週=$12 → 漲了!"] B27 --> C27["寫回 lastSnapshot = $15"] end C26 -.->|checkpointer 按 thread_id 存住再撈回| A27

⚠️ MemorySaver 只是用來看語意的,別上線

MemorySaver 把 state 存在記憶體裡——程序一重啟就全沒了。它拿來理解語意、跑測試、寫 demo 剛剛好,但你絕不能拿它上線:一次部署、一次當機,所有客戶的「上週記憶」瞬間歸零,下週每個競品都變回「第一次見到」,diff 全滅。

好消息是:checkpointer 是一個可抽換的介面,換後端只換一行。要真正撐過重啟,本機開發可以用 SQLite:

// 要真正持久化(撐過程序重啟):換掉 MemorySaver 就好,graph 其他一個字都不用改
import { SqliteSaver } from "@langchain/langgraph-checkpoint-sqlite";

const checkpointer = SqliteSaver.fromConnString("./digest.sqlite");
const graph = buildGraph(checkpointer); // 語意完全一樣,只是這次寫進磁碟

正式上線要接 Postgres、還要處理多租戶(每個付費客戶資料隔離)與 thread routing——那是第 10 課的專題,這裡先不展開。你現在只要記住這句話:語意在本課學完,換更硬的後端只是換一行。


三、只報 diff:把「變化」算出來(純邏輯 + 離線測)

上一節 diffNode 裡那句 computeDiff(state.lastSnapshot, state.current),是這個產品最值錢的一段程式碼。現在把它攤開。

關鍵設計原則:computeDiff 是一段純邏輯,它不碰網路、不碰 LLM、不需要金鑰。 它就是「拿兩個 Snapshot、回傳變了什麼」的一個函式。這正是全系列反覆強調的 seam(LLM 邊界):把純邏輯跟真正打 LLM 的節點在接縫上切開。抓取(要打 LLM、要連網)在 seam 的一邊,diff(純計算)在另一邊。好處待會你就看到——純邏輯這一邊,可以離線、無金鑰、毫秒級地測到爽。

// diff.ts —— 純邏輯:不碰網路、不碰 LLM、不需要金鑰
export interface Snapshot {
  price: number | null;    // Pro 方案每座價格(美元)
  features: string[];      // 已上線的功能名稱
  headline: string | null; // 官網首頁主標
  fetchedAt: string;       // 這次抓取的時間戳(永遠會變 → diff 必須忽略它)
}

export interface Change {
  field: "price" | "feature" | "headline";
  severity: "red" | "yellow";
  summary: string;
}

/**
 * 拿上週的 snapshot 比這週的 snapshot,只回傳「變了的東西」。
 * - prev 是 null:這條 thread 第一次跑,沒有上週可比 → 沒有 diff(這週就是基準)。
 * - 純函式:同樣的輸入永遠給同樣的輸出,可離線單元測試。
 */
export function computeDiff(prev: Snapshot | null, curr: Snapshot | null): Change[] {
  if (!curr) return [];
  if (!prev) return []; // 第一次見到這個競品,沒有「變化」可談,只建基準
  const changes: Change[] = [];

  // 1) 價格:變動達 10% 標紅,否則標黃
  if (prev.price !== null && curr.price !== null && prev.price !== curr.price) {
    const pct = Math.round(((curr.price - prev.price) / prev.price) * 100);
    changes.push({
      field: "price",
      severity: Math.abs(pct) >= 10 ? "red" : "yellow",
      summary: `Pro 方案從 $${prev.price}/座 變成 $${curr.price}/座(${pct > 0 ? "+" : ""}${pct}%)`,
    });
  }

  // 2) 新增的功能:只報「這週有、上週沒有」的
  const prevFeatures = new Set(prev.features);
  for (const f of curr.features) {
    if (!prevFeatures.has(f)) {
      changes.push({ field: "feature", severity: "red", summary: `上線新功能:${f}` });
    }
  }

  // 3) 首頁主標換了
  if (prev.headline !== null && curr.headline !== null && prev.headline !== curr.headline) {
    changes.push({
      field: "headline",
      severity: "yellow",
      summary: `首頁主標從「${prev.headline}」改成「${curr.headline}」`,
    });
  }

  // 注意:我們從頭到尾沒碰 curr.fetchedAt —— 它每週都變,但那不是「競品的變化」。
  return changes;
}

兩個地方特別提醒你:

⚠️ 一定要忽略「每週都會變、但沒意義」的欄位,像 fetchedAt 這種抓取時間戳。你要是傻傻地「整個 snapshot 比一比、只要有欄位不同就報」,那每週都會冒出一則假變化「抓取時間變了」——使用者對這種雜訊的容忍度是零。diff 的價值一半在「報出真變化」,另一半在**「對沒意義的變動閉嘴」**。

⚠️ 這版 computeDiff 只報「新增」的功能,不報「下架」的功能(看那個單向迴圈就知道)。這是個刻意的取捨,不是 bug——對競品情報來說,「對手上了什麼新功能」通常比「拿掉了什麼」更值得警覺。真要報下架,反方向再跑一次迴圈即可。把這條邊界講在明處,免得你哪天發現「對手悄悄下架某功能沒被報出來」以為程式壞了。

它是純函式,所以離線就能測到死

因為 computeDiff 不碰 LLM、不碰網路,它可以用 Node 內建的測試框架直接測,跑一整套只要幾毫秒、一把金鑰都不用

// diff.test.ts —— `node --test`(或 `npx tsx --test diff.test.ts`),零金鑰
import { test } from "node:test";
import assert from "node:assert/strict";
import { computeDiff, type Snapshot } from "./diff";

const base: Snapshot = {
  price: 12,
  features: ["看板", "甘特圖"],
  headline: "團隊協作",
  fetchedAt: "2026-06-29T03:00:00Z",
};

test("漲價達門檻 → 紅色警示,摘要含百分比", () => {
  const curr: Snapshot = { ...base, price: 15, fetchedAt: "2026-07-06T03:00:00Z" };
  const changes = computeDiff(base, curr);
  assert.equal(changes.length, 1);
  assert.equal(changes[0].field, "price");
  assert.equal(changes[0].severity, "red");
  assert.match(changes[0].summary, /\+25%/);
});

test("第一次跑(上週為 null)→ 沒有 diff,只建基準", () => {
  assert.deepEqual(computeDiff(null, base), []);
});

test("只有 fetchedAt 不同 → 視為沒變化(忽略雜訊欄位)", () => {
  const curr: Snapshot = { ...base, fetchedAt: "2026-07-06T03:00:00Z" };
  assert.deepEqual(computeDiff(base, curr), []);
});

test("上線新功能 → 紅色警示;下架的舊功能不報", () => {
  const curr: Snapshot = { ...base, features: ["甘特圖", "AI 週報"] }; // 加了「AI 週報」、少了「看板」
  const changes = computeDiff(base, curr);
  assert.equal(changes.length, 1); // 只報新增的那一則
  assert.equal(changes[0].field, "feature");
  assert.match(changes[0].summary, /AI 週報/);
});

這就是把 LLM 節點與純邏輯在 seam 切開的回報:你產品最值錢的那段邏輯(diff),是全套系統裡最好測、最穩、最不用花錢跑的一塊。 抓取那半邊會連網、會抽風、會燒 token,但 diff 這半邊永遠可以離線用 stub 資料測到死。

diff ≠ 去重(dedup):同一批 vs 跨週

第 4 課我們做過去重(dedup),這裡務必把兩者分清楚——它們長得有點像,但根本不是同一件事,混淆會讓你把架構接錯。

  • 去重(dedup,第 4 課):在同一次執行內,把重複的來源/內容合成一份。例如這週你從 Acme 的 RSS 定價頁同時抓到「漲價到 $15」——同一件事、兩個來源。dedup 把它們併成一則。它比對的是同一批資料彼此之間,完全不需要記得過去
  • diff(只報變化,本課):拿這週的狀態去比上週的狀態,只留變化。它比對的是兩次不同執行(跨週),非得記得上週不可——所以它離不開 checkpointer。

一句話記住差別:去重是「橫的」,diff 是「縱的」。 去重收拾同一週橫向的一堆重複;diff 拿收拾乾淨的這週,去跟上週縱向比一刀。它們在 graph 上是前後接力,不是二選一:

flowchart LR F["fetch:抓這週多來源"] --> DE["dedup:同一批去重(第 4 課,橫向)"] DE --> DF["diff:比上週、只留變化(本課,縱向)"] DF --> OUT["本週週報 → control panel"] CP[("checkpointer<br/>單一 thread 的上週 snapshot")] -.->|撈回上週| DF DF -.->|把這週存成下週的上週| CP
去重(dedup,第 4 課) diff(只報變化,本課)
比對對象 同一次執行內的多個來源 這週 vs 上週(兩次執行)
軸向 橫向(同一批內部) 縱向(跨時間)
要記得過去嗎 不用 必須(靠 checkpointer)
沒有它會怎樣 同一件事被重複報好幾遍 每週把「全部」倒給你,不知道哪裡變了
在 graph 裡的位置 fetch 之後、diff 之前 dedup 之後、送出之前

💡 一個檢查你有沒有分清楚的小測驗:「Acme 的 RSS 和定價頁都說漲價了,只報一次」——這是去重。「Acme 這週 $15、上週 $12,報成『漲了』」——這是 diff。前者不需要 checkpointer,後者沒有 checkpointer 就做不出來。


四、checkpointer 還是 Store?分工與各放什麼

到這裡你已經完全理解 checkpointer 了。現在補上它的搭檔 Store,並把「什麼放哪」一次分清楚——這是本課另一個非搞懂不可的分工。

  • checkpointer(跨執行狀態):持久化單一 thread 的 state。它是綁在一條追蹤線上的。「Acme 這條 thread 上週是 $12」就存在 Acme 這條 thread 的 checkpointer state 裡,BoltBoard 那條 thread 看不到、也不該看到。
  • Store(跨 thread 記憶)跨 thread、跨執行的長期記憶。它不綁定任何單一 thread,是所有 thread 都能讀到的共用知識。

用一句話分辨「這東西該放哪」:問它「屬於某一個競品,還是屬於整個帳號?」

  • 「Acme 這個競品上週賣 $12」→ 只跟 Acme 有關 → checkpointer,放在 Acme 那條 thread。
  • 這個帳號的使用者說:漲跌 10% 以內不用標紅、不在乎首頁文案改動」→ 這條偏好對 Acme、BoltBoard、Cascade 每一條 thread 都適用Store,放在帳號層。
flowchart TB STORE[("Store(跨 thread)<br/>這個帳號的偏好、門檻、追蹤名單")] subgraph ACC["帳號 acme-corp(一個付費客戶)"] subgraph T1["thread:acme-planner"] S1["checkpointer state<br/>lastSnapshot = 上週的 Acme"] end subgraph T2["thread:boltboard"] S2["checkpointer state<br/>lastSnapshot = 上週的 BoltBoard"] end subgraph T3["thread:cascade"] S3["checkpointer state<br/>lastSnapshot = 上週的 Cascade"] end end STORE -.-> T1 STORE -.-> T2 STORE -.-> T3

看這張圖就懂了:checkpointer 是每條 thread「各自」的抽屜;Store 是「架在所有 thread 之上」、大家共用的那一格。 競品情報週報各放什麼,整理成一張表:

東西 放哪 為什麼
某競品上週的 snapshot(價格/功能/主標) checkpointer(該競品的 thread) 只跟那一個競品有關,是那條追蹤線的專屬狀態
使用者偏好(標紅門檻、靜音哪類變化、週報語氣) Store(帳號層) 對這個帳號的每一條 thread 都適用
這個帳號在追的競品名單 Store(帳號層) 屬於帳號、不屬於任何單一競品
學到的正規化規則(「Acme Planner Pro」=「Pro 方案」) Store(帳號層或全域) 跨競品、跨週都想重用的知識
使用者按過「這則不重要、以後別再報」的靜音記錄 Store(帳號層) 要跨週記住,且可能跨競品套用

動手:讓 diff 節點同時吃兩邊

Store 的 API 就像一個帶命名空間的鍵值儲存——namespace 是一個字串陣列,像資料夾路徑:

// store-demo.ts
import { InMemoryStore } from "@langchain/langgraph";

async function main() {
  const store = new InMemoryStore();
  const ns = ["account", "acme-corp", "preferences"]; // 這是「acme-corp 帳號的偏好」

  await store.put(ns, "diff-rules", {
    redPriceChangePct: 10,   // 漲跌達 10% 才標紅
    muteFields: ["headline"], // 這個客戶不在乎首頁文案改動
  });

  const item = await store.get(ns, "diff-rules");
  console.log(item?.value); // → { redPriceChangePct: 10, muteFields: ["headline"] }
}

main();

要讓 graph 的節點讀得到 Store,編譯時把它跟 checkpointer 一起掛上,節點就能從 config.store 拿到。而規則要生效,我們回頭把上一節那個 computeDiff 補上第三個選用參數 rules——把原本寫死的 10、以及「靜音哪些類別」抽出來。加了預設值,所以剛剛那套兩參數的測試一個字都不用改、照樣全綠

// diff.ts(把上一節寫死的門檻抽成選用參數;函式依舊是純的)
export interface DiffRules {
  redPriceChangePct: number;          // 價格變動達此百分比才標紅
  muteFields: Array<Change["field"]>; // 這些類別的變化完全不報
}
export const DEFAULT_RULES: DiffRules = { redPriceChangePct: 10, muteFields: [] };

export function computeDiff(
  prev: Snapshot | null,
  curr: Snapshot | null,
  rules: DiffRules = DEFAULT_RULES, // ← 新增:規則從外面傳進來,函式本身不去讀 Store
): Change[] {
  if (!curr) return [];
  if (!prev) return [];
  const changes: Change[] = [];

  if (prev.price !== null && curr.price !== null && prev.price !== curr.price) {
    const pct = Math.round(((curr.price - prev.price) / prev.price) * 100);
    changes.push({
      field: "price",
      severity: Math.abs(pct) >= rules.redPriceChangePct ? "red" : "yellow", // ← 用參數,不再寫死 10
      summary: `Pro 方案從 $${prev.price}/座 變成 $${curr.price}/座(${pct > 0 ? "+" : ""}${pct}%)`,
    });
  }

  const prevFeatures = new Set(prev.features);
  for (const f of curr.features) {
    if (!prevFeatures.has(f)) {
      changes.push({ field: "feature", severity: "red", summary: `上線新功能:${f}` });
    }
  }

  if (prev.headline !== null && curr.headline !== null && prev.headline !== curr.headline) {
    changes.push({
      field: "headline",
      severity: "yellow",
      summary: `首頁主標從「${prev.headline}」改成「${curr.headline}」`,
    });
  }

  return changes.filter((c) => !rules.muteFields.includes(c.field)); // ← 靜音的類別濾掉
}

它仍然是純函式:規則是用參數傳進去的,不是在函式裡自己去讀 Store——所以它照樣離線、無金鑰可測,「規則從哪來」是節點的事、不是這顆純邏輯的事。

// graph.ts(把 diff 節點升級成「同時吃兩邊」;沿用上面的 StateAnnotation/fetchNode/computeDiff)
import { InMemoryStore } from "@langchain/langgraph";
import { DEFAULT_RULES, type DiffRules } from "./diff"; // computeDiff 已在檔案上方 import 過

async function diffNodeWithPrefs(
  state: State,
  config: LangGraphRunnableConfig,
): Promise<Partial<State>> {
  // 帳號層偏好:從 Store 讀(這個帳號的每一條競品 thread 共用)
  const account = config.configurable?.account as string;
  const pref = await config.store?.get(["account", account, "preferences"], "diff-rules");
  const rules = (pref?.value as DiffRules) ?? DEFAULT_RULES;

  // thread 層狀態:lastSnapshot 從 checkpointer 來(這個競品專屬)
  const changes = computeDiff(state.lastSnapshot, state.current, rules);
  return { changes, lastSnapshot: state.current };
}

// 編譯時同時掛上 checkpointer 與 store
export function buildGraphWithStore(checkpointer: BaseCheckpointSaver, store: InMemoryStore) {
  return new StateGraph(StateAnnotation)
    .addNode("fetch", fetchNode)
    .addNode("diff", diffNodeWithPrefs)
    .addEdge(START, "fetch")
    .addEdge("fetch", "diff")
    .addEdge("diff", END)
    .compile({ checkpointer, store }); // ← 兩個一起掛
}

這個 diffNodeWithPrefs 把整課的分工濃縮在一個節點裡,值得你盯著看三秒:

  • state.lastSnapshot —— thread 層,從 checkpointer 來,是「這個競品上週的樣子」。
  • config.store?.get(...) —— 帳號層,從 Store 來,是「這個帳號對所有競品的偏好」。
  • 兩邊在此匯流,餵給那顆依舊純、依舊可離線測computeDiff(規則只是它的第三個參數)。

⚠️ 別把兩者放反了,這是會出事的錯。 如果你把「Acme 上週 $12」塞進 Store 的全域格子,那 BoltBoard 那條 thread 也會讀到 Acme 的價格,遲早串味、算錯 diff。反過來,如果你把「標紅門檻」只存進 Acme 這一條 thread 的 checkpointer state,那使用者改了偏好,只有 Acme 生效,BoltBoard 還是舊規則——使用者會覺得你的產品壞了。判準永遠是那句話:屬於某一個競品,還是屬於整個帳號?

💡 至於「這個帳號的資料怎麼跟別的付費客戶徹底隔離、thread_id 和 Store 的 namespace 怎麼按租戶切」——那是**多租戶(multi-tenant)**的題目,第 10 課專講。本課的 "account::competitor" 命名只是先讓你看到分層的形狀,還不是完整的租戶隔離方案。


五、為什麼「跨執行狀態 + 只報 diff」是護城河

最後拉高視角,回答第 1 課埋下的問題:這一課做的事,為什麼是這門生意的護城河,是一次性 prompt/toy 抄不走的核心價值?

回到開頭那個比喻。現在整個市場都能雇到「那個很聰明、每週重來的顧問」——呼叫一次 LLM 這件事已經是大宗商品,你會、你的競爭對手會、你客戶的實習生也會。聰明本身,不值錢了。

值錢的是**「記得」**。而「記得這個客戶的這 N 個競品、一週一週累積下來的歷史」,不是一次呼叫能生出來的東西,是你的產品替這個客戶跑了一週又一週、才長出來的資產。這一課造的 checkpointer+thread+diff,就是這份資產的生產線。

這道護城河具體卡住對手三件事:

  1. 一段 prompt 抄不走它。 有人想山寨你,把你的週報餵給 GPT 說「照這個格式做一個」——他最多做出「那個每週重來的顧問」。他說不出「上週還是 $12」,因為他沒有上週。他要能說出這句話,就得自己建持久化、建 thread、建 diff——也就是把你這一課做的事重做一遍。護城河不是一句藏起來的咒語,是一套會隨時間累積的基礎設施+資料,這是 prompt 複製不了的。
  2. 它會複利。 第 1 週,你的 diff 很單薄(只是建基準)。跑到第 20 週,你手上有這個客戶 20 週的競品歷史——你能報的不只是「這週漲價」,還能報「Acme 這一季已經漲第二次了」這種趨勢,那是有 20 週記憶才說得出口的洞察。跑越久,週報越有價值。
  3. 它天然拉高轉換成本。 客戶要是換掉你,換走的那一刻,他丟掉的是這 20 週累積的記憶——新工具對他又是一張白紙,得從頭建基準。「離開你=失憶」,這道無形的鎖,正是訂閱制黏著度的來源。

這正好接回第 1 課那條線:誰扛累活,誰收月費。 這一課,我們把「記憶 + 比對」這兩件最累、最容易漏、最沒人想每週手動做的活,用 checkpointer 和 diff 從使用者身上接了過來、扛在後端。使用者付月費,買的就是「這兩件累活,我從此不用碰」。

⚠️ 誠實把邊界講清楚:這不是一道無限寬的護城河。要是有個資金充足的對手也肯認真做這一整套跨執行狀態,你的領先就會收窄成兩件事——「誰累積得更久」(你早開始跑,就早有歷史)和**「誰記得的東西更準」(抓取抽得準不準是第 3 課的活、diff 算得漂不漂亮是本課的活)。所以真正持久的優勢是:早點開始替客戶累積、並且死磕「記得的品質」。 記憶是門檻,但記得得好**,才是你持續贏的地方。


這一課建立了什麼

把這一課收攏成你可以帶走的心智模型:

  • thread(一條追蹤線)=跨越多次執行的一條線,在我們的產品裡一個競品=一條 thread,由穩定唯一的 thread_id 標定。
  • checkpointer(跨執行狀態)=持久化單一 thread 的 state 的一層,掛在 graph 上、跟著生命週期自動撈回與存回。有了它,「上週跑完的 state 這週還在」,一次性 LLM 呼叫做不到的事就成立了。示範用 MemorySaver(看語意)/SqliteSaver(撐重啟);上線接 Postgres+多租戶是第 10 課,換後端只換一行。
  • Store(跨 thread 記憶)跨 thread、跨執行的長期記憶,架在所有 thread 之上共用。分工判準一句話:屬於某一個競品 → checkpointer;屬於整個帳號 → Store。
  • diff(只報變化) 是一段純邏輯computeDiff),坐在 seam 的純計算這一邊,離線、無金鑰、毫秒級就能用 stub 資料測到死。它跟第 4 課的**去重(dedup)**是不同軸向:去重是橫的(同一次執行內合併重複、不需記憶)、diff 是縱的(跨兩次執行比對、非有 checkpointer 不可)
  • 這一課就是護城河:跨執行狀態+只報 diff,把「記憶+比對」這兩件累活從使用者身上接到後端;它抄不走(要抄就得重建這套基礎設施+資料)、會複利、會拉高轉換成本——但別忘了它不是無限寬,早累積+記得得準才是持久優勢。

你現在手上這台機器,已經能記得上週、只報 diff 了。但它有個沒回答的問題:每記一次、每比一次,到底花了多少錢? 週報準時送達的背後,是一次又一次真金白銀的 LLM 呼叫。下一課(第 8 課)我們就給每個節點裝上預算閘(budget gate)、把便宜的活丟給便宜模型(模型分流)、重複的別重付(快取)——因為第 1 課說過:能不能收費,一半取決於能不能控制成本。