LangChain RunnableBranch 怎样路由输入

我第一次把一个问答入口拆成“技术问题”和“账单问题”时，最容易写成一串 if/else。这样能跑，但路由判断、处理逻辑和后续模型链很快搅在一起。LangChain 的 RunnableBranch 更适合这个场景：把“条件 + Runnable”组成有顺序的分支，输入到来后按顺序判断，执行第一个匹配的处理 Runnable；全部不匹配时交给最后的默认 Runnable。

RunnableBranch 的关键不是“自动理解输入”，而是把已经得到的判断条件组合成可继续串接的 Runnable。条件顺序要从窄到宽，所有分支最好返回同一结构，并始终准备一个明确的默认分支。

官方地址：https://reference.langchain.com/python/langchain-core/runnables/branch/RunnableBranch

先约定输入和输出契约

我现在会先把路由输入定成一个小字典，而不是让每个条件函数自行猜测字符串。下面的示例约定 route 表示已经确定的路由标签，text 保存用户原话；每个分支最终都返回 route、answer 和 source_text，这样后面接格式化器、日志或模型链时不需要再判断分支类型。

如果输入来自上游解析器，建议让解析器负责补齐 route，让 RunnableBranch 只负责选择。这样做的好处是：路由选择可以单独记录，条件函数也不会偷偷触发网络请求或模型调用。

用条件与处理 Runnable 组成分支

RunnableBranch 的构造参数由若干个二元组和最后一个默认 Runnable 组成。二元组左侧是条件，右侧是处理逻辑；普通 Python 可调用对象会被 LangChain 转成 Runnable。示例里我使用确定性字段，方便把“路由器是否选对”与模型输出质量分开。

from typing import Any

from langchain_core.runnables import RunnableBranch


def is_technical(payload: dict[str, Any]) -> bool:
    # 条件只读取稳定字段，不在判断阶段调用外部服务。
    return payload.get("route") == "technical"


def is_billing(payload: dict[str, Any]) -> bool:
    # 账单分支与技术分支使用同一份输入契约。
    return payload.get("route") == "billing"


def handle_technical(payload: dict[str, Any]) -> dict[str, str]:
    # 分支只负责产生统一结构，后续链可以继续消费 answer。
    return {
        "route": "technical",
        "answer": f"技术队列已接收：{payload['text']}",
        "source_text": payload["text"],
    }


def handle_billing(payload: dict[str, Any]) -> dict[str, str]:
    # 账单结果保持与技术结果相同的字段，便于统一格式化。
    return {
        "route": "billing",
        "answer": f"账单队列已接收：{payload['text']}",
        "source_text": payload["text"],
    }


def handle_default(payload: dict[str, Any]) -> dict[str, str]:
    # 未识别的 route 不抛出含糊结果，而是明确交给人工或通用队列。
    return {
        "route": "default",
        "answer": "暂时无法识别路由，请补充问题类型。",
        "source_text": payload.get("text", ""),
    }


router = RunnableBranch(
    # 条件按这里的顺序依次判断。
    (is_technical, handle_technical),
    (is_billing, handle_billing),
    handle_default,  # 最后一个参数是默认分支，不是条件二元组。
)

result = router.invoke({"route": "technical", "text": "接口返回 429 怎么处理"})
print(result["route"], result["answer"])

这里的要点有三个：第一，构造器至少需要一个条件分支和一个默认分支；第二，默认处理函数放在最后，不能写成带条件的二元组；第三，invoke 的输入会传给被选中的 Runnable，而不会自动改造成另一种数据结构。

利用从上到下的首次匹配

我第一次把多个“包含关键词”的条件放在一起时，踩过一个很隐蔽的坑：宽条件写在前面，后面的精确条件永远到不了。RunnableBranch 的语义是从前往后求值，哪个条件第一次返回真，就执行对应分支并停止继续选择；没有条件返回真，才进入 default。

因此，条件排序可以按“最具体的规则在前、最宽泛的规则在后”来安排。例如同一个输入既带有 technical 标签，又满足“文本非空”的通用条件时，技术条件必须先出现。更稳妥的写法是让条件只看一个已经规范化的 route 字段，减少重叠。

from langchain_core.runnables import RunnableBranch


def has_route(payload: dict[str, str]) -> bool:
    # 只把存在明确路由的输入交给专用分支。
    return bool(payload.get("route"))


def handle_known(payload: dict[str, str]) -> dict[str, str]:
    # 保留原始 route，方便审计实际选择。
    return {"route": payload["route"], "answer": "进入专用处理链"}


def handle_any_text(payload: dict[str, str]) -> dict[str, str]:
    # 这是较宽的兜底规则，应放在专用规则之后。
    return {"route": "text", "answer": "进入通用文本处理链"}


ordered_router = RunnableBranch(
    (has_route, handle_known),
    handle_any_text,
)

这段代码并不负责判断“哪个 route 合法”，它只演示顺序。生产环境里可以在上游做枚举校验；如果未知标签仍然允许进入系统，就把它显式标记为 unknown，不要让空值静默落到一个看起来成功的分支。

把路由器接入后续链

RunnableBranch 的价值在于它仍然是 Runnable，所以可以使用管道操作符继续组合。实际项目里我通常把分支输出规范成字典，再接一个统一的格式化函数。无论前面命中哪个分支，后面的链都只处理相同字段。

from langchain_core.runnables import RunnableLambda


def format_result(result: dict[str, str]) -> str:
    # 统一出口只依赖契约字段，不需要知道前面命中了哪条分支。
    return f"[{result['route']}] {result['answer']}"


final_chain = router | RunnableLambda(format_result)
print(final_chain.invoke({"route": "billing", "text": "发票抬头怎么修改"}))

如果某个分支后面要接提示模板或聊天模型，也建议先在分支内部把输入映射成模板需要的字段，再让每个分支返回一致的结果类型。不要让一个分支返回字符串、另一个分支返回字典，除非后续链明确处理这两种类型。

处理默认分支与工程边界

默认分支不是可有可无的装饰。它决定了输入不满足任何条件时的行为，也给日志和人工接管留下稳定入口。我的做法是让默认分支返回可识别的 route=default，并在外层记录原始文本，而不是直接返回空字符串。

还要注意三个边界：

• 条件应尽量轻：条件里做纯字段判断最容易复现；如果必须异步判断，官方 API 支持异步条件，但仍应把超时、错误和成本放在路由设计中考虑。
• 不要把 RunnableBranch 当成状态机：它负责一次输入的分支选择，不负责保存跨请求状态、重试策略或人工审批。
• 统一输出比省几行代码更重要：多个分支共用下游链时，先定义输出字段，能显著降低后续拼接和排错成本。

最后可以用下面这张表快速检查设计：

相关问题

RunnableBranch 会执行多个匹配分支吗？

不会。条件按顺序判断，首个返回真的条件对应的 Runnable 被选中；后面的条件不会继续参与这次选择。

最后一个参数为什么不是二元组？

最后一个参数就是默认 Runnable。它接收原始输入，在前面的所有条件都不匹配时执行。

多个分支可以返回不同类型吗？

可以，但接统一下游链时会增加类型判断。更实用的做法是让分支返回统一字段，或者在分支末尾统一转换。

什么时候应该改用更完整的编排方案？

当需求包含跨多步状态、循环、人工介入、持久化或复杂重试时，RunnableBranch 只适合其中的局部路由，应该把状态和流程交给更合适的编排层。

对我来说，RunnableBranch 最有价值的地方不是替代所有条件逻辑，而是把“选择哪条 Runnable 链”变成一个可以继续组合、记录和复用的组件。先固定输入输出契约，再按具体到宽泛排列条件，最后用默认分支兜底，路由代码就会比散落的 if/else 更容易维护。