LCEL 并行处理

要点

• LCEL 链路通常不是一条直线，一条输入可能需要同时走多个独立分支。
• RunnableParallel 把同一份输入同时交给多个节点，按 key 收集结果，但不保留原始输入。
• RunnablePassthrough.assign() 在保留原始输入的同时，并行补新字段，更适合接回 Agent。
• 在 Agent 应用里，最常见的模式是：先用 LCEL 前置链并行补字段，再把补好的结果交给 Agent 生成最终回复。

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

上一篇已经搭起 LCEL 最基础的链路：节点怎么接、.pipe() 怎么串、Runnable 为什么能统一调用。但在真实项目里，链路通常不会只是一条直线。

例如用户发来一句问题描述，程序可能需要同时做几件事：

• 判断问题类型。
• 提取关键词。
• 判断优先级。

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

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

• RunnableParallel
• RunnablePassthrough.assign()

它们的区别不在「谁更高级」，而在于：并行结果是不是已经够用了，后面的 Agent 还要不要继续读取原始输入。

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

假设已经有三条独立子链，分别负责问题分类、关键词提取、优先级判断。最直接的写法是 Promise.all：

const input = "支付接口返回 500，订单状态未知，部分用户无法完成支付。";

const [category, keywords, priority] = await Promise.all([
  categoryChain.invoke({ input }),
  keywordChain.invoke({ input }),
  priorityChain.invoke({ input }),
]);

const result = { category, keywords, priority };

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

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

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

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

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

import { RunnableParallel } from "@langchain/core/runnables";

const parallel = RunnableParallel.from({
  category: categoryChain,
  keywords: keywordChain,
  priority: priorityChain,
});

const result = await parallel.invoke({
  input: "支付接口返回 500，订单状态未知，部分用户无法完成支付。",
});

console.log(result);

返回值大致如下：

{
  "category": { "type": "bug", "confidence": 0.92 },
  "keywords": ["支付接口", "500", "订单状态"],
  "priority": { "level": "high" }
}

这里有三个关键点：

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

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

4. RunnableParallel 的问题：原始输入不见了

RunnableParallel 的输出只包含你定义的那些 key。也就是说，上面那段代码执行完以后，得到的是：

{
  "category": { ... },
  "keywords": [...],
  "priority": { ... }
}

原来的 input 已经不在里面了。这在普通链里不一定是问题，但一旦要接回 Agent，就很容易卡住。因为后面的 Agent 往往既想知道问题分类和优先级，也想继续看到用户原话。如果原始输入没了，后面还得手动透传回来，代码就会开始别扭。

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

这时候更常用的其实不是 RunnableParallel，而是 RunnablePassthrough.assign()。它的作用可以直接记成一句话：保留当前输入对象，再把新字段补上去。

import { RunnablePassthrough } from "@langchain/core/runnables";

const enriched = RunnablePassthrough.assign({
  category: categoryChain,
  keywords: keywordChain,
  priority: priorityChain,
});

const result = await enriched.invoke({
  input: "支付接口返回 500，订单状态未知，部分用户无法完成支付。",
});

console.log(result);

输出会变成：

{
  "input": "支付接口返回 500，订单状态未知，部分用户无法完成支付。",
  "category": { "type": "bug", "confidence": 0.92 },
  "keywords": ["支付接口", "500", "订单状态"],
  "priority": { "level": "high" }
}

这里最重要的区别是：

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

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

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

这篇最重要的不是「怎么并行」，而是「并行完以后怎么接回 Agent」。更典型的写法是：

1. 先用 LCEL 前置链并行补字段。
2. 再把补好的结果交给 Agent。

import { createAgent } from "langchain";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { JsonOutputParser } from "@langchain/core/output_parsers";
import { RunnablePassthrough } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "deepseek-chat",
  apiKey: process.env.MODEL_API_KEY,
  configuration: {
    baseURL: process.env.MODEL_BASE_URL ?? "https://api.deepseek.com/v1",
  },
});

// 注意：模板里的 JSON 花括号必须用双花括号转义，
// 否则 {type}、{confidence}、{level} 会被当成模板变量。
const categoryChain = ChatPromptTemplate.fromMessages([
  ["system", "分析问题类型，只返回 JSON：{{"type":"bug|feature|doc", "confidence":0}}'],
  ["user", "{input}"],
])
  .pipe(model)
  .pipe(new JsonOutputParser());

const keywordChain = ChatPromptTemplate.fromMessages([
  ["system", "提取这句话里的关键词，只返回 JSON 数组。"],
  ["user", "{input}"],
])
  .pipe(model)
  .pipe(new JsonOutputParser());

const priorityChain = ChatPromptTemplate.fromMessages([
  ["system", "判断优先级，只返回 JSON：{{"level":"low|medium|high"}}"],
  ["user", "{input}"],
])
  .pipe(model)
  .pipe(new JsonOutputParser());

// 前置链：保留原始输入，同时并行补 category / keywords / priority
const preProcess = RunnablePassthrough.assign({
  category: categoryChain,
  keywords: keywordChain,
  priority: priorityChain,
});

const agent = createAgent({
  model,
  tools: [],
  systemPrompt: [
    "你是一个技术问题排查助手。",
    "如果 priority=high，先给出临时止血方案，再给出根因排查建议。",
    "如果 priority=low，就正常分析问题并给出建议。",
  ].join("\n"),
});

const preProcessed = await preProcess.invoke({
  input: "支付接口返回 500，订单状态未知，部分用户无法完成支付。",
});

const result = await agent.invoke({
  messages: [
    {
      role: "user",
      content: [
        `input=${preProcessed.input}`,
        `category=${JSON.stringify(preProcessed.category)}`,
        `keywords=${JSON.stringify(preProcessed.keywords)}`,
        `priority=${JSON.stringify(preProcessed.priority)}`,
      ].join("\n"),
    },
  ],
});

console.log(result.messages.at(-1)?.text ?? "");

这段代码里有两个层次：

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

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

7. RunnableParallel 和 assign() 怎么选

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

用 RunnableParallel

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

用 assign()

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

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