Langchain教程六(LCEL下篇)

在上一节Langchain教程五(LCEL 上篇)，我们介绍了 LangChain 中的 LCEL 的核心概念、基础语法以及如何通过 Runnable 将 Prompt、LLM、Parser 进行顺序与并行组合。本篇将带你更进一步，从异常处理、流式体验、调试工具以及运行时修改配置参数，真正掌握 LCEL 。

| 与 .pipe 的差异

在 LCEL 中，| 运算符和 .pipe() 方法都可以用于组合 Runnable。它们在功能上是等价的，都用于将一个 Runnable 的输出连接到另一个 Runnable 的输入。然而，它们在语法和某些特定场景下存在一些细微的差异。

| (管道运算符)

语法：

runnable1 | runnable2 | runnable3

特点：

• 更简洁、更具可读性。 这是 LCEL 最推荐和最常用的组合方式，因为它直观地表示了数据流动的方向。

• Pythonic： 类似于 Unix/Linux 中的管道命令，符合 Python 的简洁风格。

绝大多数情况下，我们都优先使用 | 运算符。它让你的链条看起来更像一个自然的数据流。

.pipe()

语法：

runnable1.pipe(runnable2).pipe(runnable3)

特点：

• 方法链式调用： 允许你像调用对象方法一样链式地组合 Runnable。

• 可读性稍差： 相对于 | 而言，可能会显得稍微冗长。

• 可以用于条件判断中

在某些旧版本的 LangChain 代码中，或者当你从其他支持 .pipe() 模式的库（如 Pandas）迁移时，可能会看到或需要使用它。

pipe 可以方便的用于动态组合中， 在一些情况下，如果我们需要动态地构建链条（例如，在循环中根据条件添加组件），.pipe() 会更便利，因为它是一个方法调用。

异常与兜底策略 （RunnableWithFallback）

在一些的实际 LLM 应用中，外部服务（如 LLM API）或内部逻辑可能会出现错误。LCEL 提供了 RunnableWithFallback 来实现健壮的兜底（fallback）策略，确保即使主组件失败，应用也能优雅地降级或提供替代响应。

它允许我们定义一个主要的 Runnable 和一个或多个备用的 Runnable。如果主 Runnable 抛出异常，RunnableWithFallback 会捕获该异常，并尝试按顺序执行提供的备用 Runnable，直到有一个成功返回结果。

demo

假设我们有一个主要的 LLM 模型，但为了防止 API 调用失败或模型返回不理想的结果，我们希望有一个更简单的、硬编码的备用响应。

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 有 50% 的概率抛出错误。当 main_llm_chain 失败时，robust_chain 会自动尝试执行 fallback_runnable。

RunnableWithFallback 是构建高可用和容错 LLM 应用的关键组件，在实际开发中一定要加上。

链的可视化

在 LangChain LCEL 中，你构建的所有 chain，本质上是一个有向图（DAG，Directed Acyclic Graph）结构。

• 每个节点代表一个 Runnable（比如 PromptTemplate、LLM、OutputParser）

• 节点之间用箭头连接，表示数据流的方向

• get_graph() 可以返回内部 DAG 结构的一个 Graph 对象

我们可以使用 chain.get_graph().print_ascii() ， 它会将 graph 可视化为一个纯文本形式的结构图（ASCII Art），类似下面这个并行链：

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 库哦，

最终打印出来效果为

           +----------------------------------+            
           | Parallel<cat_joke,dog_joke>Input |            
           +----------------------------------+            
                   ***               ***                   
                ***                     ***                
              **                           **              
+--------------------+              +--------------------+ 
| ChatPromptTemplate |              | ChatPromptTemplate | 
+--------------------+              +--------------------+ 
           *                                   *           
           *                                   *           
           *                                   *           
    +------------+                      +------------+     
    | ChatOpenAI |                      | ChatOpenAI |     
    +------------+                      +------------+     
           *                                   *           
           *                                   *           
           *                                   *           
  +-----------------+                 +-----------------+  
  | StrOutputParser |                 | StrOutputParser |  
  +-----------------+                 +-----------------+  
                   ***               ***                   
                      ***         ***                      
                         **     **                         
          +-----------------------------------+            
          | Parallel<cat_joke,dog_joke>Output |            
          +-----------------------------------+  

是不是非常的直观？

运行时修改配置

在构建复杂 LLM 应用链时，我们经常会遇到以下需求：

• 让用户自由调整温度、模型等参数；

• 动态切换不同组件，如不同模型（GPT、Claude）、不同提示模板（prompt）；

• 在不重建整个链的情况下做 A/B 测试或快速实验。

为此，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： 对于新的链条构建，优先选择 LCEL 而不是传统的 Chain 类。

• 明确输入输出类型： 在设计 Runnable 时，清晰地定义它们的预期输入和输出类型（dict, str, Message），并使用类型提示。这有助于提高代码的可读性和可维护性。

• 利用 RunnableParallel 处理多输入： 当链条的后续部分需要多个独立输入时，使用 RunnableParallel 预处理并收集这些输入到字典中。

• 善用 RunnableLambda 封装自定义逻辑： 将任何自定义的 Python 函数（同步或异步）包装成 RunnableLambda，以便无缝集成到 LCEL 链条中。

• 利用 RunnableWithFallback 增强健壮性： 为关键组件或整个链条添加兜底策略，提高应用的容错能力。

• 集成 LangSmith 进行可观测性： 尽早集成 LangSmith，它能提供强大的跟踪、调试和性能分析能力。

• 考虑流式传输： 如果用户体验很重要，尽可能使用 .stream() 或 .astream() 来实现实时响应。

• 模块化设计： 将复杂的链条分解成更小的、可复用的 Runnable 子链，提高代码的组织性和可测试性。

常见坑：

• 输入输出类型不匹配： 这是最常见的错误。一个 Runnable 的输出必须与下一个 Runnable 的预期输入类型兼容。例如，如果你期望一个字典，但上一个组件输出的是字符串，就会报错。

• 解决方案： 使用 RunnableLambda 或 RunnablePassthrough 进行显式转换和包装。

• 字典键不匹配： 当使用 PromptTemplate 或 RunnableParallel 时，确保字典的键与模板中的变量名或后续组件期望的键名完全一致。

• 异步/同步混用不当： 确保在异步上下文中调用 ainvoke/astream/abatch，在同步上下文中调用 invoke/stream/batch。LCEL 会尝试自动处理，但手动确保一致性可以避免意外。

• 忘记 invoke() 或 stream()： 组合链条只是定义了数据流，你必须显式调用 .invoke()、.stream() 等方法才能执行它。

• RunnableLambda 的参数问题： RunnableLambda 包装的函数只接收一个参数，这个参数就是上一个 Runnable 的输出。如果你需要多个输入，请确保上一个 Runnable 输出的是一个字典，并且你的 RunnableLambda 函数能正确解构这个字典。

• LLM 响应格式不一致： LLM 的输出可能不总是按照预期格式化，导致 OutputParser 失败。

• 解决方案： 增加 LLM 提示的鲁棒性，使用更强大的解析器（如 PydanticOutputParser），并结合 RunnableWithFallback。

• 依赖注入和配置： 在大型应用中，管理 LLM API 密钥、数据库连接等配置时，确保它们能正确地传递给 Runnable 组件

总结

好了，以上就是咱们的全部内容，在这篇文章中，我们了解了 | 与 .pipe() 的差异、如何处理异常（RunnableWithFallback）、链的可视化工具（get_graph().print_ascii()）、以及运行时配置字段与组件切换（configurable_fields / configurable_alternatives）。通过这些能力，咱们可以构建更灵活、健壮、可调试且可组合的 LLM 应用链。