本文 AI 產出，尚未審核

LangChain 教學：Runnable – 高度模組化流程的自訂實作

簡介

在使用 LangChain 建構聊天機器人或資訊擷取系統時，最常碰到的挑戰是 流程的可組合性 與 重複使用性。
傳統的程式碼往往把所有步驟硬寫在一起，導致 維護成本高、測試困難。
Runnable 介面正是為了解決這個問題而設計的：它把每一個處理階段抽象成 可呼叫、可串接 的單元，讓開發者可以像搭積木一樣自由組合。

本單元將深入探討 如何自訂 Runnable，從基礎概念到實作細節，並提供多個實用範例，協助你在 LangChainJS（或 TypeScript）環境中快速建立可重用、易測試的工作流程。

核心概念

1. Runnable 是什麼？

Runnable<I, O> 是一個 泛型介面，定義了一個接受型別 I、回傳型別 O 的函式： 

interface Runnable<I, O> {
  invoke(input: I, config?: RunnableConfig): Promise<O>;
}
  • I：輸入資料的型別（例如文字、向量、物件）。
  • O：輸出資料的型別（可以是同一型別，也可以是全新結構）。
  • invoke：執行邏輯的唯一入口，支援 非同步，因此可以與外部 API、LLM、資料庫等任務無縫整合。

重點：只要實作 invoke 方法，就能把任何程式碼包裝成 Runnable，進而使用 LangChain 提供的 pipe, batch, map 等高階工具。

2. 為什麼要自訂 Runnable？

  • 模組化：將複雜流程拆解成小單元，單元間不產生副作用。
  • 可測試：每個 Runnable 都可以獨立寫單元測試，提升可靠度。
  • 重用：相同的資料前處理、後處理或驗證邏輯，只要寫一次即可在多個鏈中重用。
  • 彈性：透過 RunnableConfig 可以在執行時注入額外參數（如溯源、日誌、超時設定），不需要改變程式碼本身。

3. 基本實作步驟

  1. 定義型別：明確宣告 I 與 O，讓 TypeScript 幫你檢查。
  2. 實作 invoke：將核心邏輯寫在此，盡量保持純函式（pure function）特性。
  3. 加入錯誤處理：使用 try / catch 包住非同步呼叫，並在錯誤時拋出有意義的訊息。
  4. 匯出並組合：使用 new MyRunnable() 建立實例，之後可透過 .pipe()、.map() 等方式串接。

程式碼範例

以下示範 5 個常見的自訂 Runnable 實作，涵蓋文字前處理、LLM 呼叫、向量搜尋、結果後處理與批次執行。所有範例均使用 TypeScript（可直接改成 JavaScript），並以 LangChainJS 為基礎。

範例 1️⃣：文字清理 Runnable

// cleanTextRunnable.ts
import { Runnable } from "@langchain/core/runnables";

export interface CleanTextInput {
  raw: string;
}
export interface CleanTextOutput {
  cleaned: string;
}

/**
 * 移除多餘空白、HTML 標籤與特殊符號的前處理器
 */
export class CleanTextRunnable implements Runnable<CleanTextInput, CleanTextOutput> {
  async invoke(
    input: CleanTextInput,
    _config?: any
  ): Promise<CleanTextOutput> {
    const { raw } = input;
    // 1. 去除 HTML 標籤
    let cleaned = raw.replace(/<[^>]*>/g, " ");
    // 2. 替換多個空白為單一空格
    cleaned = cleaned.replace(/\s+/g, " ").trim();
    // 3. 移除非字母數字符號（保留中文、日文等 Unicode）
    cleaned = cleaned.replace(/[^\p{L}\p{N}\s]/gu, "");
    return { cleaned };
  }
}

使用方式 

const cleaner = new CleanTextRunnable();
const result = await cleaner.invoke({ raw: "<p>Hello   World! 🚀</p>" });
console.log(result.cleaned); // "Hello World"

範例 2️⃣：呼叫 LLM 的 Runnable（OpenAI）

// openaiChatRunnable.ts
import { Runnable } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";

export interface ChatInput {
  messages: Array<{ role: "user" | "assistant" | "system"; content: string }>;
}
export interface ChatOutput {
  response: string;
}

/**
 * 包裝 ChatOpenAI，讓其符合 Runnable 介面
 */
export class OpenAIChatRunnable implements Runnable<ChatInput, ChatOutput> {
  private readonly model: ChatOpenAI;

  constructor(modelName = "gpt-4o-mini", temperature = 0) {
    this.model = new ChatOpenAI({ modelName, temperature });
  }

  async invoke(input: ChatInput, _config?: any): Promise<ChatOutput> {
    const { messages } = input;
    const response = await this.model.invoke(messages);
    // `response` 為 ChatMessage，取出文字內容
    return { response: response.content };
  }
}

串接示例 

const cleaner = new CleanTextRunnable();
const chat = new OpenAIChatRunnable();

// 先清理，再送給 LLM
const pipeline = cleaner.pipe(chat);
const result = await pipeline.invoke({ raw: "<h1>說明台北天氣</h1>" });
console.log(result.response);

範例 3️⃣：向量資料庫搜尋 Runnable（FAISS）

// faissSearchRunnable.ts
import { Runnable } from "@langchain/core/runnables";
import { FAISS } from "@langchain/community/vectorstores/faiss";
import { OpenAIEmbeddings } from "@langchain/openai";

export interface SearchInput {
  query: string;
  topK?: number;
}
export interface SearchOutput {
  docs: Array<{ pageContent: string; metadata: any }>;
}

/**
 * 使用 FAISS 向量庫進行相似度搜尋
 */
export class FAISSSearchRunnable implements Runnable<SearchInput, SearchOutput> {
  private readonly vectorStore: FAISS;

  constructor() {
    const embeddings = new OpenAIEmbeddings();
    // 假設已經有預先建立好的 FAISS 索引檔案
    this.vectorStore = FAISS.loadLocal("faiss-index", embeddings);
  }

  async invoke(input: SearchInput, _config?: any): Promise<SearchOutput> {
    const { query, topK = 4 } = input;
    const docs = await this.vectorStore.similaritySearch(query, topK);
    return { docs };
  }
}

搭配前置 Runnable 

const cleaner = new CleanTextRunnable();
const search = new FAISSSearchRunnable();
const pipeline = cleaner.pipe(search);
const result = await pipeline.invoke({ raw: "台北最近的展覽有哪些？" });
console.log(result.docs.map(d => d.pageContent));

範例 4️⃣：結果後處理 Runnable（摘要）

// summarizeRunnable.ts
import { Runnable } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";

export interface SummarizeInput {
  docs: Array<{ pageContent: string; metadata: any }>;
}
export interface SummarizeOutput {
  summary: string;
}

/**
 * 使用 LLM 產生多篇文件的摘要
 */
export class SummarizeRunnable implements Runnable<SummarizeInput, SummarizeOutput> {
  private readonly model: ChatOpenAI;

  constructor() {
    this.model = new ChatOpenAI({ modelName: "gpt-4o-mini", temperature: 0 });
  }

  async invoke(input: SummarizeInput, _config?: any): Promise<SummarizeOutput> {
    const combined = input.docs.map(d => d.pageContent).join("\n---\n");
    const prompt = [
      { role: "system", content: "請為以下內容寫一段 150 字以內的中文摘要。" },
      { role: "user", content: combined },
    ];
    const response = await this.model.invoke(prompt);
    return { summary: response.content };
  }
}

完整管線範例 

const pipeline = cleaner
  .pipe(search)          // 前處理 → 向量搜尋
  .pipe(new SummarizeRunnable()); // 搜尋結果 → 摘要

const result = await pipeline.invoke({ raw: "台北的美食推薦" });
console.log(result.summary);

範例 5️⃣：批次 (Batch) Runnable – 同時處理多筆查詢

// batchSearchRunnable.ts
import { Runnable, RunnableConfig } from "@langchain/core/runnables";
import { FAISSSearchRunnable } from "./faissSearchRunnable";

export interface BatchInput {
  queries: string[];
}
export interface BatchOutput {
  results: Array<{ query: string; docs: any[] }>;
}

/**
 * 內部使用多個 FAISSSearchRunnable 以平行方式執行
 */
export class BatchSearchRunnable implements Runnable<BatchInput, BatchOutput> {
  private readonly searchRunnable: FAISSSearchRunnable;

  constructor() {
    this.searchRunnable = new FAISSSearchRunnable();
  }

  async invoke(
    input: BatchInput,
    config?: RunnableConfig
  ): Promise<BatchOutput> {
    const promises = input.queries.map(q =>
      this.searchRunnable.invoke({ query: q }, config)
    );
    const results = await Promise.all(promises);
    return {
      results: input.queries.map((q, idx) => ({
        query: q,
        docs: results[idx].docs,
      })),
    };
  }
}

呼叫方式 

const batch = new BatchSearchRunnable();
const out = await batch.invoke({ queries: ["台北景點", "台北美食"] });
console.log(out.results);

常見陷阱與最佳實踐

陷阱 說明 解決方案
忘記回傳 Promise invoke 必須是 非同步，若直接回傳同步值會破壞管線的錯誤傳遞機制。 確保所有分支都有 async，即使內部是同步運算也使用 Promise.resolve() 包裝。
副作用 (Side‑effects) 在 Runnable 內部直接修改全域變數或外部物件，會導致測試不穩定。 盡量保持 純函式，所有需要的資料都以參數傳入，結果以回傳值輸出。
錯誤未被捕捉 直接拋出錯誤會讓整條 pipeline 中斷，無法提供友善回應。 使用 try / catch 包住外部 API 呼叫，並在錯誤物件中加入 metadata（如 errorCode），方便上層處理。
型別不匹配 pipe 時前後 Runnable 的輸入/輸出型別不對，編譯時會報錯。 利用 TypeScript 的泛型檢查，或在開發階段寫 單元測試 確認每個 Runnable 的 invoke 簽名。
過度細粒度 把每個字元的處理都拆成一個 Runnable，會產生大量的物件與上下文切換，影響效能。 適度聚合：將相關的操作（如文字清理 + 分詞）放在同一個 Runnable 中，除非真的需要獨立測試。

最佳實踐總結

  1. 單一職責：每個 Runnable 只做一件事。
  2. 明確型別：使用 TypeScript 定義 I、O，讓編譯器幫你抓錯。
  3. 錯誤標準化：拋出 LangChainError 或自行定義錯誤類別，統一在上層捕捉。
  4. 日誌與溯源：利用 RunnableConfig 的 metadata 欄位傳入 requestId、traceId，方便除錯與觀測。
  5. 可測試：為每個 Runnable 撰寫 單元測試（Jest / Vitest），確保邏輯正確且不受外部服務波動影響。

實際應用場景

場景 為何適合使用自訂 Runnable 範例流程
客服聊天機器人 前處理（清理、情感分析） → LLM 回答 → 後處理（敏感詞過濾） CleanText → SentimentRunnable → OpenAIChat → FilterRunnable
企業文件檢索 多階段：文件切片 → 向量化 → 相似度搜尋 → 摘要 → 排序 Splitter → EmbeddingRunnable → FAISSSearch → SummarizeRunnable → RankRunnable
多語言翻譯管線 文字正規化 → 語言偵測 → LLM 翻譯 → 格式化輸出 NormalizeRunnable → DetectLangRunnable → TranslateRunnable → FormatRunnable
資料標註自動化 讀取原始資料 → 生成標註提示 → LLM 產生標註 → 檢查一致性 LoadDataRunnable → PromptBuilder → OpenAIChat → ValidationRunnable
批次報告產出 同時處理多筆查詢 → 每筆產出摘要 → 合併成 PDF BatchSearchRunnable → SummarizeRunnable → MergePDFRunnable

重點：只要把業務邏輯抽成 Runnable，就能利用 LangChain 的 pipe, batch, map, reduce 等工具，快速拼湊出符合需求的端到端流程，且每個環節都可以獨立測試與重用。

總結

  • Runnable 為 LangChain 提供的核心抽象，讓開發者能以 模組化、可組合 的方式構建 LLM 應用。
  • 自訂 Runnable 只需要實作 invoke 方法，配合 TypeScript 的型別系統，即可確保資料流的正確性。
  • 透過 前處理、LLM 呼叫、搜尋、後處理 等範例，我們示範了如何把真實業務需求映射成可串接的 Runnable 鏈。
  • 注意 副作用、錯誤處理、型別匹配 等常見陷阱，遵循 單一職責、明確型別、可測試 的最佳實踐，能讓系統更穩定、維護成本更低。
  • 最後，將這些 Runnable 應用於客服、文件檢索、翻譯、標註、批次報告等場景，就能在 短時間內打造出可擴充、易除錯 的 AI 服務。

祝你在 LangChain 的世界裡玩得開心，寫出乾淨、可重用的 AI 工作流！