第 4 課 · 多競品一起跑——平行 fan-out、子圖複用與去重

第 4 課 · 多競品一起跑——平行 fan-out、子圖複用與去重

第 3 課結束時，你的 graph 已經會盯「一個」競品：一個 fetch 節點去打 RSS、changelog、pricing，把原料寫回 state，交給 summarize 濃縮成一段。單一競品，跑得漂亮。

然後你的第一個付費客戶來了。他要追的競品不是一個——是八個。

你最直覺的改法，大概是拿個 for 迴圈把那條線包起來，一個競品跑完換下一個。它會動。但它會排隊：八個競品，一個等一個，跑完可能得等上一分多鐘——而這還是在沒人盯著的凌晨三點。更糟的是，客戶每多加一個競品，這條隊就更長一截。

問題出在哪？出在這八個競品根本互不相欠。Acme 這週漲不漲價，跟 BoltBoard 出不出新功能，一點關係都沒有。既然互不相欠，讓它們排隊就是純粹的浪費——它們大可同時發車。

這一課，就是把「排隊」改成「同時跑」，而且要跑得乾淨。具體是三件事：

  1. 平行 fan-out：一次把 N 個競品全部發車，而不是一個一個來，最後把各路結果匯合回 state。
  2. subgraph（子圖）：把「單一競品的抓取→摘要」這條線打包成一個可複用的零件，每個競品都套同一套。
  3. 去重（dedup）：多來源、多競品，必然會抓到重複的東西（同一則漲價新聞被三個地方各報一次）。這段純邏輯要在花錢摘要之前把重複清掉——而它剛好是這一課裡，你可以完全離線、不用金鑰就測到爽的部分。

💡 先接上 ch5。你在 AI 應該懂的事 · 第 5 章 已經學過 Parallelization（平行） 這個設計模式：子任務互不依賴，就同時做、最後合併。這一課不重講那個「為什麼」，只講「在 LangGraph.js 裡怎麼把它做出來」——概念之後的實作。

先把這一課會用到、但第 2、3 課早就建立好的型別擺出來對照（這不是重新定義，只是讓後面的程式碼有上下文）：

// types.ts — 第 2、3 課已建立，這裡列出只為對照。
export interface Competitor {
  id: string;            // "acme"
  name: string;          // "Acme Planner"
  // 第 3 課還有各來源的設定（rss、pricing url…），這一課用不到就略。
}

export interface RawItem {   // 第 3 課的 fetch 節點寫進 state 的原料
  competitor: string;        // competitor id
  source: string;            // "pricing" | "changelog" | "rss" | "news"
  url: string;
  title: string;
  text: string;
  fetchedAt: string;         // ISO 時間字串
}

export interface DigestEntry { // 第 2 課的 summarize 節點產出的一則
  competitor: string;          // competitor id
  summary: string;
  citations: string[];         // 這則摘要引用到的來源 url
}

一、先感受排隊有多痛

在動手改之前，先把「排隊版」老實寫出來，痛一次，你才會珍惜後面的解法。它就是把第 2、3 課那條單一競品的線，包進一個 for 迴圈：

// 排隊版：一次處理一個競品，做完才做下一個。
async function runSequential(competitors: Competitor[]): Promise<DigestEntry[]> {
  const entries: DigestEntry[] = [];
  for (const competitor of competitors) {
    const items = await fetchSources(competitor);       // 抓多來源（第 3 課）
    const kept = dedup(items).kept;                      // 去重（這一課）
    const entry = await summarize(competitor, kept);     // 摘要（打 LLM，第 2 課）
    entries.push(entry);
  }
  return entries;
}

算一筆帳你就懂痛在哪。假設抓一個競品的多來源要 4 秒、摘要打一次 LLM 要 6 秒，那麼一個競品是 10 秒。排隊版是「10 秒 × 競品數」：

競品數 排隊版（一個接一個） 平行版（同時發車）
3 ~30 秒 ~10 秒
8 ~80 秒 ~10 秒
20 ~200 秒（3 分半） ~10 秒

排隊版的時間跟著競品數線性長大；平行版幾乎是一個常數——因為大家同時跑，總時間逼近「跑得最慢的那一個」，而不是「全部加起來」。這就是這一課要拿走的東西。

⚠️ 別把「平行」當萬靈丹——它只對「互不相欠」的子任務有效。 如果競品 B 的分析需要競品 A 的結果（B 要等 A），那它們之間就有依賴，只能排隊，那是 ch5 講的 Chaining（串接），不是平行。我們的競品之間剛好互不相欠，才有平行的本錢。判斷用哪個，永遠先問一句：這幾件事之間，有沒有先後依賴？

⚠️ 平行也不是免費的午餐。 八個競品同時發車，意味著在同一瞬間你可能同時打八次 LLM——花費在那一刻集中爆發，也更容易撞上供應商的 rate limit。這一課我們先讓它「跑對、跑快」，怎麼讓它跑得省（預算閘、模型分流、同時最多打幾個）是第 8、9 課的事，那裡會回來收斂這股併發火力。

二、fan-out：用 Send 讓 N 個競品同時發車

在 LangGraph.js 裡，讓事情平行有兩種寫法，差別在**「要分成幾岔」是建圖時就固定，還是跑起來才知道**。

寫法 A：靜態多邊（分岔數建圖時就固定）。 你已經會用 addEdge 連邊了（第 2 課）。從同一個節點拉出多條邊，這幾條邊指到的節點就會平行跑：

// 從 A 同時分岔到 B、C、D，三者平行，之後都匯到 E。
graph
  .addEdge("A", "B")
  .addEdge("A", "C")
  .addEdge("A", "D");

這適合分岔數寫程式的當下就定死的情況——例如「永遠同時抓 RSS、changelog、pricing 這三個固定來源」。但注意：那種「固定三來源」的平行，其實是第 3 課那個 fetch 節點內部的事，跟我們現在的問題不同。我們的競品數量是使用者設定的、跑起來才知道：這個客戶追 3 個、那個客戶追 20 個。靜態邊在建圖時就得寫死幾條，套不上「數量會變」。

寫法 B：動態 fan-out（分岔數看資料決定）。 這就是我們要的。LangGraph.js 提供一個原語 Send，讓你在一條**條件邊（conditional edge）**裡，回傳「一串」要發車的目標：

import { Send } from "@langchain/langgraph";

// 對每個競品，發一台開往 "processCompetitor" 的車，車上載著這個競品。
function fanOutToCompetitors(state: typeof DigestState.State) {
  return state.competitors.map(
    (competitor) => new Send("processCompetitor", { competitor })
  );
}

new Send(節點名, 載荷) 的意思是：「幫我啟動一個 processCompetitor 任務，它的輸入就是這個載荷 { competitor }。」一個 .map() 下去，有幾個競品就回傳幾個 Send，LangGraph.js 就會同時啟動這幾個任務。分岔數不必寫死——看 state.competitors 有多長，就分幾岔。

你把這個函式掛在一條從 START 出發的條件邊上：

import { StateGraph, START, END } from "@langchain/langgraph";

// 母圖（parent graph）的骨架，先看 fan-out 這一段。
const digestGraph = new StateGraph(DigestState)
  .addNode("processCompetitor", processCompetitor)          // 稍後定義
  .addConditionalEdges(START, fanOutToCompetitors, ["processCompetitor"])
  .addEdge("processCompetitor", END);

addConditionalEdges(START, fanOutToCompetitors, …) 就是「一開跑，就照 fanOutToCompetitors 回傳的那串 Send 去分岔」。第三個參數 ["processCompetitor"] 是列出「這條邊可能通往哪些節點」，給建圖驗證和畫圖用的。

⚠️ 「數量會變」不代表你升級成 agent 了。 ch5 分過一組很容易混的：拆法你（用資料/設定）固定 = Parallelization；拆法 LLM 臨場決定 = Orchestrator–Workers。 我們這裡雖然分岔數是動態的（3 個、20 個都行），但「要拆成幾份、每份是誰」是一個普通函式照著競品清單算出來的，不是模型在那邊即興規劃。所以這還是 ch5 的 Parallelization，只是資料驅動而已——別因為數字會變，就以為它變成了 agent，也就別為它扛上 agent 那套貴、慢、難控的代價。

三、fan-in：reducer 把各路結果匯合回 state

發車發出去了，車怎麼開回同一個月台？八個 processCompetitor 各自產出一則 DigestEntry，你得把這八則收攏成一份週報。這個「匯合」在 LangGraph.js 裡靠的是 state 上的 reducer（歸併函式）。

先看母圖的 state。關鍵是那個要接住各路結果的 channel——entries：

import { Annotation } from "@langchain/langgraph";

const DigestState = Annotation.Root({
  // 輸入：這次要追的競品清單。
  competitors: Annotation<Competitor[]>,

  // fan-in 的匯合點：每個競品分支把自己那則 append 進來。
  entries: Annotation<DigestEntry[]>({
    reducer: (existing, incoming) => existing.concat(incoming),
    default: () => [],
  }),

  // assemble 之後的成品（另開一個 channel，原因見下方 ⚠️）。
  digest: Annotation<DigestEntry[]>,
});

reducer 就是一個函式，回答「當有新值要寫進這個 channel 時，新舊兩個值怎麼合成一個」。這裡的 reducer: (existing, incoming) => existing.concat(incoming) 意思是「接起來」。於是當八個分支各自回傳 { entries: [自己那則] }，LangGraph.js 就用這個 reducer 一則一則接上去，最後 entries 裡剛好是八則——各路平行寫入，被 reducer 安全地縫成一條。這就是 fan-in。

那「等大家都到齊」呢？看母圖怎麼收尾：

const digestGraph = new StateGraph(DigestState)
  .addNode("processCompetitor", processCompetitor)
  .addNode("assemble", assembleDigest)
  .addConditionalEdges(START, fanOutToCompetitors, ["processCompetitor"])
  .addEdge("processCompetitor", "assemble")   // ← 匯合點
  .addEdge("assemble", END)
  .compile();

addEdge("processCompetitor", "assemble") 這條邊有個關鍵性質：assemble 會等到所有平行的 processCompetitor 全部跑完，才跑一次。它不會八個分支各觸發一次 assemble——它是「全部到齊，才放行」。這正是 map-reduce 的 reduce 那一步：各路匯合後，做一次收尾（排序、組版）。

// assemble：所有分支到齊後，把 entries 收成最終 digest。
function assembleDigest(state: typeof DigestState.State) {
  const digest = [...state.entries].sort((a, b) =>
    a.competitor.localeCompare(b.competitor)
  );
  return { digest };
}

⚠️ 這裡藏著一個很容易踩的雷：別把匯合後的整包，再寫回帶 reducer 的 channel。 注意 assembleDigest 回傳的是 { digest }，寫到另一個 channel，不是 { entries }。為什麼？因為 entries 帶著 concat 這個 reducer——你如果把排好序的整包 entries 再寫回 entries，reducer 會忠實地把它再接一次，於是你的週報瞬間變成兩倍長。帶 reducer 的 channel 只適合「一片片丟進去累積」；要「換掉整包」的收尾，請寫到另一個沒有 reducer 的 channel。 這是 LangGraph.js 新手最常見的重複 bug 之一。

💡 「引擎怎麼知道八個分支全跑完了、才放行 assemble」——那是編排引擎的內部工作，這一課不拆。 想親手實作那套「排程平行任務、追蹤誰跑完了、到齊才合流」的機制，那是姊妹系列 ai-pipeline-from-scratch 的主題。這一課我們只負責把 reducer 和邊接對，剩下的排程與合流，LangGraph.js 現成處理掉。一句話：手刻懂原理，框架去出貨。

到這裡，fan-out（分岔）和 fan-in（匯合）的骨架就齊了。但中間那個 processCompetitor 還是個空殼——它到底怎麼「抓取→去重→摘要」一個競品？這就輪到 subgraph 上場。

四、subgraph（子圖）：把「單一競品」打包成可複用零件

「抓取一個競品 → 去重 → 摘要」本身就是一條完整的小流程：有它自己的節點、自己的邊、自己流動的 state。與其把這幾步攤平塞進母圖、再想辦法讓它對每個競品各跑一遍，不如把它打包成一個獨立的零件——這個零件，在 LangGraph.js 裡就叫 subgraph（子圖）。

subgraph（子圖）＝一張自己 compile() 過的 graph，被當成一個節點（或一個可呼叫的零件）用在更大的 graph 裡。

為什麼要特地封裝成子圖，而不是攤平？三個很實際的好處：

  • 複用：每個競品跑的都是同一套「抓取→去重→摘要」。封裝成一個零件，就是「定義一次，對每個競品各跑一份」，不必複製貼上八遍。
  • 能單獨測：子圖是一張完整的 graph，你可以單獨餵一個競品給它、單獨驗證它，不必每次都把整個母圖拖下水。
  • 邊界乾淨：一個競品的內部細節（抓了哪些來源、丟了幾則重複）鎖在子圖裡，不會污染母圖的 state。這條乾淨的邊界，之後還會派上大用場——每個競品之後要有自己的一條 thread（跨週追蹤線），那是第 7 課的事，而子圖正好是掛上它的天然位置。

先把子圖自己的 state 定出來。它跟母圖是兩份不同的 state：母圖那份管「全體競品」，子圖這份只管「手上這一個競品」，跑完就丟：

// 子圖的 state：每個競品自己一份，跑完就丟。
const CompetitorState = Annotation.Root({
  competitor: Annotation<Competitor>,        // 輸入：這一個競品
  rawItems: Annotation<RawItem[]>,           // fetch 抓回來的原料
  dedupedItems: Annotation<RawItem[]>,       // 去重後剩下的
  entry: Annotation<DigestEntry>,            // summarize 產出的那一則
});

接著把子圖組起來——fetch（第 3 課）→ dedup（這一課的主角，下一節細講）→ summarize（第 2 課）三個節點串成一條線。注意我把 fetch 和 summarize 這兩個會碰外部世界的節點做成可注入的：真正打網路的 fetcher、真正打 LLM 的 summarizer 從外面傳進來。這就是第 2 課那條 seam（LLM 邊界）——把「純邏輯」和「要金鑰的呼叫」切開，測試時換成 stub 就能離線跑。第七節會兌現這個好處：

// graph.ts
import { StateGraph, START, END, Send, Annotation } from "@langchain/langgraph";
import { dedup } from "./dedup";
import type { Competitor, RawItem, DigestEntry } from "./types";

// 兩道 seam：抓料、摘要。真實作打外部世界，測試時換 stub。
export interface Fetcher {
  (competitor: Competitor): Promise<RawItem[]>;
}
export interface Summarizer {
  (competitor: Competitor, items: RawItem[]): Promise<DigestEntry>;
}

export function buildDigestGraph(deps: { fetcher: Fetcher; summarizer: Summarizer }) {
  // ── 子圖：單一競品的「抓取 → 去重 → 摘要」──
  const competitorSubgraph = new StateGraph(CompetitorState)
    .addNode("fetch", async (s) => ({ rawItems: await deps.fetcher(s.competitor) }))
    .addNode("dedup", (s) => ({ dedupedItems: dedup(s.rawItems).kept }))
    .addNode("summarize", async (s) => ({
      entry: await deps.summarizer(s.competitor, s.dedupedItems),
    }))
    .addEdge(START, "fetch")
    .addEdge("fetch", "dedup")
    .addEdge("dedup", "summarize")
    .addEdge("summarize", END)
    .compile();

  // ── 把子圖當成一個可複用零件，包一層接回母圖 ──
  async function processCompetitor(input: { competitor: Competitor }) {
    try {
      const result = await competitorSubgraph.invoke({ competitor: input.competitor });
      return { entries: [result.entry] };
    } catch (err) {
      // 一個競品爆掉，不該拖垮整份週報 → 降級成一則「這家這週抓取失敗」。
      return {
        entries: [
          {
            competitor: input.competitor.id,
            summary: `（本週 ${input.competitor.name} 抓取失敗，已略過）`,
            citations: [],
          },
        ],
      };
    }
  }

  // ── 母圖：fan-out → 平行跑子圖 → fan-in ──
  return new StateGraph(DigestState)
    .addNode("processCompetitor", processCompetitor)
    .addNode("assemble", assembleDigest)
    .addConditionalEdges(START, fanOutToCompetitors, ["processCompetitor"])
    .addEdge("processCompetitor", "assemble")
    .addEdge("assemble", END)
    .compile();
}

看懂這段的三個關鍵：

  1. 子圖就是一張 graph。 competitorSubgraph 用你熟的 StateGraph、addNode、addEdge、compile() 蓋出來——跟母圖同一套積木，只是尺寸小。compile() 完，它就是一個可以 .invoke() 的零件。
  2. processCompetitor 把子圖 invoke 一次。 fan-out 發來的每一台車（每個競品）都會進到這裡，各自 await competitorSubgraph.invoke({ competitor }) 跑完自己那條「抓取→去重→摘要」，拿回 result.entry，回傳 { entries: [entry] } 讓母圖的 reducer 匯合。因為 fan-out 把它們同時發車，這些 invoke 是同時在跑的——不是排隊。
  3. 一個競品爆掉，別拖垮整份週報。 那個 try/catch 很重要：平行的風險是「一顆老鼠屎壞一鍋粥」——某個競品的網站掛了、抓取 throw，預設會讓整個平行步驟失敗，你八個競品的努力全泡湯。用 try/catch 把單一分支的失敗隔離，降級成一則「這家失敗、已略過」，其他七家照樣送達。這是「平行要穩」的基本功；更完整的**降級（fallback）**策略是第 9 課。

前面為了一段段講解，把 DigestState、CompetitorState、fanOutToCompetitors、assembleDigest 拆在不同程式碼塊看——它們其實都住在同一個 graph.ts，buildDigestGraph 就是把 fan-out、子圖、fan-in 全組裝好、注入兩道 seam、回傳 compile 完成的母圖。（types.ts 放共用型別、dedup.ts 放去重、seams.ts 放 stub 與真實作。）

💡 另一種接子圖的寫法，順帶讓你知道有這條路。 你也可以把 compiled 的子圖直接當節點用：.addNode("processCompetitor", competitorSubgraph)——當子圖和母圖共用同名 channel 時，state 會自動穿過邊界流通。我們這裡刻意包一層 invoke，是因為 (a) seam 更清楚（子圖是黑盒，輸入輸出一目了然）、(b) 好加上面那個 per-競品的 try/catch、(c) 之後好給每個競品掛自己的 thread（第 7 課）。差別主要在串流的顆粒度（要不要把子圖內部每一步也串進 UI，那是第 5 課的取捨）——現在知道兩條路都在、我們選了哪條、為什麼，就夠了。

五、去重（dedup）：多來源必然重複，這段純邏輯怎麼做

現在拆子圖中間那顆螺絲——dedup。它是這一課裡唯一完全不碰外部世界、不需要金鑰的節點，也因此是你能測到最紮實的一段。

為什麼一定會有重複？

因為我們多來源、又多競品。同一件事，會從好幾個入口同時被抓進來：

  • Acme 這週把 Pro 從 $12 漲到 $15。這件事同時出現在：它的 pricing 頁（你直接抓到）、它的 changelog（它自己公告了）、還有一則科技媒體的新聞（別人報導了）。同一件事，三個來源各抓一次。
  • 一則新聞稿被兩家媒體轉載，網址不同、內容幾乎一樣。同一則，兩個網址。

如果不去重，你會付三次錢去摘要同一件事，週報上還會出現三條在講同一件事的重複條目。去重，就是在花錢摘要之前，把這些「其實是同一件事」的收成一份。

⚠️ 去重（dedup）≠ diff（只報變化）——這兩件事務必分清楚

這是這一課最容易搞混、也最重要的一個界線。你在第 1 課看過週報裡那句「上週還是 $12」——那個「只報變了的事」叫 diff，是這個產品的靈魂。去重不是它。

去重（dedup）· 這一課 diff（只報變化）· 第 7 課
比的是誰跟誰 同一次執行裡，item 跟 item 這週的狀態 跟 上週的狀態
要不要記憶 不用，資料全都在手上 要，得靠 checkpointer 記得上週
丟掉的是什麼 重複的來源／內容（同一件事被抓到很多次） 沒變的事（這週跟上週一樣）
例子 漲價被 pricing、RSS、新聞各抓一次 → 只留一份 Acme 上週 $12、這週還是 $12 → 不報

一句話記住：去重是「同一次執行裡，把重複的收成一份」；diff 是「跨越兩次執行，把沒變的藏起來」。 去重完全發生在本次執行的手上資料之間，用不到任何「上週」的記憶——所以它是純邏輯、能離線測。而 diff 非得記得上週不可，那需要 checkpointer 和 thread，是第 7 課才登場的機制。這一課講的所有「比對」，都只是去重，不是 diff。

去重的純邏輯：兩把鑰匙 + 一個權威度

判斷「兩則是不是同一件事」，我們用兩把鑰匙：

  1. 正規化網址：把網址洗成一把「同一頁就會相等」的鑰匙——去掉 utm_* 這類追蹤參數、結尾斜線、www.、大小寫與 http/https 的差異。抓「同一個網址被抓了很多次」。
  2. 內容指紋：把標題 + 正文開頭正規化後拼成一個字串。抓「網址不同、但其實是同一則」（例如新聞被兩家轉載）。

遇到重複時，保留哪一份？用來源權威度決定：官方 pricing 頁 > 官方 changelog > RSS > 第三方新聞。同一件事，優先信官方那份。

// dedup.ts — 純邏輯：不打網路、不打 LLM、不需要任何金鑰。
import type { RawItem } from "./types";

export type DropReason = "same-url" | "same-content";

export interface DedupResult {
  kept: RawItem[];
  dropped: Array<{ item: RawItem; reason: DropReason; keptUrl: string }>;
}

// 來源權威度：官方頁 > 官方 changelog > RSS > 第三方新聞。
const SOURCE_AUTHORITY: Record<string, number> = {
  pricing: 4,
  changelog: 3,
  rss: 2,
  news: 1,
};
function authority(source: string): number {
  return SOURCE_AUTHORITY[source] ?? 0;
}

// 把 URL 正規化成一把「同一頁就會相等」的鑰匙。
export function canonicalUrl(input: string): string {
  let u: URL;
  try {
    u = new URL(input.trim());
  } catch {
    return input.trim().toLowerCase(); // 不是合法網址就退回原字串
  }
  u.protocol = "https:";                       // http / https 視為同一頁
  u.hostname = u.hostname.replace(/^www\./, ""); // 去掉 www.（hostname 本來就是小寫）
  u.hash = "";                                 // 去掉 #anchor
  for (const key of [...u.searchParams.keys()]) {
    if (/^(utm_|fbclid$|gclid$|mc_|ref$|source$)/i.test(key)) {
      u.searchParams.delete(key);              // 去掉追蹤參數
    }
  }
  u.searchParams.sort();                       // 參數順序不影響同一性
  return u.toString().replace(/\/$/, "");      // 去掉結尾斜線
}

// 內容指紋：標題 + 正文開頭正規化後的字串。
export function contentFingerprint(title: string, text: string): string {
  const norm = (s: string) =>
    s
      .normalize("NFKC")
      .toLowerCase()
      .replace(/[^\p{L}\p{N}]+/gu, " ") // 標點、空白全收斂成單一空格
      .trim();
  return `${norm(title)}::${norm(text).slice(0, 160)}`;
}

export function dedup(items: RawItem[]): DedupResult {
  // 先讓權威度高的排前面 → 它會先被收下，之後的重複才會被丟掉。
  // 權威度相同時，用 fetchedAt 較早的優先（穩定、可重現）。
  const ordered = [...items].sort((a, b) => {
    const byAuthority = authority(b.source) - authority(a.source);
    if (byAuthority !== 0) return byAuthority;
    return a.fetchedAt.localeCompare(b.fetchedAt);
  });

  const seenUrl = new Map<string, RawItem>();
  const seenContent = new Map<string, RawItem>();
  const kept: RawItem[] = [];
  const dropped: DedupResult["dropped"] = [];

  for (const item of ordered) {
    const urlKey = canonicalUrl(item.url);
    if (seenUrl.has(urlKey)) {
      dropped.push({ item, reason: "same-url", keptUrl: seenUrl.get(urlKey)!.url });
      continue;
    }
    const contentKey = contentFingerprint(item.title, item.text);
    if (seenContent.has(contentKey)) {
      dropped.push({ item, reason: "same-content", keptUrl: seenContent.get(contentKey)!.url });
      continue;
    }
    seenUrl.set(urlKey, item);
    seenContent.set(contentKey, item);
    kept.push(item);
  }
  return { kept, dropped };
}

幾個刻意的設計：

  • 回傳 kept 和 dropped 兩份，不只是留下來的。dropped 讓你之後在儀表板上顯示「本週替你濾掉 4 則重複」，出問題時也看得到「到底丟了什麼、為什麼丟」。（提醒：dropped 是「被去重的重複」，不是第 1 課那個「沒變化所以略過」——那是 diff 的事，別搞混。）
  • 先排序再掃描，讓權威度高的先被收下。這樣「同一件事」留下來的永遠是最可信的那個來源，而且結果穩定可重現（同樣輸入，跑幾次都一樣）——這對「能寫測試」至關重要。

⚠️ 「去重是純邏輯」是有邊界的：它只抓得到「結構／字面重複」，抓不到「語意重複」。 同一個網址、同一段正規化後的文字，這種長得幾乎一樣的重複，純邏輯很擅長。但「Acme 漲價了」跟「Acme 的新定價更貴了」這種寫法不同、講的是同一件事的語意重複，純字串比對抓不到——那要靠 embedding 或再打一次 LLM 來判斷，會花錢，是另一種取捨。好消息是：週報這個情境，同一次抓取裡的重複，絕大多數是同來源／同轉載那種字面重複，純邏輯就吃掉九成。語意去重那條更貴的路，通常不必走——別為了追求完美去重，把一個免費、可離線測的純函式，換成一個要花錢又難測的 LLM 呼叫。

💡 為什麼去重放在 summarize 之前，而不是之後？ 因為 summarize 是要花錢的那一步（打 LLM）。先去重，你就只付「不重複的那幾則」的摘要錢；先摘要再去重，等於先付了三份錢才發現有兩份是重複的。去重擋在花錢的閘門前面——這個「便宜的純邏輯先擋掉、貴的呼叫少做幾次」的直覺，正是第 8、9 課控制預算的核心思路，這裡先種下。

💡 那「跨競品」的重複呢？ 一篇產業新聞同時提到 Acme 和 Bolt——這在「單一競品內去重」是抓不到的（它被分在兩個競品的子圖裡各看各的）。但這裡有個產品判斷：跨競品時你往往反而想各留一份，把這則新聞分別歸到 Acme 名下和 Bolt 名下，因為使用者是「一家一家競品在看」的。所以我們刻意只在單一競品內去重——這不是技術做不到全域去重，是「全域去重」在這個產品裡不一定是你要的。

六、離線把整條跑起來：stub 假模型、無金鑰

現在把整條 graph 端到端跑一遍——而且完全離線、一把金鑰都不用。這正是第 2 課那條 seam 的回報：fetch 和 summarize 這兩個碰外部世界的節點，我們是注入進去的，測試時換成 stub（假模型）就行。

先把兩道 seam 的實作擺出來——真打外部世界的版本當「示範」，離線測試的 stub 當「主力」：

// seams.ts
import type { Competitor, RawItem, DigestEntry } from "./types";
import type { Summarizer } from "./graph";

// 真打 LLM 的實作（示範；需要金鑰）。完整版是第 2 課的事，這裡只點出它是 seam 的另一側。
import OpenAI from "openai";
const client = new OpenAI();
export const realSummarizer: Summarizer = async (competitor, items) => {
  const bullets = items.map((i) => `- [${i.source}] ${i.title}：${i.text}`).join("\n");
  const res = await client.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [
      { role: "system", content: "你是競品情報分析師，把以下素材濃縮成 2-3 句本週重點。" },
      { role: "user", content: `競品：${competitor.name}\n素材：\n${bullets}` },
    ],
  });
  return {
    competitor: competitor.id,
    summary: res.choices[0]?.message.content ?? "",
    citations: items.map((i) => i.url),
  };
};

// 離線測試用的 stub：不打 LLM，回一個可預測的假摘要。
export const stubSummarizer: Summarizer = async (competitor, items) => ({
  competitor: competitor.id,
  summary: `（stub）${competitor.name}：本週 ${items.length} 則`,
  citations: items.map((i) => i.url),
});

先單獨測去重（純函式，最好測）

去重是純函式，測起來最爽——給輸入、比輸出，不用 mock 任何東西：

// dedup.test.ts
import { describe, it, expect } from "vitest";
import { dedup, canonicalUrl, type RawItem } from "./dedup";

// 小工具：造一筆假 item，只填我們關心的欄位。
function item(patch: Partial<RawItem>): RawItem {
  return {
    competitor: "acme",
    source: "rss",
    url: "https://acme.com/blog/post",
    title: "標題",
    text: "內文內文內文",
    fetchedAt: "2026-07-10T00:00:00Z",
    ...patch,
  };
}

describe("canonicalUrl", () => {
  it("去掉 utm 參數、結尾斜線、www、協定差異後，同一頁會相等", () => {
    const a = canonicalUrl("http://www.acme.com/Pricing/?utm_source=rss&utm_campaign=x");
    const b = canonicalUrl("https://acme.com/Pricing");
    expect(a).toBe(b);
  });

  it("參數順序不影響結果", () => {
    expect(canonicalUrl("https://a.com/x?b=2&a=1")).toBe(canonicalUrl("https://a.com/x?a=1&b=2"));
  });
});

describe("dedup", () => {
  it("同一個 URL（含被 utm 汙染的）只留一份", () => {
    const { kept, dropped } = dedup([
      item({ source: "rss", url: "https://acme.com/p?utm_source=rss" }),
      item({ source: "changelog", url: "https://acme.com/p" }),
    ]);
    expect(kept).toHaveLength(1);
    expect(dropped[0].reason).toBe("same-url");
  });

  it("同一頁重複時，保留權威度較高的來源（changelog > rss）", () => {
    const { kept } = dedup([
      item({ source: "rss", url: "https://acme.com/p" }),
      item({ source: "changelog", url: "https://acme.com/p" }),
    ]);
    expect(kept[0].source).toBe("changelog");
  });

  it("網址不同但內容相同，也算重複（新聞被兩家轉載）", () => {
    const { kept, dropped } = dedup([
      item({ source: "news", url: "https://site-a.com/story", title: "Acme 漲價 25%", text: "Acme 宣布 Pro 方案漲到 $15" }),
      item({ source: "news", url: "https://site-b.com/repost", title: "Acme 漲價 25%！", text: "Acme 宣布 Pro 方案漲到 $15。" }),
    ]);
    expect(kept).toHaveLength(1);
    expect(dropped[0].reason).toBe("same-content");
  });

  it("真的不同的兩則，兩則都留著", () => {
    const { kept } = dedup([
      item({ url: "https://acme.com/pricing", title: "漲價", text: "Pro 漲到 $15" }),
      item({ url: "https://acme.com/changelog", title: "新功能", text: "上線 AI 週報" }),
    ]);
    expect(kept).toHaveLength(2);
  });

  it("是純函式：同樣輸入，跑幾次結果都一樣", () => {
    const input = [
      item({ source: "news", url: "https://b.com/x", fetchedAt: "2026-07-10T02:00:00Z" }),
      item({ source: "changelog", url: "https://acme.com/p" }),
      item({ source: "rss", url: "https://acme.com/p?utm_source=rss" }),
    ];
    expect(dedup(input)).toEqual(dedup(input));
  });
});

這幾個測試一把金鑰都不用、一個網路都不打——因為去重是純邏輯。這就是把 seam 切乾淨的紅利：產品最核心的那段判斷邏輯，你可以測到滴水不漏。

再測整條 graph（含平行的證明）

最後，把整條 graph 用 stub 餵起來，證明兩件事：（1）它真的在平行跑，不是排隊；（2）去重真的在每個競品身上生效：

// graph.test.ts
import { describe, it, expect } from "vitest";
import { buildDigestGraph, type Fetcher } from "./graph";
import { stubSummarizer } from "./seams";

const competitors = [
  { id: "acme", name: "Acme Planner" },
  { id: "bolt", name: "BoltBoard" },
  { id: "cascade", name: "Cascade" },
];

describe("digest graph（離線、無金鑰）", () => {
  it("三個競品是同時在跑，不是排隊", async () => {
    let active = 0;
    let peak = 0;
    const slowFetcher: Fetcher = async (c) => {
      active++;
      peak = Math.max(peak, active); // 記錄「同時最多幾個分支在抓」
      await new Promise((r) => setTimeout(r, 50));
      active--;
      return [
        { competitor: c.id, source: "pricing", url: `https://${c.id}.com/p`, title: "x", text: "x", fetchedAt: "2026-07-10T00:00:00Z" },
      ];
    };
    const graph = buildDigestGraph({ fetcher: slowFetcher, summarizer: stubSummarizer });
    await graph.invoke({ competitors });

    // 排隊的話 peak 永遠是 1；平行時會同時有多個分支在抓。
    expect(peak).toBeGreaterThan(1);
  });

  it("多來源抓到重複，去重後每家只留不重複的那幾則", async () => {
    const dupFetcher: Fetcher = async (c) => [
      { competitor: c.id, source: "pricing",   url: `https://${c.id}.com/p`,                title: "漲價",  text: "Pro 漲到 $15", fetchedAt: "2026-07-10T00:00:00Z" },
      { competitor: c.id, source: "rss",       url: `https://${c.id}.com/p?utm_source=rss`, title: "漲價",  text: "Pro 漲到 $15", fetchedAt: "2026-07-10T01:00:00Z" }, // 跟上一筆同一頁
      { competitor: c.id, source: "changelog", url: `https://${c.id}.com/changelog`,        title: "新功能", text: "上線 AI 週報", fetchedAt: "2026-07-10T02:00:00Z" },
    ];
    const graph = buildDigestGraph({ fetcher: dupFetcher, summarizer: stubSummarizer });
    const out = await graph.invoke({ competitors });

    expect(out.digest).toHaveLength(3);          // 3 家各一則 entry
    for (const entry of out.digest) {
      expect(entry.citations).toHaveLength(2);   // 抓 3 筆、去重後剩 2 筆
    }
  });
});

第一個測試用了一個小把戲來證明平行：讓 stub 的 fetcher 在進出時記一筆「現在同時有幾個分支在抓」，取最大值 peak。如果是排隊，同一時間永遠只有一個分支在跑，peak 會等於 1；只要 peak > 1，就代表確實有多個分支同時在飛——平行成立。第二個測試餵進「每家 3 筆、其中 2 筆是同一頁」的假料，驗證每家去重後只剩 2 筆。兩個測試都不需要金鑰、不打網路，你在 CI 裡、在飛機上都能跑。

🔧 動手試試：把上面幾個檔案（types.ts / dedup.ts / seams.ts / graph.ts 以及兩個 *.test.ts）放進一個 src/，npm i @langchain/langgraph @langchain/core vitest，跑 npx vitest run。全綠。你剛剛在完全沒有 API 金鑰的情況下，跑通了一張會平行 fan-out、複用子圖、還會去重的 graph。這就是把 LLM 節點切在 seam 上、用 stub 頂替的威力——產品最核心的邏輯，離線就能守得住。（真打 LLM 的 realSummarizer 想試，補一個 OPENAI_API_KEY、把注入的 summarizer 換成它即可。）

本章建立了什麼

把這一課收攏成一張你能帶走的心智圖：

flowchart TB
    S(["START · 這次要追的競品清單"]) --> FO{{"fan-out（Send）"}}
    FO -->|"每個競品發一台車，同時跑"| P
    subgraph P["子圖 processCompetitor（每個競品各跑一份，平行）"]
      direction LR
      F["fetch：抓多來源"] --> D["dedup：去重"] --> M["summarize：摘要"]
    end
    P --> J["assemble：fan-in 用 reducer 匯合各路 entry"]
    J --> E(["END · 一份週報"])
  • 平行 fan-out：用 Send 在一條從 START 出發的條件邊（conditional edge）裡，對每個競品回傳一個 new Send("processCompetitor", { competitor })。分岔數看 state.competitors 決定，不必寫死——N 個競品同時發車，總時間逼近「最慢那一個」而不是「全部加起來」。
  • fan-in：靠 state channel 上的 reducer（例如 entries 的 concat）把各路平行寫入縫成一條；下游的 assemble 節點會等所有分支到齊才跑一次。⚠️ 別把匯合後的整包再寫回帶 reducer 的 channel（會被重複累加）——收尾寫到另一個沒 reducer 的 channel（digest）。
  • subgraph（子圖）：一張自己 compile() 過的 graph，被當成可複用零件。我們把「單一競品的抓取→去重→摘要」封裝成 competitorSubgraph，用一層 processCompetitor 把它 invoke 起來、接回母圖，並用 try/catch 把單一競品的失敗隔離，不拖垮整份週報。
  • 去重（dedup）：純邏輯，不打網路、不打 LLM、不需金鑰。用「正規化網址 + 內容指紋」兩把鑰匙抓重複，用「來源權威度」決定留誰，回傳 kept / dropped。放在 summarize 之前，省下重複摘要的錢。⚠️ 它只抓字面／結構重複，語意重複要更貴的手段、通常不必。
  • 去重 ≠ diff：去重是同一次執行內、item 跟 item 比、把重複收成一份；diff 是跨週、這週跟上週比、只留變了的（需要 checkpointer 記得上週，第 7 課）。這一課所有「比對」都只是去重。
  • 可離線測：fetch / summarize 兩道 seam 用注入 + stub 頂替，讓整條 graph——含平行與去重——都能無金鑰離線跑。去重用純函式單元測試；平行用「同時在跑的分支數 peak > 1」來證明；整條 graph 用 stub fetcher 驗證去重生效。

這一課新定義、後面可以依賴的：平行 fan-out（Send + 條件邊）、fan-in（reducer 匯合 + assemble join）、subgraph（子圖）的定義與 invoke 複用法、去重（dedup）的純邏輯與其 RawItem → { kept, dropped } 介面、以及「去重 ≠ diff」這條界線。後面各課會直接引用這些，不再重講。

下一課，我們處理另一半：這張 graph 跑起來會經過一個個節點、一段段產出——怎麼把這些中間過程一段段串進真實的 React 儀表板（不是 chat 泡泡），讓使用者看著它一個競品一個競品地完成？那就是 streaming（串流），第 5 課見。