LangChain 中如何实现条件分支和动态路由？

标签：LangChain、条件路由

🏛 业务架构师

LangChain 里实现条件分支与动态路由，三种由轻到重的方式：

方式一·RunnableBranch（简单分支）：

python

branch = RunnableBranch(
    (lambda x: x["type"] == "complaint", complaint_chain),
    (lambda x: x["type"] == "faq", faq_chain),
    default_chain,  # 兜底
)

按条件选链执行——适合分支少（2~4 条）、判断逻辑确定的场景。

方式二·LCEL 内联路由：用 RunnableLambda 做分发函数，或 RunnableParallel 后接选择逻辑——轻量场景最省事。

方式三·语义路由（业务最常用）：用 LLM 或分类器判断输入意图再路由——

python

router = (
    {"topic": classifier_prompt | model | parser}  # LLM判断意图
    | lambda x: route_map[x["topic"]]
)

适合用户输入不可预知的场景（客服分流：技术问题→知识库链、投诉→安抚链、闲聊→对话链）。工程要点：意图分类用小模型（快且便宜）、路由结果要落日志（排查"走错链"必备）、未知意图必须有兜底链。

方式四·LangGraph（复杂流程）：条件边（conditional edges）显式建模状态流转，支持循环回退——流程复杂（审批、多轮澄清）时的正解。

业务视角的设计原则：路由是"业务编排"的核心抽象——不同客群、不同意图、不同 SLA 走不同链路（也对应不同成本模型：VIP 走旗舰模型、普通流量走小模型）。我建议把路由规则外置成配置（运营可调），而不是硬编码——路由策略会随运营认知频繁演进，这是"配置驱动架构"在 LLM 应用里的典型落点。

🏗 系统架构师

LangChain 条件分支与动态路由我给三种姿势（由简到强）：

1. RunnableLambda + 字典分发（LCEL 原生）：

python

def route(info):
    if "退款" in info["query"]:
        return refund_chain
    return general_chain

branch = RunnableBranch(
    (lambda x: x["type"] == "faq", faq_chain),
    (lambda x: x["type"] == "doc", rag_chain),
    default_chain,   # 兜底
)

RunnableBranch 按条件顺序匹配执行第一个命中的链——简单路由首选。

2. 语义路由（LLM 分类决策）：

python

router = classify_prompt | model | StrOutputParser()  # 输出 "faq"/"doc"/"chat"
full = router | {"faq": faq_chain, "doc": rag_chain, "chat": chat_chain}

意图不可枚举时用小模型分类（延迟 100~300ms，可用 embedding 相似度替代进一步提速）再分发——客服/助手类系统的标准入口层。

3. LangGraph 条件边（复杂流程）：

python

graph.add_conditional_edges(
    "router",
    lambda s: "need_search" if s["confidence"] < 0.7 else "answer",
    {"need_search": "retriever", "answer": "generate"}
)

图状态机带循环与状态——路由结果影响后续多步、需要回环重试时用它。

架构师要点：

路由层是系统的 API Gateway：意图识别准确率是全线路上限（分错路全白搭）——路由器要有自己的评测集；
兜底必配：未匹配路由走 default（通用链或人工），不许静默失败；
路由可观测：记录每次路由决策与依据（分类置信度），低置信样本回流优化；
简单规则优先：能关键词/规则分流的不要上模型（零成本零延迟），模型路由留给语义模糊场景。

💻 全栈开发专家

条件分支和动态路由在 LCEL 和 LangGraph 里各有标准做法：

1. LCEL：RunnableLambda 返回下一个分支：

python

from langchain_core.runnables import RunnableLambda, RunnableBranch

tech_chain = (
    {"context": tech_retriever | format} | prompt | llm)
chat_chain = chat_prompt | llm

router = RunnableBranch(
    (lambda x: "数据库" in x["question"] or is_tech(x), tech_chain),
    (lambda x: x.get("vip"), vip_chain),
    chat_chain)          # 默认分支

或纯函数路由：RunnableLambda(lambda x: tech_chain if ... else chat_chain)。

2. LangGraph：条件边（循环/复杂路由的正解）：

python

from langgraph.graph import StateGraph, END

def route(state) -> str:                     # 看状态决定去向
    last = state["messages"][-1]
    if last.tool_calls:  return "tools"      # 需要调工具
    if state["retries"] > 2: return "human"  # 重试超限转人工
    return END

g.add_conditional_edges("agent", route,
    {"tools": "tools", "human": "human", END: END})

路由器实现的三种引擎（按可控性排）：

规则（关键词/正则/结构化字段）：快、确定、可解释——能规则的别用模型；
小分类器（BERT/小 LLM 意图分类）：意图模糊空间大时用，毫秒级；
LLM 路由（Function Calling 选分支）：最灵活最贵，兜底用。

工程要点：

路由必须有默认分支（default/END），漏配条件直接抛异常；
路由决策写日志（进了哪个分支、依据什么）——「消息没回复」排查第一站；
路由准确率要单独监控/评测（分错支的体验是灾难性的答非所问）；
LangGraph 的 route 返回值和映射表 key 必须严格一致，拼错键名是高频低级 bug（用常量枚举防）。

LangChain 中如何实现条件分支和动态路由？ ​

🏛 业务架构师 ​

🏗 系统架构师 ​

💻 全栈开发专家 ​

LangChain 中如何实现条件分支和动态路由？

🏛 业务架构师

🏗 系统架构师

💻 全栈开发专家