LangChain.js Runnable 全解

以前写 LLM 应用，逻辑是散装的：先 format 出 prompt，再 invoke 模型，最后 parse 输出——每一步都要手写、还要自己处理中间变量。 **LCEL（LangChain Expression Language，LangChain 表达式语言）**把这一切变成"搭积木"：所有组件都被统一成 Runnable，用 pipe 连成一条 Chain，一个 invoke 跑完全程。

本文基于 @langchain/core 的 runnables 模块，逐个拆解 9 个 Runnable 组合 API，并标注哪些已废弃、用什么替代。

为什么要学 Runnable / LCEL

先看最直观的对比。同样的"翻译 + 提取关键词"逻辑：

// 写法一：散装步骤（before.js）—— 每一步手写，中间变量到处飞
const formattedPrompt = await promptTemplate.format(input);
const response = await model.invoke(formattedPrompt);
const result = await outputParser.invoke(response);

// 写法二：LCEL 一条链（runable.js）—— 组装一次，随时调用
const chain = promptTemplate.pipe(model).pipe(outputParser);
const result = await chain.invoke(input);

学它的三个理由：

1. 统一接口：prompt、模型、解析器、普通函数……全都被包装成 Runnable，长一个样。
2. 组装即配置：想加步骤、换步骤、加条件、加并行，改的是"链的结构"，不是"执行逻辑"。
3. 调用方式白送：每一条链天然支持 invoke（同步）、stream（流式）、batch（批量）。

---

零、先认识 Runnable 统一接口

一句话说明：Runnable 是所有 LangChain 组件共用的"可执行单元"接口——不管底层是模板、模型还是函数，对外都长这样：

• invoke(input)：同步执行，返回最终结果。
• stream(input)：流式返回，边生成边吐（打字机效果）。
• batch(inputs[])：一次执行一批输入，内部尽量并行。
• pipe(next)：把自己和下一个 Runnable 接起来，返回一条新的链（等价于 RunnableSequence）。

// 任何能调用的组件，都是 Runnable
await promptTemplate.invoke({ text: '你好' });   // prompt 是 Runnable
await model.invoke('你好');                       // 模型是 Runnable
await outputParser.invoke('...');                 // 解析器也是 Runnable

// pipe 出链条
const chain = promptTemplate.pipe(model).pipe(outputParser);

要点：

• pipe() 是 LCEL 的精髓：a.pipe(b).pipe(c) 就是一条顺序执行的链。
• 链本身也是一个 Runnable，所以链上任意一个环节都能单独调用，调试超方便。
• 下面所有 API 都是用来"改变链的走法"的：顺序、分支、并行、循环、透传……

---

一、RunnableSequence：顺序执行的地基

一句话说明：把多个 Runnable 按顺序串起来，前一个的输出自动变成后一个的输入。它是 LCEL 最基础、也最常用的容器。

import { RunnableLambda, RunnableSequence } from '@langchain/core/runnables';

const addOne = RunnableLambda.from((input) => input + 1);
const multiplyTwo = RunnableLambda.from((input) => input * 2);

// 方式一：RunnableSequence.from([...])
const chain1 = RunnableSequence.from([addOne, multiplyTwo, addOne]);

// 方式二：.pipe() 连写（等价，更常用）
const chain2 = addOne.pipe(multiplyTwo).pipe(addOne);

console.log(await chain2.invoke(5)); // (5+1)*2+1 = 13

要点：

• 数组顺序就是执行顺序，上一步的返回值 = 下一步的参数。
• RunnableSequence.from([...]) 和 a.pipe(b).pipe(c) 是完全等价的两种写法，按口味选。

---

二、RunnableLambda：把任意函数变成 Runnable

一句话说明：把一个普通函数（同步/异步都行）包成 Runnable，这样自己写的逻辑也能接进 chain。

import { RunnableLambda, RunnableSequence } from '@langchain/core/runnables';

const formatText = RunnableLambda.from((input) => input.trim().toUpperCase());
const addTag = RunnableLambda.from(async (text) => `【译文】${text}`);

const chain = formatText.pipe(addTag);
console.log(await chain.invoke('  hello  ')); // 【译文】HELLO

要点：

• 用 RunnableLambda.from(函数) 包装，函数体里想干嘛都行，包括调外部 API、查数据库。
• 函数返回一个 Runnable 对象也能识别，能动态决定下一步（高级用法）。
• 这是"把业务代码接进 LCEL"的万能粘合剂。

---

三、RunnableMap / RunnableParallel：并行执行，结果合一

一句话说明：同时跑多条链，把各自的结果放进对象的不同属性里（互不依赖时并行，效率高）。

import { RunnableMap, RunnableParallel, RunnableLambda } from '@langchain/core/runnables';

const addOne = RunnableLambda.from((input) => input.num + 1);
const multiplyTwo = RunnableLambda.from((input) => input.num * 2);
const square = RunnableLambda.from((input) => input.num * input.num);

// RunnableMap / RunnableParallel 都可以用（RunnableParallel 是官方更推荐的名字）
const chain = RunnableParallel.from({
  add: addOne,
  multiply: multiplyTwo,
  square: square,
});

const result = await chain.invoke({ num: 5 });
console.log(result); // { add: 6, multiply: 10, square: 25 }

要点：

• 每个 key 对应一条链，输入同一个对象传进去，输出按 key 归位。
• RunnableMap 和 RunnableParallel 都能用——源码里 RunnableParallel 就是继承 RunnableMap 的子类，官方文档更推荐用 RunnableParallel。
• 各分支互不依赖时并行执行；依赖时请用顺序链。

---

四、RunnablePassthrough：保留原始输入

一句话说明：原样把输入透传下去，常用于"既要保留原始输入，又要算出派生结果"的场景。

import { RunnablePassthrough, RunnableLambda } from '@langchain/core/runnables';

// 方式一：配合 RunnableMap，让 original 原样保留
const chain = RunnableMap.from({
  original: new RunnablePassthrough(),
  upper: RunnableLambda.from((text) => text.toUpperCase()),
  length: RunnableLambda.from((text) => text.length),
});

console.log(await chain.invoke('langchain'));
// { original: 'langchain', upper: 'LANGCHAIN', length: 9 }

// 方式二（更推荐）：.assign() 在保留原输入的基础上追加字段
const chain = new RunnablePassthrough().assign({
  upper: (text) => text.toUpperCase(),
});

console.log(await chain.invoke('langchain'));
// { upper: 'LANGCHAIN' }  // 注意：单独调用时输入就是字符串本身

要点：

• new RunnablePassthrough() 就是"输入是什么，输出就是什么"。
• .assign({...}) 是它的黄金搭档：在原输入基础上追加/覆盖字段，是 RAG、Agent 里"边透传边加工"的标配写法。

---

五、RunnablePick：从输入对象里挑字段

一句话说明：从对象输入里挑出指定的几个属性返回，丢掉不需要的字段。

import { RunnablePick, RunnableSequence, RunnableLambda } from '@langchain/core/runnables';

const inputData = {
  name: 'Leo',
  age: 30,
  city: '北京',
  country: '中国',
  email: 'shenguang@example.com',
};

const chain = RunnableSequence.from([
  // 先生成一个派生字段
  RunnableLambda.from((input) => ({
    ...input,
    fullInfo: `${input.name}，${input.age}岁，来自${input.city}`,
  })),
  // 只挑出这两个字段返回
  new RunnablePick(['name', 'fullInfo']),
]);

console.log(await chain.invoke(inputData));
// { name: 'Leo', fullInfo: 'Leo，30岁，来自北京' }

要点：

• new RunnablePick(['a', 'b']) 从输入对象里挑字段；传单个字符串也行（new RunnablePick('name')）。
• 更简洁的等价写法：chain.pick(['a', 'b']) 直接作为链式方法用。
• 适合"链到后面，输入越来越精简、只留要用的"。

---

六、RunnableBranch：if / else 条件分支

一句话说明：按条件匹配，走对应的那条链，第一个命中的分支执行，都不命中走默认。

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

const isPositive = RunnableLambda.from((n) => n > 0);
const isZero = RunnableLambda.from((n) => n === 0);

const branch = RunnableBranch.from([
  [isPositive, RunnableLambda.from((n) => `${n} 是正数`)],
  [isZero, RunnableLambda.from((n) => `${n} 是零`)],
  RunnableLambda.from((n) => `${n} 是负数`), // 默认分支
]);

for (const n of [5, 0, -3]) {
  console.log(await branch.invoke(n)); // 5 是正数 / 0 是零 / -3 是负数
}

要点：

• 格式是数组套数组：[[条件Runnable, 执行Runnable], ...]，最后一项是默认分支（必填）。
• 条件 Runnable 返回 true/false；从上到下第一个命中即执行，顺序很关键。
• 这就是链里的 if / else if / else。

---

七、RouterRunnable：switch / case 路由

一句话说明：根据输入的 key，选出一条对应的链来执行，等价于 switch / case。

import { RouterRunnable, RunnableLambda } from '@langchain/core/runnables';

const toUpperCase = RunnableLambda.from((text) => text.toUpperCase());
const reverseText = RunnableLambda.from((text) => text.split('').reverse().join(''));

const router = new RouterRunnable({
  runnables: {
    toUpperCase,
    reverseText,
  },
});

console.log(await router.invoke({ key: 'reverseText', input: 'Hello World' })); // dlroW olleH
console.log(await router.invoke({ key: 'toUpperCase', input: 'Hello World' })); // HELLO WORLD

要点：

• 输入要包一层：{ key: 选哪个, input: 真正传给链的数据 }。
• 和 RunnableBranch 的区别：Branch 靠"条件判断"，Router 靠"key 精确匹配"。

---

八、RunnableEach：循环处理数组的每个元素

一句话说明：输入是一个数组，对每个元素分别跑一遍 chain，输出还是数组（数组版 map）。

import { RunnableEach, RunnableLambda, RunnableSequence } from '@langchain/core/runnables';

const processItem = RunnableSequence.from([
  RunnableLambda.from((name) => name.toUpperCase()),
  RunnableLambda.from((name) => `你好，${name}！`),
]);

const chain = new RunnableEach({ bound: processItem });

const result = await chain.invoke(['alice', 'bob', 'carol']);
console.log(result); // ['你好，ALICE！', '你好，BOB！', '你好，CAROL！']

要点：

• new RunnableEach({ bound: 单元素chain })，把"单元素处理逻辑"包一层就变成"批量处理逻辑"。
• 和 batch() 的区别：RunnableEach 是一条链对数组每个元素依次 map；batch() 是同一条链对多个输入并行跑（输入输出都是数组）。

---

九、RunnableWithMessageHistory：加记忆（⚠️ 已废弃）

一句话说明：它本来的作用是给 chain 自动管理多轮会话历史（把历史塞进 MessagesPlaceholder，结束后把新对话写回）。

但——在 @langchain/core 1.2.x 中它已被官方标记废弃：

@deprecated Use LangGraph's built-in persistence instead.

对应新方法：LangGraph 的内置持久化（Checkpointer）。 它通过"checkpointer + thread_id"自动保存和恢复对话状态，比手写历史 Map 更干净，还能断点续跑。

// 需要额外安装：pnpm add @langchain/langgraph
import { MemorySaver, MessagesAnnotation, StateGraph } from '@langchain/langgraph';
import { ChatOpenAI } from '@langchain/openai';
import { SystemMessage } from '@langchain/core/messages';

const model = new ChatOpenAI({ modelName: process.env.MODEL_NAME, apiKey: process.env.OPENAI_API_KEY });

// 1. MemorySaver 就是"存在内存里的会话存储"（短期记忆）
const checkpointer = new MemorySaver();

// 2. 用 LangGraph 画一条最简单的"模型节点"图
const graph = new StateGraph(MessagesAnnotation)
  .addNode('assistant', async (state) => ({
    messages: [await model.invoke([new SystemMessage('你是一个简洁、有帮助的中文助手。'), ...state.messages])],
  }))
  .addEdge('__start__', 'assistant')
  .addEdge('assistant', '__end__')
  .compile({ checkpointer });

// 3. 同一个 thread_id 就是同一个会话，历史自动带过去
const res1 = await graph.invoke(
  { messages: [{ role: 'user', content: '我的名字是Leo' }] },
  { configurable: { thread_id: 'user-123' } }
);
const res2 = await graph.invoke(
  { messages: [{ role: 'user', content: '我叫什么名字？' }] },
  { configurable: { thread_id: 'user-123' } }
); // 能答出来 → 记忆生效

要点：

• 别再在新代码里用 RunnableWithMessageHistory，虽然 1.x 里还能跑，但官方已明确不推荐。
• 替代方案核心就两个词：checkpointer（存状态）+ thread_id（认会话）。
• 本地学习用 MemorySaver；生产换数据库持久化（如 PostgresSaver），重启不丢。

---

十、三种调用方式：invoke / stream / batch

每条 Runnable（包括你组装的整条链）都自带三种调用方式：

const chain = promptTemplate.pipe(model).pipe(outputParser);

// 1. invoke：同步等最终结果
const result = await chain.invoke({ text: '你好' });

// 2. stream：流式返回，逐个 chunk 吐
for await (const chunk of await chain.stream({ text: '你好' })) {
  console.log(chunk); // 打字机效果
}

// 3. batch：一次跑一批输入
const results = await chain.batch([{ text: '你好' }, { text: '再见' }]);

要点：

• 三种方式只用改一个方法名，链本身不用动——这就是统一接口的好处。
• 废弃注意：
  • streamLog() 已废弃 → 用 .stream()。
  • streamEvents({ version: 'v1' }) 已废弃 → 用 version: 'v2' 或 .stream()。

---

一张图总结：怎么选

最后

LCEL 的核心心智就一句话：所有东西都是 Runnable，链是 Runnable 的组合，调用方式白送。先记住 RunnableSequence（顺序）、RunnableMap（并行）、RunnableBranch（分支）这三件套，几乎能应付 90% 的组装需求；遇到数组、透传、挑字段再分别加 RunnableEach、RunnablePassthrough、RunnablePick。

写新代码时记住两条：并行优先用 RunnableParallel 这个叫法；记忆优先用 LangGraph 的 checkpointer——别再被废弃的 RunnableWithMessageHistory 带跑了。

下一站建议：把 RunnableMap + RunnablePassthrough.assign() + RunnableBranch 组合起来，做一个"先并行检索 → 判断有没有结果 → 有就基于结果回答、没有就走兜底"的 RAG 链——这就是一个准生产级的问答管线。