[LangChain] 10. 條件路由

所謂條件路由（Conditional Routing），就是在一條 AI 推理/數(shù)據(jù)處理流水線里，先依據(jù)輸入內(nèi)容、上下文或運(yùn)行時(shí)信號(hào)做一次判別，再把請(qǐng)求分發(fā)到不同的子流程（鏈）上執(zhí)行。本質(zhì)上是受控的 if/else：先“判別”，再“選擇”，最后“執(zhí)行”，避免“一個(gè)提示詞走天下”。

在 LangChain.js 中，常見做法有兩種：

用 RunnableBranch 聲明式按順序匹配分支（可視作 if/else-if/else 的舊范式）
用 RunnableLambda 在函數(shù)里動(dòng)態(tài)返回要執(zhí)行的子鏈（官方更推薦，靈活且易組合）

路由條件既可以是規(guī)則驅(qū)動(dòng)（正則、關(guān)鍵字、用戶角色、租戶權(quán)限、時(shí)間段、閾值等），也可以是模型輔助分類（先用輕量 LLM 判斷“數(shù)學(xué)/SQL/閑聊/RAG”等意圖）；命中失敗時(shí)應(yīng)設(shè)置兜底分支，保證可用性。

總結(jié)一句話：條件路由讓系統(tǒng)不再押注單一路徑，而是把“該走哪條路”的選擇設(shè)計(jì)為一等公民。

RunnableBranch

RunnableBranch 是 LangChain.js 中提供的條件分派器。

它把一條流程拆成多條“候選子鏈”，并附帶各自的觸發(fā)條件，運(yùn)行時(shí)從左到右依次評(píng)估條件，命中第一個(gè)就執(zhí)行對(duì)應(yīng)子鏈，若都不命中則走默認(rèn)分支。可以把它理解成鏈?zhǔn)降?if/else if/else。

適用場(chǎng)景

意圖分流：把“數(shù)學(xué)問(wèn)答 / SQL 咨詢 / 常規(guī)閑聊”分到不同處理鏈。
策略切換：根據(jù)上下文選擇不同 Prompt / 模型 / 溫度或不同的工具集合。
合規(guī)與風(fēng)控：命中敏感詞/權(quán)限不足 → 分流到“拒答/脫敏/人工審核”鏈。
降本增效：先走便宜路徑（檢索/小模型），只有命中特定條件才走貴路徑（推理型大模型）。
容錯(cuò)兜底：為無(wú)法歸類或條件未命中的情況提供穩(wěn)定的默認(rèn)輸出鏈。

基礎(chǔ)語(yǔ)法

RunnableBranch 是一個(gè)類，因此使用的時(shí)候需要實(shí)例化：

new RunnableBranch({ branches, default })

branches：對(duì)應(yīng)的類型為 Branch<Runinput, Runoutput>[]，其中 Runinput 是條件，Runoutput 是被選中后的子鏈。
default：無(wú)分支命中時(shí)執(zhí)行的 Runnable<RunInput, RunOutput>

RunnableBranch 還有一個(gè)靜態(tài)方法 RunnableBranch.from([...BranchLike[], defaultRunnableLike])，該方法接受一個(gè)數(shù)組，數(shù)組的前 N 項(xiàng)是 [條件, 子鏈] 的數(shù)組，最后一項(xiàng)是默認(rèn)子鏈。

RunnableBranch.from([
  [條件1, 子鏈1],
  [條件2, 子鏈2],
  [條件3, 子鏈3],
  默認(rèn)子鏈
])

練習(xí)

使用 RunnableBranch 創(chuàng)建子鏈

import { ChatOllama } from "@langchain/ollama";
import {
  ChatPromptTemplate,
  SystemMessagePromptTemplate,
} from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import {
  RunnableBranch,
  RunnableLambda,
  RunnablePassthrough,
  RunnableSequence,
} from "@langchain/core/runnables";

// 1. 模型
const model = new ChatOllama({
  model: "llama3",
});

// 2. 解析器
const parser = new StringOutputParser();

// 3. 三條“領(lǐng)域鏈” —— 物理 / 數(shù)學(xué) / 其他
const physicsChain = ChatPromptTemplate.fromTemplate(
  `
你是一位物理學(xué)家，擅長(zhǎng)使用中文回答物理相關(guān)的問(wèn)題，當(dāng)你不知道問(wèn)題的答案時(shí)，你就回答不知道。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 物理鏈

const mathChain = ChatPromptTemplate.fromTemplate(
  `
你是一個(gè)數(shù)學(xué)家，擅長(zhǎng)使用中文回答數(shù)學(xué)相關(guān)的問(wèn)題，當(dāng)你不知道問(wèn)題的答案時(shí)，你就回答不知道。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 數(shù)學(xué)鏈

const otherChain = ChatPromptTemplate.fromTemplate(
  `
你是一個(gè)AI助手，你會(huì)使用中文回答以下問(wèn)題。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 其它鏈

// 4. 分類鏈：輸出“數(shù)學(xué)”/“物理”/“其它”
const classifyChain = ChatPromptTemplate.fromMessages([
  [
    "system",
    `
你是一個(gè)僅輸出標(biāo)簽的分類器。
任務(wù)：將用戶問(wèn)題分類為以下三類之一：數(shù)學(xué)、物理、其它。
規(guī)則：
- 只能輸出且必須輸出其中一個(gè)：數(shù)學(xué) 或 物理 或 其它
- 不要輸出任何解釋、理由、引號(hào)、標(biāo)點(diǎn)、前后綴或額外空行
- 若不確定，輸出：其它
    `.trim(),
  ],
  [
    "human",
    `
問(wèn)題：
{input}

現(xiàn)在輸出分類（僅一個(gè)詞）：`.trim(),
  ],
])
  .pipe(model)
  .pipe(parser); // 分類鏈

// 5. 分派器：根據(jù) topic 路由到對(duì)應(yīng)子鏈
const answerBranch = RunnableBranch.from([
  [
    (input) => input.topic.includes("物理"),
    RunnableLambda.from((x) => {
      console.log("執(zhí)行了物理鏈");
      return x;
    }).pipe(physicsChain),
  ],
  [
    (input) => input.topic.includes("數(shù)學(xué)"),
    RunnableLambda.from((x) => {
      console.log("執(zhí)行了數(shù)學(xué)鏈");
      return x;
    }).pipe(mathChain),
  ],
  otherChain,
]);

// 6. 最終組合 - 最終給外部使用的，就是這個(gè) finalChain
const finalChain = RunnableSequence.from([
  // 當(dāng)前的入?yún)?shù)：{ input: "什么是經(jīng)典力學(xué)？" } --> { input: "什么是經(jīng)典力學(xué)？" : topic: "物理"}
  RunnablePassthrough.assign({
    topic: classifyChain,
  }),
  answerBranch,
]);

// 測(cè)試
const run = async () => {
  console.log(await finalChain.invoke({ input: "什么是經(jīng)典力學(xué)？" }));
  console.log(
    await finalChain.invoke({ input: "對(duì) y = x 求導(dǎo)的結(jié)果是多少？" })
  );
  console.log(await finalChain.invoke({ input: "你好，你是誰(shuí)？" }));
};
run();

RunnableLambda

不過(guò)現(xiàn)在官方更加推薦使用 RunnableLambda 的方式。因?yàn)?RunnableLambda 更靈活，在函數(shù)里可以做復(fù)雜判別、動(dòng)態(tài)返回任意子鏈，便于與外部系統(tǒng)結(jié)合。

練習(xí)

使用 RunnableLambda 重構(gòu)上面例子

import { ChatOllama } from "@langchain/ollama";
import {
  ChatPromptTemplate,
  SystemMessagePromptTemplate,
} from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import {
  RunnableBranch,
  RunnableLambda,
  RunnablePassthrough,
  RunnableSequence,
} from "@langchain/core/runnables";

// 1. 模型
const model = new ChatOllama({
  model: "llama3",
});

// 2. 解析器
const parser = new StringOutputParser();

// 3. 三條“領(lǐng)域鏈” —— 物理 / 數(shù)學(xué) / 其他
const physicsChain = ChatPromptTemplate.fromTemplate(
  `
你是一位物理學(xué)家，擅長(zhǎng)使用中文回答物理相關(guān)的問(wèn)題，當(dāng)你不知道問(wèn)題的答案時(shí)，你就回答不知道。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 物理鏈

const mathChain = ChatPromptTemplate.fromTemplate(
  `
你是一個(gè)數(shù)學(xué)家，擅長(zhǎng)使用中文回答數(shù)學(xué)相關(guān)的問(wèn)題，當(dāng)你不知道問(wèn)題的答案時(shí)，你就回答不知道。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 數(shù)學(xué)鏈

const otherChain = ChatPromptTemplate.fromTemplate(
  `
你是一個(gè)AI助手，你會(huì)使用中文回答以下問(wèn)題。
具體問(wèn)題如下：
{input}
`
)
  .pipe(model)
  .pipe(parser); // 其它鏈

// 4. 分類鏈：輸出“數(shù)學(xué)”/“物理”/“其它”
const classifyChain = ChatPromptTemplate.fromMessages([
  [
    "system",
    `
你是一個(gè)僅輸出標(biāo)簽的分類器。
任務(wù)：將用戶問(wèn)題分類為以下三類之一：數(shù)學(xué)、物理、其它。
規(guī)則：
- 只能輸出且必須輸出其中一個(gè)：數(shù)學(xué) 或 物理 或 其它
- 不要輸出任何解釋、理由、引號(hào)、標(biāo)點(diǎn)、前后綴或額外空行
- 若不確定，輸出：其它
    `.trim(),
  ],
  [
    "human",
    `
問(wèn)題：
{input}

現(xiàn)在輸出分類（僅一個(gè)詞）：`.trim(),
  ],
])
  .pipe(model)
  .pipe(parser); // 分類鏈

// 5. Router
const router = RunnableLambda.from(async (input) => {
  // input: { input: "什么是經(jīng)典力學(xué)？" }
  const topic = await classifyChain.invoke({ input });
  console.log(`topic: ${topic}`);
  const chosenChain =
    topic === "物理" ? physicsChain : topic === "數(shù)學(xué)" ? mathChain : otherChain;

  return await chosenChain.invoke({ input });
});

// 6. 最終組合 - 最終給外部使用的，就是這個(gè) finalChain
const finalChain = RunnableSequence.from([
  RunnableLambda.from((x) => x),
  router,
]);

// 測(cè)試
const run = async () => {
  console.log(await finalChain.invoke({ input: "什么是經(jīng)典力學(xué)？" }));
  console.log(
    await finalChain.invoke({ input: "對(duì) y = x 求導(dǎo)的結(jié)果是多少？" })
  );
  console.log(await finalChain.invoke({ input: "你好，你是誰(shuí)？" }));
};
run();

-EOF-

posted @ 2025-11-04 15:08 Zhentiw 閱讀(5) 評(píng)論(0) 收藏舉報(bào)

刷新頁(yè)面返回頂部

Answer1215

[LangChain] 10. 條件路由

RunnableBranch

RunnableLambda

公告