AI 电子伴侣
创建时间: 2026-03-29最后更新: 2026-08-04

1. 一条输入，不一定只做一件事

上一篇已经把 LCEL 最基础的链路搭起来了：节点怎么接、.pipe() 怎么串、Runnable 为什么能统一调用。

但在真实项目里，链路通常不会只是一条直线。

拿 AI 伴侣来说，用户发来一句话以后，程序很可能要先同时做几件事：

  • 判断情绪
  • 提取关键词
  • 做一次风险检查

这三件事经常互不依赖。
如果一个接一个串行跑，总耗时就是三段相加；如果能并行跑，耗时就只看最慢的那一段。

这一篇要讲的就是 LCEL 里处理这类问题的两个工具：

  • RunnableParallel
  • RunnablePassthrough.assign()

它们的区别不在“谁更高级”，而在于：

  • 并行结果是不是已经够用了
  • 后面的 Agent 还要不要继续读取原始输入
正在加载图示...

2. 先看最容易写出来的并行代码

假设你已经有三条独立子链，分别负责：

  • 情绪判断
  • 关键词提取
  • 风险检查

最直接的写法，通常是 Promise.all：

parallel-manual.ts
1
const input = '今天线上刚修完，脑子还是绷着的。'

2


3
const [emotion, keywords, risk] = await Promise.all([

4
  emotionChain.invoke({ input }),

5
  keywordChain.invoke({ input }),

6
  riskChain.invoke({ input }),

7
])

8


9
const result = { emotion, keywords, risk }

这段代码当然能跑，但它有几个问题：

  • 并行结果要自己手动拼
  • 结果不容易继续接进 LCEL 链
  • 相同输入要重复传三次

一开始步骤少的时候还好，后面链路一长，就会越来越散。

3. RunnableParallel：同一份输入，同时喂给多个节点

RunnableParallel 做的事情很直接：
把同一份输入同时交给多个节点，再按你定义的 key 收集结果。

runnable-parallel.ts
01
import { RunnableParallel } from '@langchain/core/runnables'

02


03
const parallel = RunnableParallel.from({

04
  emotion: emotionChain,

05
  keywords: keywordChain,

06
  risk: riskChain,

07
})

08


09
const result = await parallel.invoke({

10
  input: '今天线上刚修完，脑子还是绷着的。',

11
})

12


13
console.log(result)

返回值会长成这样：

parallel-result.json
1
{

2
  "emotion": { "emotion": "anxious", "confidence": 0.88 },

3
  "keywords": ["线上", "修复", "紧张"],

4
  "risk": { "level": "medium" }

5
}

这里有三个关键点：

  1. 输入只传一次
  2. 子链同时执行
  3. 输出按 key 自动组装

如果后面的节点只需要这几个并行结果，不需要用户原话，那 RunnableParallel 已经够用了。

4. 一个很快会碰到的问题：原始输入不见了

RunnableParallel 的输出只包含你定义的那些 key。

也就是说，上面那段代码执行完以后，得到的是：

parallel-output.json
1
{

2
  "emotion": { ... },

3
  "keywords": [...],

4
  "risk": { ... }

5
}

原来的 input 已经不在里面了。

这在普通链里不一定是问题，但一旦要接回 Agent，就很容易卡住。
因为后面的 Agent 往往既想知道：

  • 用户情绪是什么
  • 风险高不高

也想继续看到用户原话。

如果原始输入没了，后面还得自己手动透传回来，代码就会开始别扭。

5. assign()：保留原始输入，再并行补字段

这时候更常用的其实不是 RunnableParallel，而是 RunnablePassthrough.assign()。

它的作用可以直接记成一句话：

保留当前输入对象，再把新字段补上去。

assign-basic.ts
01
import { RunnablePassthrough } from '@langchain/core/runnables'

02


03
const enriched = RunnablePassthrough.assign({

04
  emotion: emotionChain,

05
  keywords: keywordChain,

06
  risk: riskChain,

07
})

08


09
const result = await enriched.invoke({

10
  input: '今天开会被否了三次，我有点乱。',

11
})

12


13
console.log(result)

输出会变成：

assign-result.json
1
{

2
  "input": "今天开会被否了三次，我有点乱。",

3
  "emotion": { "emotion": "sad", "confidence": 0.79 },

4
  "keywords": ["开会", "被否定", "混乱"],

5
  "risk": { "level": "low" }

6
}

这里最重要的区别是：

  • input 还在
  • 并行结果是追加进去的

所以如果后面还要继续接 Prompt、Agent、结构化输出链，assign() 往往更顺。

6. 把它接回 Agent：前置并行链

这篇最重要的不是“怎么并行”，而是“并行完以后怎么接回 Agent”。

更典型的写法是：

  1. 先用 LCEL 前置链并行补字段
  2. 再把补好的结果交给 Agent
agent-parallel-prefilter.ts
01
import { createAgent } from 'langchain'

02
import { ChatPromptTemplate } from '@langchain/core/prompts'

03
import { JsonOutputParser } from '@langchain/core/output_parsers'

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

05
import { ChatOpenAI } from '@langchain/openai'

06


07
const model = new ChatOpenAI({

08
  model: 'deepseek-chat',

09
  apiKey: process.env.DEEPSEEK_API_KEY,

10
  configuration: {

11
    baseURL: process.env.DEEPSEEK_BASE_URL ?? 'https://api.deepseek.com/v1',

12
  },

13
})

14


15
// 注意：模板里的 JSON 花括号必须用双花括号转义

16
// 否则 {emotion}、{confidence}、{level} 会被当成模板变量

17
const emotionChain = ChatPromptTemplate.fromMessages([

18
  ['system', '分析用户情绪，只返回 JSON：{{"emotion":"", "confidence":0}}'],

19
  ['user', '{input}'],

20
])

21
  .pipe(model)

22
  .pipe(new JsonOutputParser())

23


24
const keywordChain = ChatPromptTemplate.fromMessages([

25
  ['system', '提取这句话里的关键词，只返回 JSON 数组。'],

26
  ['user', '{input}'],

27
])

28
  .pipe(model)

29
  .pipe(new JsonOutputParser())

30


31
const riskChain = ChatPromptTemplate.fromMessages([

32
  ['system', '判断这句话是否需要额外安抚，只返回 JSON：{{"level":"low|medium|high"}}'],

33
  ['user', '{input}'],

34
])

35
  .pipe(model)

36
  .pipe(new JsonOutputParser())

37


38
// 前置链：保留原始输入，同时并行补 emotion / keywords / risk

39
const preProcess = RunnablePassthrough.assign({

40
  emotion: emotionChain,

41
  keywords: keywordChain,

42
  risk: riskChain,

43
})

44


45
const agent = createAgent({

46
  model,

47
  tools: [],

48
  systemPrompt: [

49
    '你是一个前端陪伴助手。',

50
    '如果 risk=high，先安抚，再给一个很小的动作建议。',

51
    '如果 risk=low，就正常交流，不要过度放大情绪。',

52
  ].join('\n'),

53
})

54


55
const preProcessed = await preProcess.invoke({

56
  input: '今天线上刚修完，脑子还是绷着的。',

57
})

58


59
const result = await agent.invoke({

60
  messages: [

61
    {

62
      role: 'user',

63
      content: [

64
        `input=${preProcessed.input}`,

65
        `emotion=${JSON.stringify(preProcessed.emotion)}`,

66
        `keywords=${JSON.stringify(preProcessed.keywords)}`,

67
        `risk=${JSON.stringify(preProcessed.risk)}`,

68
      ].join('\n'),

69
    },

70
  ],

71
})

72


73
console.log(result.messages.at(-1)?.text ?? '')

这段代码里有两个层次：

  • preProcess 是 LCEL 前置链
  • agent 是最终回复入口

前置链负责把上下文补全，Agent 负责拿这些上下文生成最终回复。
这就是 assign() 在 Agent 场景里最常见的位置。

7. RunnableParallel 和 assign() 怎么选

可以直接按这个标准判断：

用 RunnableParallel

  • 并行结果本身就是最终结果
  • 后面不需要原始输入
  • 你只想收一个并行结果对象

用 assign()

  • 后面还要继续接 Prompt 或 Agent
  • 原始输入不能丢
  • 你想在原对象上逐步补字段

在单个 Agent 应用里，assign() 往往更常见。
因为 Agent 几乎总还要继续读取用户原话。