第 5 課 · 串流進控制面：把 graph 中間結果餵進真實 React 儀表板

第 5 課 · 串流進控制面：把 graph 中間結果餵進真實 React 儀表板

第 2～4 課，你已經蓋出一張像樣的 graph：它平行盯 N 個競品、每個競品各自抓料（RSS／changelog／pricing）、去重（dedup），再摘要。現在把它跑起來，盯著你的 terminal——

然後是漫長的 40 秒。畫面什麼都沒有。沒有一個字。40 秒後，「唰」一下，一大坨 JSON 砸在螢幕上。

對你來說這完全沒問題，你知道它在忙。但把這個東西擺到一個付你月費的陌生人面前，情況就變了：他打開網頁、點下「開始分析」，然後盯著一個一動不動的轉圈圈看 40 秒。他不知道它是在認真工作，還是已經當掉了。第 41 秒他關掉了分頁，順手把退訂信也寫好了。

這一課要解決的，就是這 40 秒的黑箱。解法不是「讓它變快」——一次真材實料的情報蒐集本來就要花時間（一堆抓取、一堆 LLM 呼叫，快不了）。解法是：把過程一段段亮出來，讓使用者看到它正在幹活。 這就是 streaming（串流）。

💡 一個你早就用過的比喻：沒有人想跟宅配司機一直傳訊息問「到哪了？」。你要的是一個物流追蹤頁——它自己一格一格亮起來：「已取件 → 運送中 → 派送中 → 已送達」。你不必開口，狀態自己更新。

我們的 control panel（控制面）就是競品情報的物流追蹤頁：三個競品像三個包裹，每張卡片自己從「排隊中 → 抓取中 → 摘要中 → 完成」亮過去。這正是第 1 課那句「答案自己送上門」在畫面上的長相。

這一課，我們回答三個問題：

  1. LangGraph.js 有哪些 streaming 模式？怎麼把 graph 執行中的中間結果一段段串流出來？
  2. Node 後端怎麼把這個串流吐給前端、React 這邊怎麼接住並即時更新情報儀表板？
  3. 為什麼這個前端是 control panel 而不是 chat 泡泡——它呈現的是「多競品情報的狀態與進度」，不是一問一答的對話框？

這一課只做「把即時中間結果串上畫面」。真人審核卡（interrupt() / resume / HITL）是第 6 課、把 UI 接上跨週 diff 是第 7 課——這裡都先不碰。至於「大腦為什麼在後端、前端為什麼是 control panel」，第 1 課已經釘死，這裡只引用、不重講。

一、先看成品：我們要讓畫面怎麼動

在寫任何一行 streaming 程式碼之前，先把「目標畫面」釘在牆上。我們要的 control panel 長這樣——假設使用者追三個競品，按下「開始分析」後，這塊區域會像這樣自己動起來（下面是不同時間點的三張快照）：

▶ 剛按下（t=0s）
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Acme    ⚪排隊 │ │ BoltBoard ⚪排隊│ │ Cascade ⚪排隊 │
└──────────────┘ └──────────────┘ └──────────────┘

▶ 抓取中（t=2s，三張卡各自亮起、互不等待）
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Acme  🟡抓取中 │ │ BoltBoard🟢摘要中│ │ Cascade🟡抓取中│
│ 找到 4 則來源  │ │ 找到 3 則來源   │ │ …            │
└──────────────┘ │ 「BoltBoard 上… │ └──────────────┘
                 └──────────────┘   ← 摘要文字正在逐字長出來

▶ 完成（t=6s）
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Acme    ✅完成 │ │ BoltBoard ✅完成│ │ Cascade ✅完成 │
│ 找到 4 則來源  │ │ 找到 3 則來源   │ │ 找到 5 則來源  │
│ 「Acme 把 Pro… │ │ 「BoltBoard 上… │ │ 「Cascade 換掉…│
└──────────────┘ └──────────────┘ └──────────────┘

盯著這三張快照看，先記住兩個之後要一直回來的觀察：

  • 三張卡互不等待、各自更新。Acme 還在抓料時，BoltBoard 可能已經在摘要了——因為第 4 課的平行 fan-out 讓它們同時在跑。串流的價值就是把「同時在跑」這件事如實畫出來。
  • 卡片的數量是固定的、位置是固定的。從頭到尾就是「一個競品一張卡」，卡片就地更新內容，而不是一直往下冒出新訊息。這句話現在聽起來像廢話，但它就是本課第三個問題的答案，我們最後會回來把它釘死。

好，目標畫面有了。整條資料流長這樣——這一課就是從左到右，把這條管線一段一段接起來：

flowchart LR
    subgraph BE["後端"]
      N["graph 節點<br/>config.writer 吐事件"]
      G["graph.stream<br/>updates ＋ custom"]
      H["Node handler<br/>每個事件寫一行 NDJSON"]
      N --> G --> H
    end
    subgraph FE["前端 React"]
      R["fetch 串流讀取<br/>處理半行 buffer"]
      D["reducer<br/>reduce 進以競品為 key 的看板"]
      U["control panel<br/>每個競品一張卡就地更新"]
      R --> D --> U
    end
    H -->|"一條 HTTP 串流"| R

現在從最上游——graph 怎麼把中間結果吐出來——開始。

二、LangGraph.js 的 streaming 模式：一個 chunk 長什麼樣

你已經很熟一件事了：呼叫 OpenAI／Anthropic SDK 時可以開 stream: true，讓一段文字逐字吐回來。LangGraph.js 的 streaming 是同一個概念、升一個維度：串出來的不只是「一段文字」，而是「整張 graph 執行途中的各種中間結果」——哪個節點（node）跑完了、state 多了什麼、某個 LLM 節點正在吐字、你自己想報的任何進度。

拿到中間結果的入口只有一個函式：graph.stream(input, options)。它回傳一個非同步可迭代物件，你 for await 它，一段一段拿。關鍵全在 options.streamMode——它決定「一個 chunk 長什麼樣」。LangGraph.js 有五種模式：

streamMode 一個 chunk 長什麼樣 適合拿來做什麼
"values" 每走一步後的整包 state 想要每步的完整快照——但整包重傳，過網路很浪費
"updates" { 節點名: 這步寫回 state 的那一小塊 } 免費的粗粒度進度：哪個節點跑完、寫了什麼
"messages" [訊息 chunk, metadata]（LLM 的 token） 讓摘要文字逐字浮現（需要用 LangChain chat model）
"custom" 你自己用 config.writer(...) 吐的任意物件 精準的自訂進度：抓到幾則來源、跑到第幾步…
"debug" 詳細的除錯事件 本機除錯用，不對外

💡 這裡不需要背五個。對一個 control panel，真正的主力是兩個：updates（免費、粗粒度）＋ custom（你精準控制）。messages 是「想讓摘要逐字長出來」時的加菜，values 和 debug 這個場景基本用不到。下面我們就把主力這兩個講透、跑起來。

先各看一眼 updates 和 custom 的 chunk 實際長怎樣，你才有畫面：

// streamMode: "updates" —— 每個節點跑完，吐一包「它寫回 state 的東西」
{ "fetchAll":     { "raw": { "acme": [/* …4 則來源… */] } } }
{ "summarizeAll": { "results": { "acme": { "sourceCount": 4, "summary": "…" } } } }

// streamMode: "custom" —— 完全是你自己定義的形狀（下一節會定義它）
{ "type": "node_status",   "competitor": "acme", "status": "fetching" }
{ "type": "source_found",  "competitor": "acme", "count": 4 }
{ "type": "summary_delta", "competitor": "acme", "delta": "Acme 把 " }

看出差別了嗎？

  • updates 是 graph「自動」給你的：你什麼都不用做，它就在每個節點跑完時，把「這個節點寫回 state 的那一塊」吐給你。粗粒度、免費——很適合做一條「graph 執行日誌」。
  • custom 是你「主動」吐的：形狀由你決定，時機由你決定。你想在「抓取途中」就報「正在抓 Acme」、想在摘要還沒寫完時就把半句話推出去——這些 updates 做不到（它只在節點整個跑完才有），全靠 custom。

一句話分工：updates 報「一個節點跑完了」，custom 報「一個節點跑到一半的細節」。 control panel 要的即時感，主要來自後者。

⚠️ 別看到 values 就順手用。它每一步都把整包 state 塞給你——當 state 裡堆滿了每個競品的原始抓取內容（可能好幾 KB），你等於每一步都把整包資料重傳一次過網路。要嘛用 updates（只傳變化的那塊），要嘛用 custom（只傳你精挑的欄位）。values 留給「我在本機想 dump 每步完整快照來 debug」的場合。

還有一個入口叫 graph.streamEvents()（LangChain 底層的事件流，顆粒更細）。它更底層、更囉嗦，這門課用不到；graph.stream 的 streamMode 已經覆蓋我們需要的一切。知道有這東西、但先用 streamMode。

三、從節點吐出中間結果：config.writer 與那條 seam

現在把上一節的 custom chunk 真的生出來。這一節的程式碼分四個檔案，全部離線可跑、不需要任何 API 金鑰——這正是第 2 課那條 seam（LLM 邊界）在這一課的兌現：純邏輯（去重、進度計算）和真正打 LLM 的部分切開，純邏輯用 stub（假模型）就能測。

3.1 先定義「事件」——順便撿回第 1 課的槓桿

第 1 課說過，我們全端都用 TypeScript，圖的一個好處是「state 的形狀從後端流到前端，是同一種語言、可以共用型別」。streaming 是這個槓桿最直接的兌現場合：後端吐出的事件、前端接住的事件，就是同一個型別。我們先把它定義在一個共用檔：

// shared/events.ts —— 後端吐它、前端接它，同一份型別
export type NodeStatus = "pending" | "fetching" | "summarizing" | "done";

export interface Source {
  url: string;
  kind: "pricing" | "changelog" | "blog" | "jobs";
  text: string;
}

/** 我們自己用 custom 模式吐出去的領域事件 */
export type DigestEvent =
  | { type: "node_status"; competitor: string; status: NodeStatus }
  | { type: "source_found"; competitor: string; count: number }
  | { type: "summary_delta"; competitor: string; delta: string };

/** 後端最後真正寫進 HTTP 串流的東西：custom 事件 ＋ 幾個框架級／收尾事件 */
export type WireEvent =
  | DigestEvent
  | { type: "state_update"; node: string; update: unknown } // 來自 updates 模式
  | { type: "done" }
  | { type: "error"; message: string };

💡 DigestEvent 是一個 TypeScript 的 discriminated union（用 type 欄位辨別）。它讓後端「吐事件」和前端「用 switch 收事件」兩邊都被編譯器盯著——少打一個欄位、多一種沒處理的 case，tsc 當場報給你看。這種「一份型別貫穿全端」的踏實感，就是第 1 課選全 TypeScript 的理由之一。

3.2 seam 的兩邊：純邏輯 ＋ 可抽換的模型

抓取和摘要是「會打外部世界／LLM」的部分，我們把它們定義成介面，這樣測試時可以塞 stub、上線時塞真貨。去重（dedup）則是純函式，不需要任何外部依賴：

// digest/seam.ts
import type { Source } from "../shared/events";

/** 抓料的 seam：真實版打 RSS/changelog/pricing（第 3 課），測試版用 stub */
export interface Fetcher {
  fetch(competitor: string): Promise<Source[]>;
}

/** 摘要的 seam：串流版，一段一段 yield 文字（真實版打 LLM，測試版用 stub） */
export interface Summarizer {
  summarizeStream(competitor: string, sources: Source[]): AsyncIterable<string>;
}

/** 純邏輯：同一次執行內去重。不需要金鑰、可離線單元測試。（機制見第 4 課） */
export function dedup(sources: Source[]): Source[] {
  const seen = new Set<string>();
  return sources.filter((s) => {
    const key = `${s.kind}::${s.text}`;
    if (seen.has(key)) return false;
    seen.add(key);
    return true;
  });
}

再寫離線用的 stub。重點：stub 也能串流——summarizeStream 是個 async generator，一個字一個字 yield，正好模擬真 LLM 逐字吐 token 的節奏，讓我們不花一毛錢就能把「摘要逐字長出來」整條串流測起來：

// digest/stubs.ts —— 離線假貨，一毛錢不花
import type { Fetcher, Summarizer } from "./seam";
import type { Source } from "../shared/events";

const delay = (ms: number) => new Promise((r) => setTimeout(r, ms));

export const stubFetcher: Fetcher = {
  async fetch(name) {
    await delay(200 + Math.random() * 400); // 模擬網路延遲，讓各競品交錯
    return [
      { url: `https://${name}.com/pricing`,   kind: "pricing",   text: `${name} Pro $15/座` },
      { url: `https://${name}.com/changelog`, kind: "changelog", text: `${name} 上線了 AI 週報` },
      { url: `https://${name}.com/blog`,       kind: "blog",      text: `${name} 連兩週談 SOC 2` },
      { url: `https://${name}.com/jobs`,        kind: "jobs",      text: `${name} 在招企業銷售` },
    ];
  },
};

export const stubSummarizer: Summarizer = {
  async *summarizeStream(name, sources) {
    const text = `${name}：本週偵測到 ${sources.length} 則來源。Pro 方案漲價、上線 AI 週報，疑似進軍企業市場。`;
    for (const ch of text) {
      await delay(12); // 一個字一個字吐，模擬 LLM token 節奏
      yield ch;
    }
  },
};

3.3 節點裡吐 custom 事件：config.writer

現在是這一節的主角。在 LangGraph.js 裡，每個節點函式的第二個參數是 config（型別 LangGraphRunnableConfig），它身上帶一個 writer 函式——你丟給 config.writer(...) 的任何東西，都會以 custom chunk 的形式流出去。這就是我們把「跑到一半的細節」推出去的水龍頭。

我把 graph 壓成三個粗粒度階段（第 4 課的 graph 級平行 fan-out 這裡不重講；為了聚焦串流，平行併發用節點內 Promise.all 示範，串出來多競品交錯的效果一樣）：

// digest/graph.ts
import {
  Annotation, StateGraph, START, END,
  type LangGraphRunnableConfig,
} from "@langchain/langgraph";
import type { Fetcher, Summarizer } from "./seam";
import { dedup } from "./seam";
import type { DigestEvent, Source } from "../shared/events";

interface CompetitorResult { name: string; sourceCount: number; summary: string; }

// state 的定義方式第 2 課已教，這裡照用（reducer 把平行寫回的片段合併）
const DigestState = Annotation.Root({
  competitors: Annotation<string[]>(),
  raw: Annotation<Record<string, Source[]>>({
    reducer: (a, b) => ({ ...a, ...b }),
    default: () => ({}),
  }),
  results: Annotation<Record<string, CompetitorResult>>({
    reducer: (a, b) => ({ ...a, ...b }),
    default: () => ({}),
  }),
});
type DigestStateT = typeof DigestState.State;

// 節點 1：抓料。每個競品各自吐「開始抓」「抓到幾則」
function makeFetchAll(fetcher: Fetcher) {
  return async (state: DigestStateT, config: LangGraphRunnableConfig) => {
    const emit = (e: DigestEvent) => config.writer?.(e); // ← 水龍頭
    const raw: Record<string, Source[]> = {};
    await Promise.all(
      state.competitors.map(async (name) => {
        emit({ type: "node_status", competitor: name, status: "fetching" });
        const sources = await fetcher.fetch(name);
        raw[name] = sources;
        emit({ type: "source_found", competitor: name, count: sources.length });
      }),
    );
    return { raw };
  };
}

// 節點 2：去重 + 摘要。摘要「逐字」吐 summary_delta，最後標 done
function makeSummarizeAll(summarizer: Summarizer) {
  return async (state: DigestStateT, config: LangGraphRunnableConfig) => {
    const emit = (e: DigestEvent) => config.writer?.(e);
    const results: Record<string, CompetitorResult> = {};
    await Promise.all(
      state.competitors.map(async (name) => {
        emit({ type: "node_status", competitor: name, status: "summarizing" });
        const clean = dedup(state.raw[name] ?? []); // 純邏輯（第 4 課）
        let summary = "";
        for await (const piece of summarizer.summarizeStream(name, clean)) {
          summary += piece;
          emit({ type: "summary_delta", competitor: name, delta: piece });
        }
        results[name] = { name, sourceCount: clean.length, summary };
        emit({ type: "node_status", competitor: name, status: "done" });
      }),
    );
    return { results };
  };
}

// 節點 3：打包本週結果。⚠️ 跨週 diff 是第 7 課，這裡只把「本週」整理好
async function assemble(state: DigestStateT) {
  return { results: state.results };
}

/** 依賴用注入的（seam）——換 stub 或真貨，這裡一行都不用改 */
export function makeDigestGraph(deps: { fetcher: Fetcher; summarizer: Summarizer }) {
  return new StateGraph(DigestState)
    .addNode("fetchAll", makeFetchAll(deps.fetcher))
    .addNode("summarizeAll", makeSummarizeAll(deps.summarizer))
    .addNode("assemble", assemble)
    .addEdge(START, "fetchAll")
    .addEdge("fetchAll", "summarizeAll")
    .addEdge("summarizeAll", "assemble")
    .addEdge("assemble", END)
    .compile();
}

注意 config.writer?.(e) 那個 ?.：只有當你用會產生 custom 的 streamMode 去跑時，writer 才存在；用 graph.invoke()（不串流）跑同一張 graph 時它是 undefined，?. 讓節點自動變回一句話都不吐、照常運作。同一張 graph，串流時會報進度、不串流時安靜跑完——不用改任何節點程式碼。

3.4 一次消費兩種模式：streamMode: ["updates", "custom"]

streamMode 可以給一個陣列，同時要好幾種模式。這時每個 chunk 會變成 [模式名, 資料] 的 tuple，你就知道這段是哪種模式來的：

// digest/run.ts —— 把 graph 跑起來、把兩種模式的中間結果印出來
import { makeDigestGraph } from "./graph";
import { stubFetcher, stubSummarizer } from "./stubs";

export async function run(competitors: string[]) {
  const graph = makeDigestGraph({ fetcher: stubFetcher, summarizer: stubSummarizer });

  const stream = await graph.stream(
    { competitors },
    { streamMode: ["updates", "custom"] }, // ← 兩種一起要
  );

  for await (const [mode, chunk] of stream) {
    if (mode === "custom") {
      console.log("custom ", chunk);              // 我們自己吐的 DigestEvent
    } else {
      console.log("updates", JSON.stringify(chunk)); // 框架給的 { 節點: 更新 }
    }
  }
}

run(["Acme", "BoltBoard", "Cascade"]);

跑 tsx digest/run.ts（不需要金鑰），你會看到 custom 事件密集地、交錯地吐出來（三個競品同時在跑），中間穿插著三次粗粒度的 updates（fetchAll、summarizeAll、assemble 各跑完一次）。這就是那 40 秒黑箱裡真正發生的事，現在它全都看得見了。

3.5 這條 seam 讓你能離線斷言串流——不花一毛錢

因為抓取和摘要都是注入的 stub，整條串流可以寫成一個不需要金鑰、CI 裡就能跑的測試。這是 seam 最實際的回報：

// digest/stream.test.ts —— node --test，零金鑰
import { test } from "node:test";
import assert from "node:assert/strict";
import { makeDigestGraph } from "./graph";
import { stubFetcher, stubSummarizer } from "./stubs";
import type { DigestEvent } from "../shared/events";

test("每個競品都串出 fetching → summarizing → done，且摘要有逐字 delta", async () => {
  const graph = makeDigestGraph({ fetcher: stubFetcher, summarizer: stubSummarizer });
  const stream = await graph.stream(
    { competitors: ["Acme", "BoltBoard"] },
    { streamMode: "custom" },
  );

  const events: DigestEvent[] = [];
  for await (const e of stream) events.push(e as DigestEvent);

  for (const name of ["Acme", "BoltBoard"]) {
    const statuses = events
      .filter((e) => e.type === "node_status" && e.competitor === name)
      .map((e) => (e as Extract<DigestEvent, { type: "node_status" }>).status);
    assert.deepEqual(statuses, ["fetching", "summarizing", "done"]);

    const deltas = events.filter(
      (e) => e.type === "summary_delta" && e.competitor === name,
    );
    assert.ok(deltas.length > 5, "摘要應該是逐字串流、不是一次吐完");
  }
});

💡 這裡也順手示範了 streamMode: "custom"（給單一字串，不是陣列）——chunk 就直接是你吐的物件，不再包成 [模式, 資料] tuple。給陣列才有 tuple；給單一模式就是裸資料。

3.6 真貨（示範，需金鑰）：一個會串流的 LLM 摘要器

離線測完，換上真的。seam 的回報就在這一刻：只要有另一個實作了 Summarizer 的東西，上面 graph／server 一行都不用改。 這個真貨用你已經熟的 Anthropic SDK 串流：

// digest/anthropic.ts —— 示範：需要 ANTHROPIC_API_KEY。換上它，其餘程式碼一行不動
import Anthropic from "@anthropic-ai/sdk";
import type { Summarizer } from "./seam";
import type { Source } from "../shared/events";

function buildPrompt(name: string, sources: Source[]): string {
  const body = sources.map((s) => `- [${s.kind}] ${s.url}\n  ${s.text}`).join("\n");
  return (
    `你是競品情報分析師。下面是本週從「${name}」抓到的來源，` +
    `請用繁體中文寫 2–3 句話，只點出「值得決策者注意的變化」，不要客套：\n\n${body}`
  );
}

export function makeAnthropicSummarizer(apiKey: string): Summarizer {
  const client = new Anthropic({ apiKey });
  return {
    async *summarizeStream(name, sources) {
      const stream = client.messages.stream({
        model: "claude-3-5-haiku-latest",
        max_tokens: 300,
        messages: [{ role: "user", content: buildPrompt(name, sources) }],
      });
      for await (const event of stream) {
        if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
          yield event.delta.text; // 真 LLM 的 token，逐段 yield 出去
        }
      }
    },
  };
}

節點裡那段 for await (const piece of summarizer.summarizeStream(...)) 完全不知道背後是 stub 還是 Anthropic——它只看到「一個會逐段吐字串的 async generator」。這就是 seam 的意義：測試零金鑰、上線換真貨、中間的串流管線一次寫好、兩邊共用。

💡 我們這裡用 Anthropic 原生 SDK 接 token、再自己用 config.writer 轉成 summary_delta。另一條路是把節點裡的模型換成 LangChain 的 chat model（如 ChatAnthropic）——那樣 streamMode: "messages" 會自動把 token 以 [chunk, metadata] 流出來，metadata.langgraph_node 還會告訴你這段 token 是哪個節點吐的，你連 summary_delta 都不用自己組。兩條路都對；我們選前者，是因為它讓 seam 保持乾淨——摘要器就是「一個吐字串的 async generator」，跟框架無關、stub 也三行就寫完。

到這裡，第一個問題答完了：LangGraph.js 用 graph.stream(input, { streamMode }) 串流，五種模式裡 control panel 主要用 updates（框架自動的粗粒度進度）＋ custom（你用 config.writer 精準吐的細節）；把它們串出來，就是一連串你能 for await 的中間結果。

四、後端把串流吐給前端：一條 HTTP 串流

現在中間結果在後端手上了，得把它送過網路到瀏覽器。這一節回答第二個問題的前半：Node 後端怎麼把這個串流吐給前端。

4.1 選傳輸方式：為什麼用「一行一個 JSON」

graph.stream 給你的是一串 JS 物件；HTTP 傳的是位元組。中間需要一個線路格式（wire format）。這門課用最簡單、最好 debug 的一種：NDJSON——一行一個 JSON 物件（JSON.stringify(event) + "\n"）。伺服器每拿到一個事件，就寫一行；前端每讀到一行，就 JSON.parse 一個事件。

為什麼不是別的？給你一個誠實的對照，順便標出邊界：

  • NDJSON over fetch 串流（本課選這個）：支援 POST（我們要把「追哪些競品」送上去）、格式簡到不可能寫錯、前端用 fetch 的 ReadableStream 直接讀。缺點是要自己處理「一次讀到的位元組可能切在半行中間」——這其實是個很值得學的細節，下一節就處理。
  • SSE（Server-Sent Events）＋ EventSource：瀏覽器原生、會自動重連，是「伺服器單向推事件」的標準。但 EventSource 只能 GET，沒法直接帶 POST body；而且它的框架格式（data: 前綴、\n\n 分隔）你還是得配合。若你的觸發是純 GET、又想要免費的自動重連，SSE 是更好的選擇——這就是它的適用邊界。
  • WebSocket：雙向即時。但我們這裡是單向（伺服器 → 瀏覽器）推進度，用 WebSocket 是殺雞用牛刀，還多一套連線狀態要管。

換句話說：不是「NDJSON 最好」，而是「在『POST 送設定、單向串回進度』這個情境下，NDJSON 最省事」。 你的情境變了（例如需要斷線自動重連），結論就可能倒向 SSE。

4.2 一個最小、但真的能跑的 Node 串流伺服器

第 1 課提過 LangGraph.js 沒有內建 HTTP server——這是它真實的缺口，也是第 12 課「把 graph 包成可上線服務」要正式處理的事（CORS 收緊、金鑰、觀測、優雅關閉…）。這一節只做「能在本機把串流吐給 React」的最小版，用 Node 內建的 http，不裝任何框架：

// server.ts —— 最小可跑的串流端點。硬化（CORS/金鑰/觀測）留給第 12 課
import { createServer, type IncomingMessage } from "node:http";
import { makeDigestGraph } from "./digest/graph";
import { stubFetcher, stubSummarizer } from "./digest/stubs";
import { makeAnthropicSummarizer } from "./digest/anthropic"; // 真貨，見 3.6
import type { WireEvent } from "./shared/events";

function readJson(req: IncomingMessage): Promise<{ competitors?: string[] }> {
  return new Promise((resolve, reject) => {
    let data = "";
    req.on("data", (c) => (data += c));
    req.on("end", () => {
      try { resolve(JSON.parse(data || "{}")); } catch (e) { reject(e); }
    });
    req.on("error", reject);
  });
}

// 沒金鑰就用 stub，整個伺服器＋前端可以完全離線跑起來
const summarizer = process.env.ANTHROPIC_API_KEY
  ? makeAnthropicSummarizer(process.env.ANTHROPIC_API_KEY)
  : stubSummarizer;
const fetcher = stubFetcher; // 真實抓取見第 3 課

const server = createServer(async (req, res) => {
  // 本機開發用的最小 CORS —— 第 12 課會收緊成白名單
  res.setHeader("Access-Control-Allow-Origin", "*");
  res.setHeader("Access-Control-Allow-Headers", "content-type");
  if (req.method === "OPTIONS") { res.writeHead(204); res.end(); return; }

  if (req.method === "POST" && req.url === "/api/digest/stream") {
    const { competitors = [] } = await readJson(req);

    res.writeHead(200, {
      "Content-Type": "application/x-ndjson; charset=utf-8",
      "Cache-Control": "no-cache",
      "X-Accel-Buffering": "no", // 叫 nginx 之類的反向代理「別替我緩衝」，否則串流會被卡住
    });

    const write = (e: WireEvent) => res.write(JSON.stringify(e) + "\n");

    // 使用者關掉分頁 → 中止 graph，別再白燒 LLM 的錢
    const ac = new AbortController();
    req.on("close", () => ac.abort());

    try {
      const graph = makeDigestGraph({ fetcher, summarizer });
      const stream = await graph.stream(
        { competitors },
        { streamMode: ["updates", "custom"], signal: ac.signal },
      );

      for await (const [mode, chunk] of stream) {
        if (mode === "custom") {
          write(chunk as WireEvent); // 已經是 DigestEvent
        } else {
          // updates 的 chunk 是 { 節點名: 更新 }，包成一個 state_update 事件
          const [node, update] = Object.entries(chunk as Record<string, unknown>)[0];
          write({ type: "state_update", node, update });
        }
      }
      write({ type: "done" });
    } catch (err) {
      write({ type: "error", message: err instanceof Error ? err.message : String(err) });
    } finally {
      res.end();
    }
    return;
  }

  res.writeHead(404);
  res.end();
});

server.listen(8787, () => console.log("digest stream on http://localhost:8787"));

這段有三個「產品級」的細節，比「能吐字」更重要，值得停下來看：

  • X-Accel-Buffering: no（和它代表的整類問題）：串流最經典的坑是——你程式碼明明一段段寫了，使用者卻還是等到最後才一次看到全部。九成是中間有個反向代理／CDN 把你的回應緩衝起來湊成一大包才轉發。這個 header 是對 nginx 說「這條別緩衝」。⚠️ 上線時「為什麼串流不即時」的頭號嫌犯就是它，先記著。
  • AbortController ＋ req.on("close")：使用者關掉分頁，req 觸發 close，我們 ac.abort()，把 signal 傳進 graph.stream 讓它停下來。少了這段，使用者一關頁面，後端還在傻傻地把三個競品的 LLM 呼叫全跑完——燒的是你的錢。這一刀直接連到第 8～9 課「控制預算」的精神：別為沒人要看的結果付費。
  • try/finally 一定 res.end()：串流回應必須明確關閉，否則瀏覽器那頭會一直空等。錯誤也走同一條線路——包成 { type: "error" } 吐給前端，讓 UI 能顯示「這張卡爆了」，而不是整頁卡死。

💡 這裡我用了 graph.stream(input, { signal }) 把中止訊號傳進去。順帶一提，res.write() 在寫得太快、對端來不及收時會回傳 false（背壓訊號），嚴謹的伺服器要在這時等 drain 事件。本課的資料量小、先不處理；真要上量時的完整背壓處理，一樣歸第 12 課。規則有了，邊界也標了。

五、React 接住串流：reduce 成一塊看板

線路另一端是瀏覽器。這一節回答第二個問題的後半：React 怎麼接住串流、即時更新儀表板。 這也是本課技術上的高潮——因為**「怎麼接」直接決定了你做出來的是 control panel 還是 chat 泡泡**。

5.1 讀一條 fetch 串流：那個「切在半行中間」的坑

前端用 fetch 打端點，然後不要 await res.json()（那會等整包收完，等於放棄串流）。改成讀 res.body 這個 ReadableStream，一塊一塊拿位元組：

// web/useDigestStream.ts
import { useCallback, useReducer, useState } from "react";
import type { WireEvent } from "../shared/events";
import { cardsReducer, type CardsState } from "./cardsReducer";

type Phase = "idle" | "running" | "done" | "error";

export function useDigestStream() {
  const [cards, dispatch] = useReducer(cardsReducer, {} as CardsState);
  const [phase, setPhase] = useState<Phase>("idle");

  const start = useCallback(async (competitors: string[]) => {
    dispatch({ kind: "reset", competitors });
    setPhase("running");

    const res = await fetch("http://localhost:8787/api/digest/stream", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ competitors }),
    });
    if (!res.body) { setPhase("error"); return; }

    const reader = res.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
      const { value, done } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split("\n");
      buffer = lines.pop() ?? ""; // ← 最後一段可能是「半行」，留在 buffer 等下一塊

      for (const line of lines) {
        if (!line.trim()) continue;
        dispatch({ kind: "event", event: JSON.parse(line) as WireEvent });
      }
    }

    setPhase("done");
  }, []);

  return { cards, phase, start };
}

整段最關鍵的三行是 buffer 那段。⚠️ reader.read() 給你的位元組，切點是任意的——它很可能在 {"type":"summ 這種半行的地方就斷了，因為網路封包不管你的 JSON 邊界。所以我們：把新位元組接到 buffer 後面 → 用 \n 切開 → 最後一段（可能是半行）用 lines.pop() 留回 buffer，等下一塊補齊 → 其餘完整的行才拿去 JSON.parse。少了這一步，你遲早會 JSON.parse 到半行然後整條串流炸掉。這個坑 SSE 也一樣有，是所有串流解析的共同必修課。

⚠️ 用 <button onClick> 觸發 start，別塞進 useEffect。React 18 的 StrictMode 會把 effect 跑兩次，你會莫名其妙發兩條串流。使用者按鈕觸發天生就沒這問題。

5.2 核心：把事件 reduce 進一塊「以競品為 key」的看板

到了。這是整課的樞紐。看事件怎麼變成畫面狀態——用一個 reducer，把 WireEvent 收攏進一個「競品名 → 卡片」的 map：

// web/cardsReducer.ts
import type { WireEvent, NodeStatus } from "../shared/events";

export type CardStatus = NodeStatus | "error";
export interface Card {
  name: string;
  status: CardStatus;
  sourceCount: number;
  summary: string; // 逐字長出來的摘要
}
export type CardsState = Record<string, Card>; // ← 以「競品名」為 key，數量固定

export type CardsAction =
  | { kind: "reset"; competitors: string[] }
  | { kind: "event"; event: WireEvent };

export function cardsReducer(state: CardsState, action: CardsAction): CardsState {
  if (action.kind === "reset") {
    const next: CardsState = {};
    for (const name of action.competitors) {
      next[name] = { name, status: "pending", sourceCount: 0, summary: "" };
    }
    return next;
  }

  const e = action.event;
  switch (e.type) {
    case "node_status":
      return patch(state, e.competitor, { status: e.status });
    case "source_found":
      return patch(state, e.competitor, { sourceCount: e.count });
    case "summary_delta": // 就地把新的一段字「接」在後面
      return patch(state, e.competitor, {
        summary: (state[e.competitor]?.summary ?? "") + e.delta,
      });
    case "error":
      // 全域錯誤：把還沒完成的卡片標成 error
      return Object.fromEntries(
        Object.entries(state).map(([k, c]) => [
          k, c.status === "done" ? c : { ...c, status: "error" as const },
        ]),
      );
    case "state_update": // 來自 updates 模式，這裡不驅動卡片，留給 5.4 的原始日誌
    case "done":
    default:
      return state;
  }
}

function patch(state: CardsState, name: string, fields: Partial<Card>): CardsState {
  const cur = state[name];
  if (!cur) return state;
  return { ...state, [name]: { ...cur, ...fields } };
}

盯著 CardsState = Record<string, Card> 這行看三秒。 這就是「control panel 不是 chat 泡泡」在程式碼層面的分水嶺：

  • 我們的畫面狀態是一個以競品名為 key、數量固定的 map。每來一個事件，我們是去更新那個 key 底下的卡片（patch）——狀態燈換色、來源數填上、摘要文字接長。卡片就地變化。
  • 如果這是個 chat 泡泡，畫面狀態會是一個 Message[] 陣列，每來一個事件就 push 一則新訊息，list 越滾越長。

同一條事件串流，餵給 reducer[key] = ... 就長成儀表板，餵給 list.push(...) 就長成聊天室。 差別不在後端、不在網路、就在前端這個資料結構的選擇。這一節就是把第 1 課那個定位，落到了一行型別上。

5.3 畫出來：一張競品一張卡

元件就很直白了——把 map 的值攤成卡片。注意它渲染的是「固定幾張、就地更新」的卡，不是「一直冒出來」的訊息：

// web/ControlPanel.tsx
import { useDigestStream } from "./useDigestStream";
import type { CardStatus } from "./cardsReducer";

const LIGHT: Record<CardStatus, string> = {
  pending: "⚪ 排隊中",
  fetching: "🟡 抓取中",
  summarizing: "🟢 摘要中",
  done: "✅ 完成",
  error: "🔴 失敗",
};

export function ControlPanel() {
  const { cards, phase, start } = useDigestStream();
  const list = Object.values(cards);

  return (
    <div>
      <button
        disabled={phase === "running"}
        onClick={() => start(["Acme", "BoltBoard", "Cascade"])}
      >
        {phase === "running" ? "分析中…" : "開始分析本週競品"}
      </button>

      <div style={{ display: "flex", gap: 16, marginTop: 16 }}>
        {list.map((c) => (
          <article key={c.name} style={{ width: 260, border: "1px solid #ddd", borderRadius: 8, padding: 12 }}>
            <header style={{ display: "flex", justifyContent: "space-between" }}>
              <strong>{c.name}</strong>
              <span>{LIGHT[c.status]}</span>
            </header>
            {c.sourceCount > 0 && <p>找到 {c.sourceCount} 則來源</p>}
            <p style={{ color: "#444", minHeight: 48 }}>{c.summary}</p>
          </article>
        ))}
      </div>
    </div>
  );
}

用 Vite 起一個 React app、把這個元件放進去，另一個 terminal tsx server.ts（不設金鑰就是 stub），按下按鈕——你就會看到本課開頭那三張快照真的動起來：三張卡各自亮燈、來源數各自跳出、摘要各自逐字長出來。完全離線、一毛錢不花。 想看真 LLM 摘要，設好 ANTHROPIC_API_KEY 再跑一次，前端一個字都不用改。

5.4 （選讀）updates 事件拿來做什麼？一條原始執行日誌

還記得後端也把 updates 模式包成 state_update 事件送過來了嗎？它不驅動卡片（卡片由精準的 custom 事件驅動），但很適合做一條給你自己看的**「graph 執行日誌」**——哪個節點在哪個時間點跑完、寫了什麼。在 reducer 裡加個 log: string[] 累積它即可：

// 在收到 state_update 時（示意，接到你的 log 狀態上）
// case "state_update":
//   return { ...state, log: [...state.log, `${new Date().toISOString()} ✓ ${e.node}`] };

這正好體現前面那句分工：custom 是精修的產品 UI，updates 是免費的粗粒度骨架。 兩種模式各司其職，你不用二選一。

到這裡，第二個問題答完了：後端把 graph.stream 的中間結果逐一寫成 NDJSON、用一條 HTTP 串流吐出；React 用 fetch 的 ReadableStream 逐塊讀、處理好半行 buffer、JSON.parse 成事件，再用 reducer reduce 進一塊以競品為 key 的看板——每個事件就地更新對應卡片，畫面即時活起來。

六、為什麼這是 control panel、不是 chat 泡泡

第三個問題，其實你在 5.2 已經摸到了答案。這一節把它講到底、釘死。

第 1 課從產品角度論證過：忙碌的 PM／創辦人不想再多一個要他打字伺候的框，他要「答案自己送上門」。這裡不重複那個論證——這一節補的是「串流之後」才浮現的技術面差異：拿到同一條即時事件串流，做成 chat 和做成 control panel，差在哪？

flowchart TB
    E["同一條事件串流"]
    subgraph CHAT["chat 泡泡：把事件 append 進一個 list"]
      direction TB
      C1["訊息 1"] --> C2["訊息 2"] --> C3["訊息 3 …越滾越長"]
    end
    subgraph PANEL["control panel：把事件 reduce 進一塊固定看板"]
      direction LR
      P1["Acme 卡（就地更新）"]
      P2["BoltBoard 卡（就地更新）"]
      P3["Cascade 卡（就地更新）"]
    end
    E --> CHAT
    E --> PANEL

三個決定性的差別：

  • 資料結構不同：chat 是 Message[]（一直 push、無上限地變長，順序＝時間）；control panel 是 Record<競品, 卡片>（key 固定、就地更新，結構＝領域）。這正是 5.2 那一行 CardsState = Record<string, Card>。
  • 回答的問題不同：chat 回答「我剛剛問的那句，它怎麼回」（對話的一來一回）；control panel 回答「我盯的這幾個競品，現在各自到哪一步、狀態如何」（多實體的狀態與進度）。我們的產品從頭到尾沒有「使用者問了一句」這回事——是排程或按鈕觸發，然後看板自己亮完。
  • 注意力成本不同：chat 要使用者讀完整串對話才知道結論，還得自己記著上一輪講過什麼；control panel 一眼掃過去就知道「三個競品、兩個完成、一個在抓料」。誰扛下「整理成一眼看懂的狀態」這個累活——是我們（後端 graph ＋ 前端 reducer）——誰就收得起月費。這條線，第 1 課就畫過，這一課用程式碼把它兌現了。

⚠️ 一個要講清楚的邊界，免得你矯枉過正：「摘要文字逐字長出來」並不會讓它變成 chat 泡泡。 逐字浮現只是串流的視覺效果，重點在那段字流進的容器是什麼——它流進的是「Acme 這張狀態卡」的一個欄位，不是「對話串」裡新冒出來的一則發言。容器是以競品為 key 的固定看板，它就是 control panel；容器是越滾越長的訊息 list，它才是 chat 泡泡。分野在資料結構，不在文字會不會動。

所以，第三個問題答完了：這個前端是 control panel，因為它把串流事件 reduce 進一塊「以競品為 key、就地更新」的看板，呈現的是「多競品情報的狀態與進度」；chat 泡泡則是把事件 append 進一個越滾越長的訊息 list，呈現「一問一答的對話」。同一條串流，資料結構一換，產品的靈魂就換了。

這一課建立了什麼

  • graph.stream(input, { streamMode }) 是串流的唯一入口。五種模式裡，control panel 的主力是 updates（框架自動吐的粗粒度「哪個節點跑完、寫了什麼」）＋ custom（你在節點裡用 config.writer(...) 精準吐的細節事件）。messages 是「讓 LLM token 逐字浮現」的加菜、values／debug 這場景少用。給 streamMode 一個陣列時，chunk 變成 [模式, 資料] tuple。
  • 中間結果從節點吐出，靠 config.writer：節點第二參數 config（LangGraphRunnableConfig）上的 writer，你丟什麼它就以 custom chunk 流出什麼；用 ?. 讓同一張 graph 在不串流時安靜跑。這條路完全落在第 2 課那條 seam 上——抓取／摘要是注入的介面，純邏輯（去重）＋ stub（假模型）讓整條串流離線、零金鑰、可測。
  • 後端→前端的傳輸＝一條 HTTP 串流 ＋ NDJSON（一行一個 JSON）。挑它是因為「POST 送設定、單向串回進度」這情境最省事；情境不同（要自動重連）則 SSE 更合適。伺服器端要注意三件產品級小事：別被代理緩衝（X-Accel-Buffering: no）、使用者關頁面就 AbortController 中止（別白燒錢）、finally 一定 res.end()。
  • React 端＝讀 res.body 串流 ＋ 處理半行 buffer ＋ reduce 進看板。reader.read() 的切點是任意的，一定要把「最後半行」留回 buffer 等下一塊；再把事件 reduce 進 Record<競品, 卡片>，每個事件就地更新對應卡片。
  • control panel vs chat 泡泡的技術分野，就是一行資料結構：Record<競品, 卡片>（就地更新、key 固定、呈現多實體狀態與進度）vs Message[]（無限 append、呈現一問一答）。同一條串流，reducer[key]=… 得到儀表板，list.push(…) 得到聊天室。第 1 課的定位，這一課用程式碼兌現。

下一課（第 6 課）：現在畫面能即時看到 graph 跑到哪了，我們就要在「送出週報之前」插一道真人審核卡——用 interrupt() 把 graph 暫停在分析師面前、等他按下「核准」再 resume。串流讓他看得到、HITL 讓他攔得住。至於「記得上週、只報 diff」怎麼把跨週結果接上這塊看板，是第 7 課那記殺手鐧的事。