LangChain LCEL 高级模式：分支、降级与自定义 Runnable

前几章介绍了 LCEL 的基本链式调用。实际生产环境中，链不是线性的——需要根据输入条件路由到不同分支，需要在主链失败时自动切换备用模型，需要注入自定义的 Python 逻辑。本章系统讲解 LCEL 的高级模式，让链具备生产级的健壮性。

1.1 RunnableBranch：条件路由

LCEL高级模式图
LCEL三种高级模式——RunnableBranch条件路由、with_fallbacks降级备用、自定义RunnableLambda

RunnableBranch 实现了类似 switch-case 的语义：根据输入满足的条件，把执行路径路由到对应的链。

1.1.1 基本语法

python

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (condition1, chain1),   # 如果 condition1(input) 为真，执行 chain1
    (condition2, chain2),   # 如果 condition2(input) 为真，执行 chain2
    default_chain           # 所有条件都不满足时，执行 default_chain
)

条件函数接收整个输入 dict，返回布尔值。条件按顺序求值，第一个为 True 的分支执行，其余跳过。

1.1.2 实战：意图识别→路由到不同处理链

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
strong_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# Step 1：意图识别链（第一步，便宜模型即可）
intent_prompt = ChatPromptTemplate.from_messages([
    ("system", """识别用户问题的意图，只输出以下类型之一：
- code_review: 代码审查/优化请求
- architecture: 架构设计/技术选型问题
- debug: 排查 Bug/错误分析
- general: 一般技术问题"""),
    ("human", "{question}")
])

intent_chain = intent_prompt | llm | StrOutputParser()

# Step 2：三条专用处理链

# 代码审查链：需要强模型
code_review_chain = (
    ChatPromptTemplate.from_messages([
        ("system", """你是代码审查专家。对提交的代码从以下维度审查：
安全性、性能、可读性、最佳实践。
对每个问题给出严重程度（Critical/Warning/Info）和具体修改建议。"""),
        ("human", "{question}")
    ])
    | strong_llm
    | StrOutputParser()
)

# 架构设计链：需要强模型
architecture_chain = (
    ChatPromptTemplate.from_messages([
        ("system", """你是系统架构师，擅长分布式系统和微服务设计。
回答架构问题时需要：
1. 分析需求和约束
2. 给出 2-3 种方案对比
3. 推荐最适合的方案及理由
4. 指出潜在风险"""),
        ("human", "{question}")
    ])
    | strong_llm
    | StrOutputParser()
)

# Debug 链：中等模型，步骤化输出
debug_chain = (
    ChatPromptTemplate.from_messages([
        ("system", """你是调试专家。帮助分析和解决技术问题。
使用以下结构回答：
1. 问题定位（可能的原因）
2. 排查步骤
3. 解决方案
4. 预防措施"""),
        ("human", "{question}")
    ])
    | llm
    | StrOutputParser()
)

# 通用链
general_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "你是一名全栈工程师，回答各类技术问题。"),
        ("human", "{question}")
    ])
    | llm
    | StrOutputParser()
)

# Step 3：组装分支路由
branch_router = RunnableBranch(
    (lambda x: "code_review" in x["intent"], code_review_chain),
    (lambda x: "architecture" in x["intent"], architecture_chain),
    (lambda x: "debug" in x["intent"], debug_chain),
    general_chain  # 默认分支
)

# Step 4：完整链（意图识别 + 路由）
full_chain = (
    RunnablePassthrough.assign(
        intent=lambda x: intent_chain.invoke({"question": x["question"]})
    )
    | branch_router
)

# 测试不同意图的路由
test_questions = [
    "帮我审查这段 Python 代码是否有 SQL 注入风险",
    "微服务架构下如何设计订单系统？",
    "FastAPI 返回 422 Validation Error 怎么排查？",
    "Python 的 GIL 是什么？"
]

for q in test_questions:
    result = full_chain.invoke({"question": q})
    print(f"Q: {q[:30]}...\nA: {result[:100]}...\n")

1.2 with_fallbacks：主链失败时自动切换备用链

with_fallbacks（降级兜底：当主链报错时自动切换到备用链，保证服务不中断）为任何 Runnable 添加降级能力：主链抛出异常时，自动尝试备用链，直到成功或所有备用链都失败。

1.2.1 场景：OpenAI 限流→国产模型兜底

python

from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatDeepInfra
from langchain_anthropic import ChatAnthropic
from langchain_core.exceptions import OutputParserException

# 主链：OpenAI GPT-4o（可能限流或费用超支）
primary_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# 备用链 1：Anthropic Claude（不同供应商，独立限流池）
fallback_llm_1 = ChatAnthropic(model="claude-opus-4-6", temperature=0.3)

# 备用链 2：更便宜的模型（作为最后保底）
fallback_llm_2 = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# 指定哪些异常触发降级
# exceptions_to_handle 默认是 Exception（所有异常都降级）
# 可以精确指定，避免业务逻辑错误也触发降级
from openai import RateLimitError, APIStatusError

resilient_llm = primary_llm.with_fallbacks(
    [fallback_llm_1, fallback_llm_2],
    exceptions_to_handle=(RateLimitError, APIStatusError)
)

# 在完整链中使用
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名技术写作专家。"),
    ("human", "{topic}")
])

# 整个链都具备降级能力
resilient_chain = prompt | resilient_llm | StrOutputParser()

result = resilient_chain.invoke({"topic": "解释什么是向量数据库"})

1.2.2 为整个链配置降级

降级不限于 LLM，可以为任意子链配置：

python

# 主 RAG 链（依赖向量数据库）
primary_rag_chain = retriever | format_docs_prompt | llm | StrOutputParser()

# 备用链（直接用 LLM 回答，不检索）
fallback_direct_chain = direct_prompt | llm | StrOutputParser()

# 整个 RAG 链的降级
resilient_rag = primary_rag_chain.with_fallbacks(
    [fallback_direct_chain],
    exceptions_to_handle=(ConnectionError, TimeoutError)
)

1.3 RunnablePassthrough 与 RunnableLambda

1.3.1 RunnablePassthrough：数据透传

RunnablePassthrough 把输入原样传递（或添加新字段后传递），不做任何转换：

python

from langchain_core.runnables import RunnablePassthrough

# 用法 1：透传（通常作为链的第一步，让后续步骤能访问原始输入）
passthrough = RunnablePassthrough()

# 用法 2：assign——在原有 dict 上增加新字段
chain_with_context = (
    RunnablePassthrough.assign(
        # 并行调用多个函数，结果以新字段加入 dict
        word_count=lambda x: len(x["text"].split()),
        char_count=lambda x: len(x["text"]),
        language=detect_language_chain  # 也可以是另一条链
    )
    | summarize_prompt
    | llm
)

# 调用时，原始 text 和新增的 word_count 等字段都在 dict 中
result = chain_with_context.invoke({"text": "这是一段需要总结的文章..."})

1.3.2 RunnableLambda：把 Python 函数变成 Runnable

python

from langchain_core.runnables import RunnableLambda

# 同步函数
def format_query(inputs: dict) -> dict:
    """预处理用户输入：清理空白、转小写、截断"""
    query = inputs["query"].strip().lower()[:500]
    return {"query": query, "original": inputs["query"]}

# 异步函数
async def async_lookup(inputs: dict) -> dict:
    """异步查询外部 API"""
    user_id = inputs["user_id"]
    user_profile = await fetch_user_profile(user_id)  # 异步 IO
    return {**inputs, "user_profile": user_profile}

# 转换为 Runnable
format_runnable = RunnableLambda(format_query)
lookup_runnable = RunnableLambda(async_lookup)

# 组装进链中
preprocessing_chain = (
    format_runnable
    | lookup_runnable
    | main_prompt
    | llm
    | StrOutputParser()
)

1.4 @chain 装饰器：函数即链

@chain 装饰器是更优雅的写法，把任意 Python 函数转换成 Runnable，支持流式输出：

python

from langchain_core.runnables import chain

@chain
def smart_router(inputs: dict) -> str:
    """复杂的路由逻辑，比 RunnableBranch 更灵活"""
    question = inputs["question"]

    # 根据关键词判断路由，比 LLM 分类更快更稳定
    if any(kw in question for kw in ["代码", "函数", "bug", "错误", "报错"]):
        return code_review_chain.invoke(inputs)
    elif any(kw in question for kw in ["架构", "设计", "选型", "方案"]):
        return architecture_chain.invoke(inputs)
    else:
        # 复杂情况才调用 LLM 分类
        intent = intent_chain.invoke({"question": question})
        return branch_router.invoke({**inputs, "intent": intent})

# @chain 装饰的函数可以直接用管道符组合
full_pipeline = preprocessing_chain | smart_router

1.5 完整实战：多路由+降级的健壮 RAG 链

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_anthropic import ChatAnthropic
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import (
    RunnableBranch, RunnablePassthrough, RunnableLambda, chain
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from openai import RateLimitError

# --------- 模型配置（带降级）---------
primary_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
fallback_llm = ChatAnthropic(model="claude-opus-4-6", temperature=0.2)
resilient_llm = primary_llm.with_fallbacks(
    [fallback_llm],
    exceptions_to_handle=(RateLimitError,)
)

# --------- 向量检索配置 ---------
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# --------- 子链定义 ---------

# RAG 问答链
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是知识库问答助手。仅基于以下上下文回答问题，不要编造信息。

<context>
{context}
</context>"""),
    ("human", "{question}")
])

def format_docs(docs) -> str:
    return "\n\n---\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    RunnablePassthrough.assign(
        context=lambda x: format_docs(retriever.invoke(x["question"]))
    )
    | rag_prompt
    | resilient_llm
    | StrOutputParser()
)

# 直接问答链（RAG 失败时的 fallback）
direct_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名知识助手。基于你的训练知识回答问题，如果不确定请说明。"),
    ("human", "{question}")
])

direct_chain = direct_prompt | resilient_llm | StrOutputParser()

# 不支持的问题类型
@chain
def unsupported_handler(inputs: dict) -> str:
    return f"抱歉，当前系统不支持处理「{inputs.get('intent', '此类')}」类型的问题。请联系人工客服。"

# --------- 意图分类 ---------
classify_prompt = ChatPromptTemplate.from_messages([
    ("system", """将用户问题分类为以下类型之一（只输出类型名）：
- knowledge_query: 知识查询（可用 RAG 回答）
- calculation: 数学计算
- unsupported: 超出范围的请求"""),
    ("human", "{question}")
])

classify_chain = classify_prompt | primary_llm.with_fallbacks([fallback_llm]) | StrOutputParser()

# --------- 路由分支 ---------
router = RunnableBranch(
    (lambda x: "knowledge" in x.get("intent", ""), rag_chain.with_fallbacks([direct_chain])),
    (lambda x: "calculation" in x.get("intent", ""), direct_chain),
    unsupported_handler
)

# --------- 完整流水线 ---------
full_pipeline = (
    RunnablePassthrough.assign(
        intent=lambda x: classify_chain.invoke({"question": x["question"]})
    )
    | router
)

# --------- 流程图 ---------

python

# 测试完整流水线
if __name__ == "__main__":
    test_cases = [
        {"question": "LangChain 的 LCEL 是什么？"},
        {"question": "2 的 100 次方是多少？"},
        {"question": "帮我订一张机票"}
    ]

    for case in test_cases:
        print(f"\n问题: {case['question']}")
        result = full_pipeline.invoke(case)
        print(f"回答: {result}")

1.6 LCEL 高级模式对比

模式	使用场景	关键 API	注意事项
RunnableBranch	基于输入条件路由	`RunnableBranch((cond, chain), default)`	条件按顺序求值，第一个匹配为准
with_fallbacks	主链失败时降级	`.with_fallbacks([chain1, chain2])`	指定 `exceptions_to_handle` 避免误降级
RunnablePassthrough	透传或增加字段	`.assign(field=lambda x: ...)`	并行 assign 性能更好
RunnableLambda	注入自定义逻辑	`RunnableLambda(fn)`	支持异步函数
@chain	复杂函数变 Runnable	`@chain def fn(inputs): ...`	支持流式输出，推荐用于复杂逻辑

LCEL 的逻辑是：所有东西都是 Runnable，因此任何组件都可以用管道符串联、添加降级、并行执行。这五种高级模式覆盖了生产环境中最常见的几类需求。