本文 AI 產出，尚未審核

LangChain 教學：RunnableSeq 與 RunnableBranch 的高度模組化流程

簡介

在 LangChain 生態系統中，Runnable 是一個抽象概念，代表「可執行的流程單元」。透過將不同的模型、工具或自訂函式包裝成 Runnable，開發者可以像拼圖一樣自由組合，打造出符合業務需求的複雜工作流。

在實務開發裡，兩個常用的組合器是 RunnableSeq（序列執行）與 RunnableBranch（條件分支）。它們分別負責 「依序串接」 與 「根據條件分流」，讓整條 pipeline 的可讀性、可測試性與可維護性大幅提升。

本篇文章將以 LangChainJS（JavaScript/TypeScript 版）為例，深入說明這兩個組合器的使用方式、實作細節與最佳實踐，幫助初學者快速上手，同時提供中級開發者在大型專案中如何善用模組化流程的參考。

核心概念

1. Runnable 基礎回顧

在 LangChain 中，Runnable 是一個遵守 invoke(input: any): Promise<any> 介面的物件。只要實作 invoke 方法，就可以被其他 Runnable 包裝或串接。最常見的 Runnable 包括：

類型 代表 典型用途
RunnableLambda 任意函式 快速包裝簡單邏輯
ChatOpenAI OpenAI Chat Completion 呼叫 LLM
Tool 外部 API、資料庫查詢 與外部系統互動

Tip：在設計自訂 Runnable 時，盡量保持 純函式（pure function）特性，讓它只依賴輸入、回傳輸出，避免全域狀態，這樣才能在序列或分支中安全重用。

2. RunnableSeq：串接多個 Runnable

RunnableSeq 讓你把多個 Runnable 依序 執行，前一個的輸出自動成為下一個的輸入。概念上類似 Unix pipeline (cat | grep | awk)。

2.1 基本語法

import { RunnableSeq, RunnableLambda } from "@langchain/core/runnables";

// 定義三個簡單的 Runnable
const step1 = new RunnableLambda(async (input) => {
  // 把字串轉成大寫
  return input.toUpperCase();
});

const step2 = new RunnableLambda(async (input) => {
  // 取前 5 個字元
  return input.slice(0, 5);
});

const step3 = new RunnableLambda(async (input) => {
  // 加上前綴
  return `Result: ${input}`;
});

// 用 RunnableSeq 組成 pipeline
const pipeline = RunnableSeq.from([step1, step2, step3]);

// 執行
const output = await pipeline.invoke("langchain tutorial");
console.log(output); // => "Result: LANGC"

重點：RunnableSeq.from([...]) 會自動把陣列中的每個元素包裝成 Runnable，若已經是 Runnable，則直接使用。

2.2 結合 LLM 與工具

import { ChatOpenAI } from "@langchain/openai";
import { Tool } from "@langchain/core/tools";

// LLM：產生問題
const generateQuestion = new ChatOpenAI({ temperature: 0.7 }).bind({
  prompt: "請根據以下主題產生一個簡短問題：{{input}}"
});

// 工具：呼叫外部 API（模擬）
class WeatherTool extends Tool {
  async _call(input) {
    // 假設回傳天氣資訊的字串
    return `今天 ${input} 的天氣是晴時多雲，溫度 25°C。`;
  }
}
const weatherTool = new WeatherTool();

// 從 LLM 產生關鍵字，再交給工具查詢
const pipeline = RunnableSeq.from([
  generateQuestion,
  new RunnableLambda(async (question) => {
    // 從問題抽取城市名稱（簡易正則）
    const match = question.match(/關於 (.+?) 的天氣/);
    return match ? match[1] : "台北";
  }),
  weatherTool
]);

const result = await pipeline.invoke("台北的天氣");
console.log(result);
// => "今天 台北 的天氣是晴時多雲，溫度 25°C。"

2.3 使用 RunnableMap 在 Seq 中分支

有時候需要 同時 執行多條支線，然後把結果合併。可以在 RunnableSeq 中插入 RunnableMap：

import { RunnableMap } from "@langchain/core/runnables";

const translate = new RunnableLambda(async (text) => {
  // 假裝呼叫翻譯 API
  return `EN: ${text}`;
});

const sentiment = new RunnableLambda(async (text) => {
  // 假裝做情感分析
  return text.includes("好") ? "positive" : "neutral";
});

const map = RunnableMap.from({
  translation: translate,
  sentiment: sentiment,
});

const pipeline = RunnableSeq.from([
  new RunnableLambda((input) => input.trim()),
  map, // 同時得到 translation 與 sentiment
  new RunnableLambda((obj) => ({
    ...obj,
    summary: `${obj.translation} (${obj.sentiment})`,
  })),
]);

const out = await pipeline.invoke("這是一段很好的文字。");
console.log(out);
// => { translation: 'EN: 這是一段很好的文字。', sentiment: 'positive', summary: 'EN: 這是一段很好的文字。 (positive)' }

3. RunnableBranch：條件分支執行

RunnableBranch 讓你根據 輸入的某個屬性 或 計算結果，選擇不同的 Runnable 走向。相當於程式語言的 if...else，但在 workflow 中保持 純函式 與 可組合性。

3.1 基本用法

import { RunnableBranch, RunnableLambda } from "@langchain/core/runnables";

const englishHandler = new RunnableLambda(async (input) => `英語回應：${input}`);
const chineseHandler = new RunnableLambda(async (input) => `中文回應：${input}`);

// 判斷語言的 branch
const branch = RunnableBranch.from({
  // 判斷函式回傳布林值，決定走向
  condition: async (input) => {
    // 簡易偵測：是否包含中文字符
    return /[\u4e00-\u9fff]/.test(input);
  },
  // 若 condition 為 true，走中文處理
  true: chineseHandler,
  // 否則走英文處理
  false: englishHandler,
});

const out1 = await branch.invoke("Hello, how are you?");
const out2 = await branch.invoke("你好，今天過得如何？");
console.log(out1); // => 英語回應：Hello, how are you?
console.log(out2); // => 中文回應：你好，今天過得如何？

3.2 多條分支（switch 風格）

RunnableBranch 也支援 多條分支，透過 cases 物件與 default 處理未匹配的情況：

const faqHandler = new RunnableLambda(async (q) => `FAQ 回答：${q}`);
const smallTalkHandler = new RunnableLambda(async (q) => `閒聊回應：${q}`);
const fallbackHandler = new RunnableLambda(async (q) => `抱歉，我不太懂，請再說一次。`);

const multiBranch = RunnableBranch.from({
  // 依照問題類型切換
  cases: {
    // 判斷是否為 FAQ（關鍵字包含「什麼」或「如何」）
    faq: async (input) => /什麼|如何/.test(input),
    // 判斷是否為小聊天（含有「你」或「我」）
    smallTalk: async (input) => /你|我/.test(input),
  },
  // 各 case 對應的 Runnable
  handlers: {
    faq: faqHandler,
    smallTalk: smallTalkHandler,
  },
  // 若都不符合，走 fallback
  default: fallbackHandler,
});

console.log(await multiBranch.invoke("什麼是 LangChain？"));
// => FAQ 回答：什麼是 LangChain？

console.log(await multiBranch.invoke("你今天過得好嗎？"));
// => 閒聊回應：你今天過得好嗎？

console.log(await multiBranch.invoke("請幫我預訂機票。"));
// => 抱歉，我不太懂，請再說一次。

3.3 結合 RunnableSeq 與 RunnableBranch

在實務專案中，序列 與 分支 常常交錯使用。例如：

const preprocess = new RunnableLambda(async (msg) => msg.trim().toLowerCase());

const seq = RunnableSeq.from([
  preprocess,
  multiBranch, // 依內容分支
  new RunnableLambda(async (response) => `${response}（已完成）`),
]);

const final = await seq.invoke("  什麼是向量資料庫？  ");
console.log(final);
// => FAQ 回答：什麼是向量資料庫？（已完成）

常見陷阱與最佳實踐

陷阱 說明 解決方案 / 最佳實踐
1. 輸入/輸出型別不一致 RunnableSeq 會直接把前一步的輸出當作下一步的輸入，若型別不匹配會拋錯。 在每個 RunnableLambda 中加入 型別檢查 或使用 TypeScript 定義 Input / Output 泛型。
2. 條件函式過於耗時 RunnableBranch 的 condition 必須是 快速 判斷，否則會拖慢整條 pipeline。 把重量級運算（例如遠端 API）抽離到 前置步驟（例如 RunnableSeq 中的前置處理），在 condition 只做簡單判斷。
3. 分支結果不一致的結構 多條分支回傳的資料結構不同，後續步驟可能無法正確處理。 統一 每條分支的回傳結構（例如 {type: 'faq', payload: ...}），或在 RunnableSeq 後加入 標準化 步驟。
4. 無法追蹤錯誤來源 當 pipeline 中任一 Runnable 發生例外，堆疊資訊可能被包裝層遮蔽。 使用 try/catch 包住每個 Runnable，或在 RunnableSeq 前後插入 LoggingRunnable 以記錄 input / output。
5. 重複建立相同的 Runnable 每次呼叫 RunnableSeq.from([...]) 都會新建物件，造成不必要的資源浪費。 把 可重用的 Runnable 抽成常量，使用 單例（例如 const translate = new ChatOpenAI(...);）。

其他最佳實踐

  1. 保持純函式：讓每個 Runnable 只依賴傳入參數，避免全域變數或副作用。
  2. 使用 RunnableMap 取代手寫多輸出：RunnableMap 自動平行化子 Runnable，提升效能。
  3. 加入超時與重試機制：對於外部 API（LLM、Tool）可包裝成 RunnableRetry，減少因一次失敗導致整條 pipeline 中斷。
  4. 文件化每條分支的業務規則：在代碼註解或 README 中說明 condition 判斷的依據，方便日後維護。

實際應用場景

場景 為何適合使用 RunnableSeq + RunnableBranch 範例概念
客服聊天機器人 需要先 前處理 → 判斷問題類型（FAQ / 小聊 / 轉人工） → 對應處理。 前處理 → Branch（FAQ / SmallTalk / Transfer） → 各自的 LLM / API。
多語言文件翻譯工作流 先 偵測語言 → 依語言走 不同的翻譯模型 → 統一後處理。 Seq：Detect → Branch（中文、英文、其他） → 各自的 Translator → Seq：CleanUp。
資料管線（ETL） 抽取 → 根據資料類型分支清洗 → 合併 → 載入。 Seq：Extract → Branch（CSV / JSON / XML） → Cleaners → Seq：Merge → Load。
A/B 測試推薦系統 根據使用者屬性 分支至 不同的推薦模型，最後 統一回傳。 Seq：UserInfo → Branch（新手 / 老手） → ModelA / ModelB → Seq：FormatResult。
金融風險評估 先做基本檢查 → 根據風險等級分支至不同的審核流程 → 彙總結果。 Seq：Validate → Branch（Low / Medium / High） → 各自的審核工作流 → Seq：Aggregate。

總結

  • RunnableSeq 為 「串接」 提供了簡潔且可組合的方式，讓多個模型、工具或自訂函式在同一條 pipeline 中依序執行。
  • RunnableBranch 則讓 「條件分流」 成為可能，透過布林判斷或多條 cases，可以根據輸入內容或前置計算結果選擇不同的執行路徑。
  • 兩者結合使用，可構築出 高度模組化、易測試、易維護 的 AI 工作流，從簡單的聊天機器人到複雜的企業級 ETL 系統皆適用。

關鍵要點：保持每個 Runnable 的 純函式、型別一致、錯誤可追蹤，並善用 RunnableMap、RunnableRetry 等輔助工具，才能發揮 LangChain 的最大威力。

祝你在 LangChain 的世界裡玩得開心，打造出更聰明、更彈性的 AI 應用！ 🚀