LCEL 分支路由

要点

• 上一篇讲的是并行：同一份输入同时做几件事。这一篇讲分支：不同输入走不同链路。
• RunnableBranch 的结构很像 if / else if / else，按条件从上到下匹配，命中即进入对应链。
• 实际项目里最常见的做法不是直接靠关键词判断，而是先让模型做一次轻量分类，再根据分类结果路由。
• 在 Agent 应用里，RunnableBranch 通常放在 Agent 前面做前置路由，而不是替代 Agent。

1. 背景：不是所有输入都该走同一条链

上一篇讲的是并行：同一份输入可以同时做几件事。这一篇刚好相反，要处理的是另一类很常见的问题：不同输入，要走不同的链。

以技术问题处理为例，用户发来一句话以后，程序通常要先判断它属于哪一类：

• 技术问题（需要详细排查步骤）。
• 功能需求（需要产品评估）。
• 文档问题（需要补充说明）。

这三类输入如果都走同一个 Prompt，结果往往会很别扭：技术问题会被回得太泛泛，功能需求会被回得像 FAQ，文档问题又可能被回得太技术化。

这就是 RunnableBranch 最适合出现的地方。它负责做路由：先判断输入属于哪一类，再把它送进对应的处理链。

2. 为什么不直接写 if/else

最直接的写法通常是这样：

async function handleMessage(input: string) {
  if (
    input.includes("报错") ||
    input.includes("502") ||
    input.includes("超时")
  ) {
    return techChain.invoke({ input });
  }

  if (input.includes("希望") || input.includes("能不能")) {
    return featureChain.invoke({ input });
  }

  return docChain.invoke({ input });
}

代码能跑，但有三个问题：

• 路由逻辑跑到了链外。
• 后面不容易继续接 .pipe()、assign()、fallback。
• 调用入口变成了一个普通函数，不再是 Runnable。

如果你前几篇已经把 LCEL 当成一条稳定管线在用，这里再突然切回 if/else，整条链就断开了。

3. RunnableBranch 的结构

RunnableBranch 的结构很像 if / else if / else。前面是一组 [条件, 处理链]，最后放一个兜底链：

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

const routeByType = RunnableBranch.from([
  [({ type }: { type: string }) => type === "bug", techChain],
  [({ type }: { type: string }) => type === "feature", featureChain],
  docChain,
]);

执行顺序也和 if / else if / else 一样：

1. 从上到下检查条件。
2. 命中第一个 true 就立刻进入对应链。
3. 后面的条件不再继续检查。
4. 全都不满足时，走最后一个兜底链。

顺序很重要。如果两条条件都有可能命中，写在前面的那条会优先拿到机会。

4. 最常见的写法：先分类，再路由

在实际项目里，最常见的不是直接靠关键词判断，而是先让模型做一次轻量分类，再根据分类结果路由。这套写法比较顺，因为它和前一篇的 assign() 能自然接起来。

import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import { RunnableBranch, 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",
  },
});

const classifyChain = ChatPromptTemplate.fromMessages([
  [
    "system",
    [
      "判断用户消息类型，只输出以下三个类别之一：",
      "- bug",
      "- feature",
      "- doc",
    ].join("\n"),
  ],
  ["user", "{input}"],
])
  .pipe(model)
  .pipe(new StringOutputParser());

const routeByType = RunnableBranch.from([
  [({ type }: { type: string }) => type.trim() === "bug", techChain],
  [({ type }: { type: string }) => type.trim() === "feature", featureChain],
  docChain,
]);

const chain = RunnablePassthrough.assign({ type: classifyChain }).pipe(
  routeByType,
);

这条链可以直接按步骤理解：

1. 输入先进来。
2. assign() 先补一个 type。
3. RunnableBranch 再根据 type 选链。

这比纯关键词匹配稳很多，因为分类不是在匹配几个单词，而是在理解整句话的大意。

5. 把它接回 Agent：前置路由链

这篇真正要讲的重点，不是「怎么分支」，而是「怎么把分支接回 Agent」。更典型的结构是：

• LCEL 先做类型分类和路由。
• 不同分支负责补自己的上下文。
• 最后再把整理好的结果交给 Agent。

import { createAgent } from "langchain";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import { RunnableBranch, 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",
  },
});

// 先用一个轻量分类链，判断这条输入属于哪一类。
const classifyChain = ChatPromptTemplate.fromMessages([
  [
    "system",
    [
      "判断用户消息类型，只输出以下三个类别之一：",
      "- bug",
      "- feature",
      "- doc",
    ].join("\n"),
  ],
  ["user", "{input}"],
])
  .pipe(model)
  .pipe(new StringOutputParser());

// 三条前置链分别给不同场景补一个 scene 字段。
const techPrefilter = RunnablePassthrough.assign({
  scene: () => "bug",
});

const featurePrefilter = RunnablePassthrough.assign({
  scene: () => "feature",
});

const docPrefilter = RunnablePassthrough.assign({
  scene: () => "doc",
});

// 根据 classifyChain 的结果选择不同前置链。
const routeByType = RunnableBranch.from([
  [({ type }: { type: string }) => type.trim() === "bug", techPrefilter],
  [({ type }: { type: string }) => type.trim() === "feature", featurePrefilter],
  docPrefilter,
]);

// 这一段还是 LCEL 前置链：先分类，再分支。
const preProcess = RunnablePassthrough.assign({ type: classifyChain }).pipe(
  routeByType,
);

// Agent 负责拿到整理后的结果，生成最终回复。
const agent = createAgent({
  model,
  tools: [],
  systemPrompt: [
    "你是一个技术任务处理助手。",
    "scene=bug 时，优先给出排查步骤和临时止血方案。",
    "scene=feature 时，先确认需求范围，再给出实现建议。",
    "scene=doc 时，指出需要补充的文档位置和关键说明。",
  ].join("\n"),
});

// 先跑前置链，拿到 type / scene / input。
const preProcessed = await preProcess.invoke({
  input: "支付接口偶尔返回 502，订单状态显示为未知，用户重复提交了三次。",
});

// 再把路由后的结果整理成消息，交给 Agent。
const result = await agent.invoke({
  messages: [
    {
      role: "user",
      content: [
        `scene=${preProcessed.scene}`,
        `type=${preProcessed.type}`,
        `input=${preProcessed.input}`,
      ].join("\n"),
    },
  ],
});

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

这段代码里，分工是清楚的：

• classifyChain 负责类型分类。
• RunnableBranch 负责把不同输入送进不同前置链。
• preProcess 负责把路由结果整理好。
• agent 负责最终回复。

所以这里的 RunnableBranch 不是在替代 Agent，而是在 Agent 前面做路由。

6. 什么时候用 RunnableBranch，什么时候用 RunnableLambda

这两个都能做路由，但适合的场景不一样。

优先用 RunnableBranch

• 分支数量固定。
• 条件规则清楚。
• 你希望结构一眼就能看懂。

再考虑 RunnableLambda

• 分支来自配置、数据库或别的外部来源。
• 不是简单的布尔条件，而是更动态的计算。
• 你能接受把一部分控制逻辑拿回函数里。

简单说：分支固定用 RunnableBranch，路由特别动态再考虑 RunnableLambda。如果只是普通的类型路由，RunnableBranch 通常已经够用了。

7. 一个判断标准

可以直接记一个实用的判断标准：

• 如果问题是「同一份输入，要不要同时做几件事」，那通常是上一篇讲的 assign() / RunnableParallel。
• 如果问题是「不同输入，要不要走不同链路」，那通常就是这篇的 RunnableBranch。