pipe三次不是炫技，是少写屎山—Agent开发第4课

读完本节你会分清：langchain的简单链、串行链、并行链——各烧几次模型、各适合什么活

LangChain既然这么命名，那用法里肯定有chain，chain翻译过来是链条的意思。

这里的 chain 不是抽象名词，而是 .pipe() 串起来后得到的 Runnable，一次 invoke 跑完全程。

一、简单链

示例，如果一个链有x、y、z三个节点我们就可以用pipe把这些节点串联起来

const chain = x.pipe(y).pipe(z)await chain.invoke('……')

​

这个怎么应用呢？

有这样一个场景，需要大模型给一个产品起三个中文名 ,实现流程如下

原料 { product: '便携咖啡机' }

工序1：拼好 prompt 字符串

工序2：发给模型，拿回 AIMessage

工序3：从 AIMessage 里抽出 content

成品 “….”

就有了下面这示例代码

import { ChatDeepSeek } from '@langchain/deepseek'

import { PromptTemplate } from '@langchain/core/prompts'

import { StringOutputParser } from '@langchain/core/output_parsers'

import 'dotenv/config'

​

const model = new ChatDeepSeek({ model: 'deepseek-chat' })

​

const namePrompt = PromptTemplate.fromTemplate(

'给{product}起 3 个中文产品名，每个不超过 6 个字，用顿号分隔'

)

​

const pipeline = namePrompt.pipe(model).pipe(new StringOutputParser())

​

const text = await pipeline.invoke({ product: '便携咖啡机' })

console.log(text)

执行上述代码 结果如下

上述代码里这个StringOutputParser再讲解下，

如果不加 StringOutputParser，llm 返回的是 AIMessage 对象，要拿答案得自己写 response.content。

StringOutputParser 的作用就是：自动从 AIMessage 里取出 .content，转成 string，方便 console.log 或传给下一步 chain。

流程简化如下

{ product: '便携咖啡机' }

namePrompt

"给便携咖啡机起 3 个中文产品名..."

model

AIMessage { content: "随行萃，掌上咖，一刻鲜" }

StringOutputParser

"随行萃，掌上咖，一刻鲜"   纯字符串

ok，有了上面这个简单基础，接下来我们继续讲串行任务跟并行任务

二、串行任务

字面意思，串行任务就是任务x、y后还能加z加a、b、c等等

继续用上面给产品起名的例子 ，我们接着给产品名追加评价

import { ChatDeepSeek } from '@langchain/deepseek'

import { ChatPromptTemplate, PromptTemplate } from '@langchain/core/prompts'

import { StringOutputParser } from '@langchain/core/output_parsers'

//串行链多了 RunnableLambda，记得从 @langchain/core/runnables 引入，否则 new RunnableLambda 会报 not defined。

import { RunnableLambda ，RunnableMap} from '@langchain/core/runnables'

import 'dotenv/config'

​

const llm = new ChatDeepSeek({ model: 'deepseek-chat' })

​

const namePrompt = PromptTemplate.fromTemplate(

'给{product}起 3 个中文产品名，每个不超过 6 个字，用顿号分隔'

)

​

const chain = namePrompt.pipe(llm).pipe(new StringOutputParser())

​

const reviewPrompt = ChatPromptTemplate.fromTemplate(

'下面这组产品名好不好记、有没有辨识度？请简要点评：{names}'

)

// 第 1 步：先起名

const composedChain = new RunnableLambda({

func: async (input) => {

const result = await chain.invoke(input)

return { names: result }

},

})

// 第 2 步：拼点评 prompt

.pipe(reviewPrompt)

// 第 3 步：模型点评

.pipe(llm)

// 第 4 步：取纯文本

.pipe(new StringOutputParser())

​

const res = await composedChain.invoke({ product: '便携咖啡机' })

console.log(res)

前面 namePrompt llm StringOutputParser 这条“起名链”和简单链一样。

真正变成串行链的，是从 RunnableLambda 开始：先跑完起名链，再 pipe 点评 prompt 模型 StringOutputParser。

第1步里又做了两件事

1）调用前面的 chain

const result = await chain.invoke(input)

输入 input：{ product: '便携咖啡机' }

chain 是上面的「起名链」：namePrompt llm StringOutputParser

输出 result：字符串，比如 "随行萃，掌上咖，一刻鲜"

2）把1）里结果换个包装，继续往下传

return { names: result }

为什么要这步？因为后面的

const reviewPrompt = ChatPromptTemplate.fromTemplate(

'下面这组产品名好不好记、有没有辨识度？请简要点评：{names}'

)

要的是{ names: "..." }，不是纯字符串，格式不对，就会出错

第2步 .pipe(reviewPrompt)

这1步只拼字符串，不调模型。

收到 { names: "随行萃，掌上咖，一刻鲜" } 后，填入 {names}，变成完整 prompt：下面这组产品名好不好记、有没有辨识度？请简要点评：随行萃，掌上咖，一刻鲜

第3步：.pipe(llm)

数据都处理完毕，传给大模型并生成答案

第4步：.pipe(newStringOutputParser())

之前说过从 AIMessage 里取出 .content，变成纯文本

流程如下

invoke({ product: '便携咖啡机' })

RunnableLambda

chain.invoke "随行萃，掌上咖，一刻鲜"

return { names: "随行萃，掌上咖，一刻鲜" }

reviewPrompt

"下面这组产品名... 随行萃，掌上咖，一刻鲜"

llm

AIMessage { content: "这组名字辨识度很高..." }

StringOutputParser

"这组名字辨识度很高..." res

最后执行结果如下

有串行任务 自然就有并行任务

三、并行任务

第一部分 两个chain

//第1个chain

const sloganChain = PromptTemplate.fromTemplate(

'为 {concept} 写一句 12 字以内的中文 Slogan'

)

.pipe(llm)

.pipe(new StringOutputParser())

​

//第2个chain

const overviewChain = PromptTemplate.fromTemplate(

'用 40 字概括 {concept} 的核心价值'

)

.pipe(llm)

.pipe(new StringOutputParser())

第二部分 通过 RunnableMap.from 将两个chain并联

const mapChain = RunnableMap.from({

slogan: sloganChain,

overview: overviewChain,

})

​

const result = await mapChain.invoke({ concept: '共享单车' })

一次 invoke({ concept: '共享单车' })，同一份输入会同时传给 sloganChain 和 overviewChain，两条链并行执行，最后拼成一个对象：

{ concept: '共享单车' }

├─ sloganChain   result.slogan

└─ overviewChain result.overview

补充两点，避免误解：

RunnableMap 的「并行」= 同一份输入同时走多条链，不是 for 循环排队。

本例 sloganChain 和 overviewChain 各调一次大模型，一般等于两次 API 请求，费用和耗时会高于只问一次。

完整代码如下

import { ChatDeepSeek } from '@langchain/deepseek'

import { PromptTemplate } from '@langchain/core/prompts'

import { StringOutputParser } from '@langchain/core/output_parsers'

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

import 'dotenv/config'

​

const llm = new ChatDeepSeek({ model: 'deepseek-chat' })

​

const sloganChain = PromptTemplate.fromTemplate(

'为 {concept} 写一句 12 字以内的中文 Slogan'

)

.pipe(llm)

.pipe(new StringOutputParser())

​

const overviewChain = PromptTemplate.fromTemplate(

'用 40 字概括 {concept} 的核心价值'

)

.pipe(llm)

.pipe(new StringOutputParser())

​

const mapChain = RunnableMap.from({

slogan: sloganChain,

overview: overviewChain,

})

​

const result = await mapChain.invoke({ concept: '共享单车' })

console.log(result)

执行效果如下：

本节小结

Chain 解决的是怎么把步骤串起来；下一章 Agent 则是在此之上加工具调用和循环决策。