https://python.langchain.com.cn/docs/expression_language/how_to/routing

LangChain 表达式语言动态路由总结

LangChain 动态路由用于创建非确定性链，核心是让前一步的输出决定下一步执行的逻辑，以此在与 LLMs 交互中提供结构和一致性。文档主要介绍了两种实现方法：RunnableBranch 和自定义工厂函数，均基于“先分类输入，再路由到对应子链”的两步序列逻辑（先将输入分类为 LangChain、Anthropic 或 Other，再匹配对应处理链）。

一、核心前提：先构建“分类链”与“子链”

两种路由方法的基础一致，需先准备两类组件，后续路由逻辑均依赖它们：

1. 分类链：将用户输入（问题）分类为 LangChain、Anthropic 或 Other，输出仅为单个分类关键词（确保后续路由判断准确）。

   from langchain.prompts import PromptTemplate
   from langchain_community.chat_models import ChatAnthropic
   from langchain_core.output_parsers import StrOutputParser
   
   # 分类链：输入问题，输出“LangChain/Anthropic/Other”
   classification_chain = (
       PromptTemplate.from_template("""Given the user question below, classify it as either being about `LangChain`, `Anthropic`, or `Other`.
   Do not respond with more than one word.
   <question>{question}</question>
   Classification:""")
       | ChatAnthropic()  # 调用Anthropic模型
       | StrOutputParser()  # 解析输出为字符串
   )
   

2. 子链：针对不同分类，定义差异化的处理链（如回答开头固定格式、专业领域适配），共3个对应子链：
   • langchain_chain：处理 LangChain 相关问题，回答以“As Harrison Chase told me”开头。
   • anthropic_chain：处理 Anthropic 相关问题，回答以“As Dario Amodei told me”开头。
   • general_chain：处理“Other”类问题，无固定开头格式。

   # LangChain子链示例（其他子链结构类似，仅提示词不同）
   langchain_chain = (
       PromptTemplate.from_template("""You are an expert in langchain. 
   Always answer questions starting with "As Harrison Chase told me". 
   Respond to the following question:
   Question: {question}
   Answer:"""")
       | ChatAnthropic()
   )
   

二、方法一：使用 RunnableBranch 实现路由

RunnableBranch 是 LangChain 提供的官方路由组件，通过“条件-可运行对”列表 + 默认可运行实现路由，逻辑为“匹配第一个满足条件的分支，无匹配则走默认”。

1. 核心原理

• 初始化参数：接收多个 (条件函数, 对应可运行对象) 的元组，最后一个参数为“默认可运行对象”。
• 执行逻辑：将输入传入每个条件函数，返回 True 的第一个条件对应的可运行对象会被执行。

2. 实现步骤

from langchain_core.runnables import RunnableBranch

# 1. 定义RunnableBranch：条件-子链匹配
branch = RunnableBranch(
    # 条件1：若分类结果（topic）含“anthropic”，走anthropic_chain
    (lambda x: "anthropic" in x["topic"].lower(), anthropic_chain),
    # 条件2：若分类结果含“langchain”，走langchain_chain
    (lambda x: "langchain" in x["topic"].lower(), langchain_chain),
    # 默认分支：无匹配时走general_chain
    general_chain
)

# 2. 组合全链：先获取“分类结果（topic）”和“原始问题（question）”，再路由
full_chain = {
    "topic": classification_chain,  # 用分类链生成topic
    "question": lambda x: x["question"]  # 直接传递原始问题
} | branch  # 连接路由分支

# 3. 测试：不同问题会路由到对应子链
full_chain.invoke({"question": "how do I use Anthropic?"})  # 走anthropic_chain
full_chain.invoke({"question": "how do I use LangChain?"})  # 走langchain_chain
full_chain.invoke({"question": "whats 2 + 2"})  # 走general_chain

三、方法二：使用自定义函数实现路由

通过编写自定义函数判断输入分类，返回对应的“可运行对象”（而非执行结果），再用 RunnableLambda 包装函数以适配 LangChain 链的调用逻辑。

1. 核心原理

• 自定义函数：接收前一步的输出（含 topic 分类结果和 question 原始问题），根据 topic 判断并返回对应子链（可运行对象）。
• RunnableLambda：将自定义函数包装为 LangChain 可运行对象，使其能嵌入链的流程中。

2. 实现步骤

from langchain_core.runnables import RunnableLambda

# 1. 定义自定义路由函数：返回对应子链（可运行对象），不执行
def route(info):
    if "anthropic" in info["topic"].lower():
        return anthropic_chain
    elif "langchain" in info["topic"].lower():
        return langchain_chain
    else:
        return general_chain

# 2. 组合全链：先获取分类结果和原始问题，再通过RunnableLambda调用路由函数
full_chain = {
    "topic": classification_chain,
    "question": lambda x: x["question"]
} | RunnableLambda(route)  # 包装自定义路由函数

# 3. 测试：效果与RunnableBranch一致
full_chain.invoke({"question": "how do I use Anthropic?"})
full_chain.invoke({"question": "how do I use LangChain?"})
full_chain.invoke({"question": "whats 2 + 2"})

四、核心总结

两种方法最终效果一致，选择需根据路由规则的复杂度：简单规则用 RunnableBranch（配置更简洁），复杂规则用自定义函数（逻辑更易扩展）。

结尾交付物提议

要不要我帮你整理一份动态路由简化代码模板？包含分类链、子链的基础代码，以及两种路由方法的可直接运行示例，你只需替换模型或提示词就能快速适配自己的场景。