Langchain教程六：LCEL下篇实战精讲

深入掌握LangChain中的LCEL高级特性，涵盖异常处理与流式响应机制，全面提升AI应用开发效率。
核心要点：
1. `|`管道运算符与`.pipe()`方法的核心差异及适用场景
2. 借助`RunnableWithFallback`实现异常处理与降级策略
3. 动态组合链条与运行时配置灵活调整技巧

上一节我们探讨了LCEL核心概念与基础语法，包括如何利用Runnable串联Prompt、LLM和Parser，以及并行执行。今天深入挖掘异常兜底、流式输出、链条可视化以及运行时的参数动态调整——这些才是LCEL真正发挥威力的关键点。

| 与 .pipe 的差异

在LCEL中，|运算符和.pipe()方法均用于组合Runnable，功能完全一致——都将前一个Runnable的输出传递至后一个的输入。但仔细区分，语法和适用场景存在细微差异。

| (管道运算符)

语法：

runnable1 | runnable2 | runnable3

特点：

• 更简洁，可读性突出。这也是LCEL最推荐、最常用的组合方式，一眼就能看出数据流向。

• 风格贴近Python，类似Unix/Linux管道命令，使用顺手。

多数场景下，直接使用 | 即可。它让链条看起来像一条自然的数据流水线。

.pipe()

语法：

runnable1.pipe(runnable2).pipe(runnable3)

特点：

• 链式调用，像调用对象方法一样逐个组合。

• 可读性稍逊，写起来略显冗长。

• 但在条件判断或循环中动态构建链条时，.pipe() 更灵活——因为是方法调用，能嵌入循环按条件拼接。

如果你从Pandas这类库迁移过来，或者需要在运行时根据逻辑动态决定下一个组件，.pipe() 便大显身手。

异常与兜底策略 （RunnableWithFallback）

实际开发中，LLM API可能宕机、网络可能抖动、内部逻辑也可能出错。LCEL提供 RunnableWithFallback 实现兜底：定义一条主链，再配置一个或多个备用链。主链一旦抛出异常，自动切换至备用链，直至成功返回结果——优雅降级，避免系统崩溃。

demo

例如，主模型存在失败风险，可预先设置硬编码的备用回复：

from langchain_core.runnables import RunnableWithFallback, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 模拟一个可能失败的 LLM
class FailingChatModel(ChatOpenAI):
    def invoke(self, input, config=None):
        import random
        if random.random() < 0.5:  # 50% 的概率失败
            raise ValueError("模拟 LLM API 调用失败！")
        return super().invoke(input, config)

# 主要的 LLM 链
main_llm_chain = (
    ChatPromptTemplate.from_template("请用一句话描述{topic}。")
    | FailingChatModel(temperature=0.7)  # 使用可能失败的 LLM
    | StrOutputParser()
)

# 兜底的 Runnable：一个简单的硬编码响应
fallback_runnable = RunnableLambda(lambda x: "抱歉，当前无法生成完整描述，这是一个通用回复。")

# 使用 RunnableWithFallback 组合
robust_chain = RunnableWithFallback(
    main_llm_chain,
    fallback_runnable
)

这里 FailingChatModel 有一半概率抛异常。主链失败时，robust_chain 自动降级至 fallback_runnable，确保用户至少得到答复。在生产环境中，这是标配——不要指望所有API调用永远稳定。

链的可视化

LCEL构建的每条链本质上是一个有向无环图（DAG）。每个节点是一个Runnable，箭头表示数据流向。使用 chain.get_graph().print_ascii() 可将该图打印为ASCII艺术字符，直观呈现结构。

举个例子，一个并行笑话生成链：

from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

joke_prompt_cat = ChatPromptTemplate.from_template("请讲一个关于猫的笑话")
joke_prompt_dog = ChatPromptTemplate.from_template("请讲一个关于狗的笑话")

llm = ChatOpenAI(
    model="Qwen/Qwen3-32B",
    openai_api_key="your_key",
    base_url="https://api.siliconflow.cn/v1",
    temperature=0
)

parser = StrOutputParser()

parallel_jokes = RunnableParallel(
    cat_joke=joke_prompt_cat | llm | parser,
    dog_joke=joke_prompt_dog | llm | parser
)

parallel_jokes.get_graph().print_ascii()

注意：需安装 grandalf 库，否则print_ascii会报错。

打印结果大致如下：

           +----------------------------------+            
           | ParallelInput |            
           +----------------------------------+            
                   ***               ***                   
                ***                     ***                
              **                           **              
+--------------------+              +--------------------+ 
| ChatPromptTemplate |              | ChatPromptTemplate | 
+--------------------+              +--------------------+ 
          *                               *           
          *                               *           
          *                               *           
    +------------+                  +------------+     
    | ChatOpenAI |                  | ChatOpenAI |     
    +------------+                  +------------+     
          *                               *           
          *                               *           
          *                               *           
  +-----------------+              +-----------------+  
  | StrOutputParser |              | StrOutputParser |  
  +-----------------+              +-----------------+  
                   ***               ***                   
                      ***         ***                      
                         **     **                         
          +-----------------------------------+            
          | ParallelOutput |            
          +-----------------------------------+  

是否一目了然？调试链条结构时尤为实用。

运行时修改配置

构建复杂链时，经常需要：用户动态调整参数（如温度）、在不同模型间切换、或直接替换整个prompt模板。LCEL提供两大利器：

• configurable_fields：运行时修改某个模块的字段值（例如温度）
• configurable_alternatives：运行时将子模块替换为其他备选项

两者均可配合 .with_config() 使用，让链条变成可插拔的乐高积木。

配置字段：configurable_fields

例如，动态调整温度而非写死：

from langchain_openai import ChatOpenAI
from langchain_core.runnables import ConfigurableField

llm = ChatOpenAI(temperature=0).configurable_fields(
    temperature=ConfigurableField(
        id="llm_temperature",  # 运行时配置时的字段名
        name="LLM Temperature",
        description="用于控制模型输出的随机性"
    )
)

运行时使用 .with_config() 覆盖：

# 提高模型随机性
llm.with_config(configurable={"llm_temperature": 0.9}).invoke("讲个笑话")

将其嵌入链中，同样可以动态控制：

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("请基于以下主题生成笑话：{topic}")
chain = prompt | llm

chain.invoke({"topic": "猫"})  # 默认温度 0

# 设置温度为 0.9，提升创意度
chain.with_config(configurable={"llm_temperature": 0.9}).invoke({"topic": "猫"})

模块替换：configurable_alternatives

假设你想在运行时切换模型（GPT-4 vs Claude）：

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-haiku-20240307").configurable_alternatives(
    ConfigurableField(id="llm"),  # 设置字段 ID
    default_key="anthropic",      # 默认值为当前这个 claude
    openai=ChatOpenAI(),          # 替代选项 1
    gpt4=ChatOpenAI(model="gpt-4")  # 替代选项 2
)

在链中使用：

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("给我讲一个关于 {topic} 的笑话")
chain = prompt | llm

# 默认使用 Claude 模型
chain.invoke({"topic": "熊"})

# 切换为 OpenAI
chain.with_config(configurable={"llm": "openai"}).invoke({"topic": "熊"})

同理，prompt模板也能动态替换：

prompt = PromptTemplate.from_template("给我一个关于 {topic} 的笑话").configurable_alternatives(
    ConfigurableField(id="prompt"),
    default_key="joke",
    poem=PromptTemplate.from_template("写一首关于 {topic} 的诗")
)

llm = ChatOpenAI()
chain = prompt | llm

# 默认使用 joke 模板
chain.invoke({"topic": "狗"})

# 切换为 poem 模板
chain.with_config(configurable={"prompt": "poem"}).invoke({"topic": "狗"})

甚至可以将配置好的链保存为一个“版本”反复调用：

openai_joke_chain = chain.with_config(configurable={"llm": "openai", "prompt": "joke"})
openai_joke_chain.invoke({"topic": "猫"})  # 无需重复配置，直接调用

最佳实践与常见坑总结

最后，梳理实践经验与踩过的坑：

最佳实践：

• 优先选用LCEL：新链一律使用LCEL，避免传统Chain类。
• 明确输入输出类型：提前规划每个Runnable的输入输出（dict、str或Message），添加类型注解，代码可读性倍增。
• 用RunnableParallel处理多输入：当后续组件需要多个独立输入时，先用并行链收集为字典。
• 用RunnableLambda封装自定义逻辑：任何自定义函数（同步或异步）均可包装为RunnableLambda，无缝嵌入链条。
• 添加RunnableWithFallback：关键环节务必兜底，容错是基本素养。
• 集成LangSmith：尽早接入，方便跟踪、调试、性能分析。
• 优先使用流式：用户体验至上，.stream()和.astream()值得投入。
• 模块化拆解：将大链拆分为小链，便于测试和复用。

常见坑：

• 输入输出类型不匹配：最常见的错误。上一个Runnable输出字符串，下一个却期望字典，直接崩溃。解决方法：使用RunnableLambda或RunnablePassthrough显式转换。
• 字典键不匹配：PromptTemplate中的变量名与传入的字典键必须严格一致，多一个少一个都不行。
• 异步/同步混用：异步环境使用ainvoke/astream/abatch，同步环境使用invoke/stream/batch。LCEL虽然能自动适配部分场景，但最好保持一致。
• 忘记invoke/stream：链条只是定义，必须显式调用才能执行。
• RunnableLambda只能接收一个参数：需多输入时，确保前一个Runnable输出字典，函数可解构字典。
• LLM输出格式不稳定：导致OutputParser解析失败。解决：加强prompt引导，或使用更鲁棒的解析器（如PydanticOutputParser），并搭配RunnableWithFallback兜底。
• 配置管理混乱：大型应用中，API Key、数据库连接等配置需确保正确传递到每个Runnable。

总结

以上就是本篇全部内容。我们深入剖析了 | 与 .pipe() 的适用场景、异常兜底（RunnableWithFallback）、链条可视化（get_graph().print_ascii()），以及运行时灵活调参（configurable_fields / configurable_alternatives）。掌握这些，你就能用LCEL构建出既灵活、又健壮、还方便调试的LLM应用链。

来源：互联网