多链组合与动态路由
用 RunnableLambda、RunnableBranch 与条件边构建分支逻辑,让一个应用按输入动态选择不同处理链。
为什么需要路由
真实应用往往不是一条链走到底:闲聊走轻量链、问答走 RAG 链、计算走工具链。路由 = 按输入特征动态选择处理路径。
RunnableLambda:在链中插自定义逻辑
from langchain_core.runnables import RunnableLambda
def length_guard(text: str) -> str:
return text[:2000]
chain = prompt | length_guard | llm
按条件路由
from langchain_core.runnables import RunnableBranch
branch = RunnableBranch(
(lambda x: "翻译" in x["q"], translate_chain),
(lambda x: "计算" in x["q"], math_chain),
general_chain, # 兜底分支
)
语义路由:让模型自己分类
条件复杂时,先用一次便宜的分类调用判断意图,再把结果交给 RunnableLambda 分发:
classify = (intent_prompt | llm | StrOutputParser())
route = (
{"intent": classify, "q": RunnablePassthrough()}
| RunnableLambda(lambda x: chains[x["intent"]])
)
result = route.invoke("帮我算一下 23*7")
自定义 Runnable
需要复用复杂逻辑时,继承 Runnable 实现自定义组件,即可无缝参与 LCEL 的 stream/batch 能力。
📝 课后练习
quiz-1 RunnableBranch 的作用是?