Runnable接口

在 LCEL 中，几乎所有的模块：

• 提示词模板
• 模型
• 解析器

都是实现了 Runnable 接口的，可以将这些模块称之为 Runnable 类型。这种类型的模块可以快速插入到链条里面。

RunnableLambda

RunnableLambda 是 LangChain.js 提供的一种轻量级工具，它能把 普通函数 封装成符合 Runnable 接口规范的实例，从而让该函数能够无缝参与到 LCEL 的链式调用与流式处理流程中。

1. 快速上手案例

   import { RunnableLambda } from '@langchain/core/runnables'
   
   const fn = (text) => {
     return text.toUpperCase()
   }
   
   const runnableFn = RunnableLambda.from(fn)
   
   const res = await runnableFn.invoke('hello')
   
   console.log(res)

2. 关键词高亮插件

   闭包 —> *&*闭包*&*

   import { ChatOllama } from '@langchain/ollama'
   import { PromptTemplate } from '@langchain/core/prompts'
   import { StringOutputParser } from '@langchain/core/output_parsers'
   import { RunnableLambda } from '@langchain/core/runnables'
   
   // 1. 创建一个提示词模板
   const pt = PromptTemplate.fromTemplate('请使用中文解释下面的概念：{topic}')
   
   // 2. 创建模型
   const model = new ChatOllama({
     model: 'llama3',
     temperature: 0.7,
   })
   
   // 3. 解析器
   const parser = new StringOutputParser()
   
   // 4. 创建一个链条
   let chain = pt.pipe(model).pipe(parser)
   
   // 5. 创建一个简单的插件
   const fn = (text) => text.replace(/闭包/g, '*&*闭包*&*')
   const highlight = RunnableLambda.from(fn)
   
   chain = chain.pipe(highlight)
   
   const res = await chain.invoke({
     topic: '闭包',
   })
   
   console.log(res)

更多场景：

• 插入日志收集模块
• 插入翻译模块（调用外部 API）
• 插入开关：判断某个条件是否中断执行
• 格式调整、结构清洗、敏感词过滤

pipe() 方法可接受的三种类型：

1. Runnable 实例：这是最常用的，也是最推荐的

2. 普通函数：LCEL 内部会自动用 RunnableLambda.from(fn) 包装成一个 Runnable。

3. 对象：将上游的输入（或结果）分别传给多个 runnable，然后返回一个对象。

   import { ChatOllama } from '@langchain/ollama'
   import { PromptTemplate } from '@langchain/core/prompts'
   import { StringOutputParser } from '@langchain/core/output_parsers'
   import { RunnableLambda } from '@langchain/core/runnables'
   
   // 创建模型
   const model = new ChatOllama({
     model: 'llama3',
     temperature: 0.7,
   })
   
   // 解析器
   const parser = new StringOutputParser()
   
   // 创建两个子链
   const chain1 = PromptTemplate.fromTemplate(
     '请用中文用 2-3 句概括以下主题的核心含义：{input}'
   )
     .pipe(model)
     .pipe(parser)
   
   const chain2 = PromptTemplate.fromTemplate(
     '请用中文从以下主题中提取 5 个关键词，以逗号分隔：{input}'
   )
     .pipe(model)
     .pipe(parser)
   
   let chain = RunnableLambda.from((x) => x)
   
   chain = chain.pipe({
     summary: chain1,
     keywords: chain2,
   })
   
   const res = await chain.invoke({
     input: '闭包',
   })
   console.log(res)

RunnableMap

RunnableMap（也叫 RunnableParallel），它可以让多个链条 并发执行，并返回一个结构化的对象结果。

RunnableMap.from({ ... }) 会并发执行多个子链，并将它们的结果组合成一个对象返回。

import { ChatOllama } from '@langchain/ollama'
import { PromptTemplate } from '@langchain/core/prompts'
import { StringOutputParser } from '@langchain/core/output_parsers'
import { RunnableMap } from '@langchain/core/runnables'

// 创建模型
const model = new ChatOllama({
  model: 'llama3',
  temperature: 0.7,
})

// 解析器
const parser = new StringOutputParser()

const chain1 = PromptTemplate.fromTemplate('用中文讲一个关于 {topic} 的笑话')
  .pipe(model)
  .pipe(parser)
const chain2 = PromptTemplate.fromTemplate('用中文写一首关于 {topic} 的两行诗')
  .pipe(model)
  .pipe(parser)

const chain = RunnableMap.from({
  joke: chain1,
  poem: chain2,
})

const res = await chain.invoke({
  topic: '小狗',
})
console.log(res)

RunnableSequence

.pipe() 是逐个拼接，RunnableSequence.from([...]) 则是显式声明流程结构，将多个步骤写成数组更清晰。

课堂练习：快速上手示例

import { ChatOllama } from '@langchain/ollama'
import { PromptTemplate } from '@langchain/core/prompts'
import { StringOutputParser } from '@langchain/core/output_parsers'
import { RunnableLambda, RunnableSequence } from '@langchain/core/runnables'

// 1. 创建 Prompt 模板
const pt = PromptTemplate.fromTemplate('请使用中文解释以下概念：{topic}')

// 2. 创建本地模型
const model = new ChatOllama({ model: 'llama3', temperature: 0.7 })

// 3. 创建字符串解析器
const parser = new StringOutputParser()

// 4. 插件
const fn = (text) => {
  return text.replace(/闭包/g, '*&*闭包*&*')
}
const highlight = RunnableLambda.from(fn)

const chain = RunnableSequence.from([pt, model, parser, highlight])

const res = await chain.invoke({
  topic: '闭包',
})

console.log(res)

RunnablePassthrough

将输入原样传给输出，中间不做任何处理。

import { RunnablePassthrough } from '@langchain/core/runnables'

const passthrough = new RunnablePassthrough()

// 相当于输入什么，输出就是什么
const result = await passthrough.invoke('Hello, LangChain!')
console.log(result) // 输出：Hello, LangChain!

🤔 这有什么意义？

实例化 RunnablePassthrough 的时候，接收一个配置对象，该对象中可以配置 func 副作用函数，用于对输入做一些副作用处理，例如记录日志、写入数据库、做埋点等。

import { RunnablePassthrough } from '@langchain/core/runnables'

const logger = new RunnablePassthrough({
  // 一个副作用函数
  func: async (input) => {
    console.log('收到输入：', input)
    // 对输入做一些副作用处理，例如记录日志、写入数据库、做埋点等
  },
})

const res = await logger.invoke('LangChain')
console.log(res)

有时希望为传入的对象补充字段（如时间戳、用户信息、API 结果等），这时可以使用 .assign()。 它的作用是“给上下文添加字段”，就像在中间插入一步 Object.assign：

import { RunnablePassthrough } from '@langchain/core/runnables'

const injector = RunnablePassthrough.assign({
  timestamp: async () => new Date().toISOString(),
  meta: async () => ({
    region: 'us-east',
    requestId: 'req-456',
  }),
})

const result = await injector.invoke({ query: 'Vue 是什么？' })

console.log(result)
/*
  {
    query: 'Vue 是什么？',
    timestamp: '2025-08-11T08:07:35.358Z',
    meta: { region: 'us-east', requestId: 'req-456' }
  }
*/

可以用它来：

• 插入时间戳、用户 ID、请求 ID
• 注入工具结果：摘要、标签、翻译、情绪分析
• 为 Prompt 添加额外字段

课堂练习：

1. 打印用户输入日志
2. 注入 userId
3. 拼接 prompt
4. 调用本地模型
5. 解析返回结果