深度拆解LangChain Chains与LCEL：从Runnable到生产级AI工作流

前言在LangChain中Chains链是构建AI应用工作流的核心概念。早期的LangChain提供了SequentialChain等传统方式但配置繁琐且不够灵活。LangChain表达式语言LCEL的诞生正是为了解决这些问题——它提供了一种声明式的、基于管道的方法来组合链使得构建复杂、生产级的任务链变得异常简单和直观。LCEL的出现让LangChain真正成为了一整套AI应用框架。无论你是处理文本预处理、检索增强生成RAG还是构建Agent工作流LCEL都能以统一、简洁、高效的方式串联各个组件。本文将深入拆解Runnable接口和LCEL的八大核心组件通过大量代码示例带你从零到一掌握LangChain工作流的构建精髓。一、Runnable一切可运行单元的基石1.1 什么是RunnableRunnable是LangChain中可以调用、批处理、流式传输、转换和组合的工作单元。简单来说所有能被LangChain“运行”的东西都实现了Runnable接口——语言模型、输出解析器、检索器、编译的LangGraph图等无一例外。Runnable接口强制要求所有LCEL组件实现一组标准方法方法功能描述invoke / ainvoke将单个输入转换为输出batch / abatch批量将多个输入转换为输出stream / astream从单个输入生成流式输出1.2 为什么需要统一调用方式在LCEL出现之前LangChain各组件的调用方式各不相同提示词渲染用 .format()模型调用用 .generate()解析器解析用 .parse()工具调用用 .run()如果你需要串联一个“提示词 → 模型 → 解析器”的流程代码会变得像这样# 传统方式各组件的调用接口五花八门 prompt_text prompt.format(topic猫) # 方法1format model_out model.generate(prompt_text) # 方法2generate result parser.parse(model_out) # 方法3parse每种组件都有自己的调用方式你需要在不同API之间来回切换代码的可读性和可维护性都大打折扣。Runnable统一调用方式后一切都变得优雅了# 分步调用所有组件统一使用invoke prompt_text prompt.invoke({topic: 猫}) # 方法1invoke model_out model.invoke(prompt_text) # 方法2invoke result parser.invoke(model_out) # 方法3invoke无论组件的功能多么复杂模型/提示词/工具调用方式完全相同。这就是统一接口带来的巨大价值。技术要点Runnable接口不仅统一了调用方式还内置了批处理和异步优化。默认情况下batch()方法使用线程池并行执行invoke()而异步方法ainvoke、abatch、astream默认使用asyncio的线程池执行同步版本。二、LCELLangChain表达式语言2.1 LCEL是什么LCELLangChain Expression Language是一种声明式语言用于从现有的Runnable构建新的Runnable。我们称使用LCEL创建的Runnable为“链”Chain而“链”本身也是Runnable——这意味着你可以链中套链无限组合。LCEL的两个主要组合原语是RunnableSequence顺序执行RunnableParallel并行执行许多其他组合原语如RunnableBranch、RunnableWithFallbacks都可以看作是这两个原语的变体。2.2 LCEL的核心优势根据LangChain官方文档LCEL具有以下关键特性自动并行化当LCEL链条中有可以并行执行的步骤时例如从多个检索器中获取文档LCEL会自动执行并行化以最小化延迟。流式支持支持在生成过程中逐步输出结果。异步支持提供完整的异步API支持高并发场景。跟踪和调试自动生成执行轨迹便于调试。2.3 管道符的魔法LCEL最直观的特性就是重载了 | 运算符你可以像搭积木一样连接各个Runnable# 管道式组合 chain prompt | model | parser # 一次性调用整个链 result chain.invoke({topic: 猫})一行代码就完成了“提示词格式化 → 模型调用 → 输出解析”三个步骤的串联简洁程度令人惊叹。三、RunnableSequence可运行序列3.1 核心概念RunnableSequence按顺序“链接”多个可运行对象其中一个对象的输出作为下一个对象的输入。这是LangChain中使用最广泛的组合方式——几乎每条链都用到了它。3.2 基础用法LCEL重载了 | 运算符从两个Runnables创建RunnableSequencechain runnable1 | runnable2 # 等价于 chain RunnableSequence([runnable1, runnable2])3.3 实战示例笑话生成器让我们通过一个完整的示例来感受RunnableSequence的魅力import os from langchain.chat_models import init_chat_model from langchain.core.prompts import PromptTemplate from langchain.core.output_parsers import StrOutputParser # Step 1: 创建提示词模板 # PromptTemplate是Runnable可以使用invoke方法 prompt_template PromptTemplate( template讲一个关于{topic}的笑话, input_variables[topic], ) # Step 2: 初始化聊天模型 # init_chat_model返回的对象也是Runnable llm init_chat_model( modelopenai/gpt-oss-20b:free, # 模型名称 model_provideropenai, # 模型提供商 base_urlhttps://openrouter.ai/api/v1, # API端点 api_keyos.getenv(OPENROUTER_API_KEY), ) # Step 3: 创建输出解析器 # StrOutputParser将模型输出转换为纯文本字符串 parser StrOutputParser() # Step 4: 使用管道符构建链 # prompt_template的输出 → llm的输入 → parser的输入 → 最终结果 chain prompt_template | llm | parser # Step 5: 执行链 resp chain.invoke({topic: 人工智能}) print(resp)代码解读PromptTemplate.invoke() 接收字典将模板中的占位符替换为实际值输出字符串llm.invoke() 接收字符串返回AIMessage对象parser.invoke() 接收AIMessage提取其中的文本内容并返回三个组件通过 | 无缝衔接数据自动流转。3.4 高级特性批处理与流式RunnableSequence 自动支持批处理和流式处理# 批量处理多个输入 topics [人工智能, 机器学习, 深度学习] results chain.batch([{topic: t} for t in topics]) # 流式输出实时逐字返回 for chunk in chain.stream({topic: 人工智能}): print(chunk, end, flushTrue)性能提示RunnableSequence的batch()和abatch()方法默认使用线程池和asyncio.gather对于I/O密集型Runnables如LLM调用比顺序调用invoke快得多。四、RunnableParallel可运行并行4.1 为什么需要并行在实际AI应用中经常需要同时执行多个独立任务。例如同时生成笑话和诗歌同时查询多个知识库同时调用多个API获取数据如果串行执行这些任务总耗时是各任务耗时之和。但如果它们是独立的完全可以并行执行总耗时≈最长任务的耗时。4.2 RunnableParallel的核心机制RunnableParallel同时运行多个可运行对象并为每个对象提供相同的输入。它的内部实现非常巧妙同步执行使用ThreadPoolExecutor在线程池中并发执行异步执行使用asyncio.gather并发执行在LCEL表达式中字典会自动转换为RunnableParallel——这是一个极其方便的语法糖。4.3 实战示例多任务并行处理import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.runnables import RunnableParallel from langchain_core.output_parsers import StrOutputParser llm init_chat_model( modelopenai/gpt-oss-20b:free, model_provideropenai, base_urlhttps://openrouter.ai/api/v1, api_keyos.getenv(OPENROUTER_API_KEY), ) # 定义第一个子链生成笑话 joke_chain ( PromptTemplate.from_template(讲一个关于{topic}的笑话) | llm | StrOutputParser() ) # 定义第二个子链生成诗歌 poem_chain ( PromptTemplate.from_template(写一首关于{topic}的诗歌) | llm | StrOutputParser() ) # 方式1显式使用RunnableParallel map_chain RunnableParallel(jokejoke_chain, poempoem_chain) # 方式2使用字典语法糖推荐 # map_chain {joke: joke_chain, poem: poem_chain} # 执行两个任务并行运行 resp map_chain.invoke({topic: 人工智能}) print(resp) # 输出: {joke: ...笑话内容..., poem: ...诗歌内容...}代码解读joke_chain和poem_chain共享同一个输入{topic: 人工智能}两个链同时执行互不干扰最终输出是一个字典键为joke和poem值为各自链的输出4.4 字典语法的隐式转换机制这里有一个经常让初学者困惑的点为什么可以直接在链中写字典答案是LangChain的隐式转换Coercion机制。当你使用管道符|构建链时RunnableSequence会检查每一个步骤如果它是一个字典Dict系统会自动调用RunnableParallel(dict)将其包装字典中的每个Value也会被递归地转换为Runnable例如lambda函数会被转为RunnableLambda因此你可以写出如此简洁的代码# 这种写法... chain {joke: joke_chain, poem: poem_chain} | combine_chain # ...等价于这种 chain RunnableParallel({joke: joke_chain, poem: poem_chain}) | combine_chain⚡ 性能实战如果每个任务耗时1秒串行执行需要2秒而并行执行只需约1秒。这就是并行带来的性能提升。五、RunnableLambda自定义函数转换5.1 核心概念RunnableLambda将Python可调用函数转换为Runnable使得函数可以在同步或异步上下文中使用。这意味着你可以将任何自定义函数无缝集成到LCEL链中。5.2 基础用法from langchain_core.runnables import RunnableLambda # 方式1构造函数 chain { text1: lambda x: x world, text2: lambda x: x , how are you, } | RunnableLambda(lambda x: len(x[text1]) len(x[text2])) result chain.invoke(hello) print(result) # 输出: 295.3 装饰器语法LangChain还提供了便捷的chain装饰器功能等同于RunnableLambdafrom langchain_core.runnables import chain chain def total_len(x): return len(x[text1]) len(x[text2]) chain { text1: lambda x: x world, text2: lambda x: x , how are you, } | total_len result chain.invoke(hello) print(result) # 输出: 295.4 ⚠️ 重要限制函数必须接收单个参数一个容易被忽视的限制是自定义函数必须接收单个参数。如果你的函数需要多个参数应该使用字典来包装输入# ❌ 错误写法函数需要多个参数 def multiple_length(text1, text2): return len(text1) * len(text2) # ✅ 正确写法用字典包装 def multiple_length(data): return len(data[text1]) * len(data[text2])5.5 流式限制RunnableLambda默认不支持流式传输stream。如果你需要在自定义函数中支持流式处理需要使用RunnableGenerator替代。六、RunnablePassthrough数据透传与上下文保留6.1 核心概念RunnablePassthrough接收输入并将其原样输出是LCEL体系中的“无操作节点”。它的作用看起来很简单但在实际应用中极其有用在流水线中透传输入或保留上下文向输出中添加额外的键值对6.2 基础用法保留原始输入from langchain_core.runnables import RunnablePassthrough, RunnableParallel # 同时输出原始输入和计算结果 chain RunnableParallel( originalRunnablePassthrough(), # 原样透传输入 word_countlambda x: len(x), # 计算单词数量 ) result chain.invoke(hello world) print(result) # {original: hello world, word_count: 11}6.3 高级用法assign()添加键assign()方法可以在透传输入的同时向输出中添加额外的键值对from langchain_core.runnables import RunnablePassthrough chain { text1: lambda x: x world, text2: lambda x: x , how are you, } | RunnablePassthrough.assign( word_countlambda x: len(x[text1] x[text2]) ) result chain.invoke(hello) print(result) # 输出: {text1: hello world, text2: hello, how are you, word_count: 29}6.4 实战场景RAG中的上下文保留在RAG检索增强生成应用中RunnablePassthrough尤为有用# 典型的RAG链结构 rag_chain ( { context: retriever, # 检索相关内容 question: RunnablePassthrough() # 透传用户问题 } | prompt_template | llm | parser )这个例子中retriever从向量数据库检索相关文档RunnablePassthrough()将用户问题原样传递两者合并后送入提示词模板生成最终的LLM输入七、RunnableBranch条件分支与智能路由7.1 核心概念RunnableBranch使用条件Runnable对列表和默认分支进行初始化。对输入进行操作时选择第一个计算结果为True的条件并在输入上运行相应的Runnable。如果没有条件为True则在输入上运行默认分支。简单来说RunnableBranch就是链式if/elif/else在LangChain中的实现。7.2 基础示例from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: isinstance(x, str), lambda x: x.upper()), # 条件1: 字符串 → 转大写 (lambda x: isinstance(x, int), lambda x: x 1), # 条件2: 整数 → 加1 (lambda x: isinstance(x, float), lambda x: x * 2), # 条件3: 浮点数 → 乘2 lambda x: goodbye, # 默认分支 ) print(branch.invoke(hello)) # 输出: HELLO (字符串 → 大写) print(branch.invoke(5)) # 输出: 6 (整数 → 加1) print(branch.invoke(None)) # 输出: goodbye (无匹配 → 默认分支)7.3 智能客服路由实战在实际应用中RunnableBranch非常适合做智能路由——根据用户输入类型选择不同的处理链def detect_topic(input_text): 根据输入文本检测主题类型 if 天气 in input_text: return weather elif 新闻 in input_text: return news else: return general branch_chain RunnableBranch( (lambda x: detect_topic(x[input]) weather, weather_chain), (lambda x: detect_topic(x[input]) news, news_chain), general_chain, # 默认分支 ) result branch_chain.invoke({input: 今天天气怎么样})7.4 执行机制要点顺序评估条件按照声明顺序依次检查短路执行第一个满足条件的分支被执行后后续条件不再检查必须有默认分支如果所有条件均不满足且未设置默认分支系统将抛出异常八、RunnableWithFallbacks容错与回退8.1 核心概念在生产环境中外部API如语言模型的API可能会经历性能下降甚至停机。在这些情况下拥有一个回退Runnable非常有用。RunnableWithFallbacks使得Runnable失败后可以回退到其他Runnable。回退按顺序尝试直到某个成功或全部失败。8.2 基础用法import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.runnables import RunnableLambda llm init_chat_model( modelopenai/gpt-oss-20b:free, model_provideropenai, base_urlhttps://openrouter.ai/api/v1, api_keyos.getenv(OPENROUTER_API_KEY), ) # 原始链可能会失败 chain PromptTemplate.from_template(hello) | llm # 添加回退机制 chain_with_fallback chain.with_fallbacks([ RunnableLambda(lambda x: 抱歉服务暂时不可用) ]) # 示例提示词模板中没有需要填充的变量会报错 # with_fallbacks会自动捕获异常并执行回退 result chain_with_fallback.invoke(1) print(result) # 输出: 抱歉服务暂时不可用8.3 更优雅的用法按需回退你可以通过with_fallbacks方法的exceptions_to_handle参数指定哪些异常触发回退# 只在特定异常如API限流时触发回退 chain_with_fallback chain.with_fallbacks( fallbacks[fallback_chain], exceptions_to_handle(RateLimitError, APIConnectionError) )8.4 实用场景多模型备份一个典型的使用场景是模型降级——当主力模型如GPT-4因限流不可用时自动切换到备用模型如GPT-3.5-turbo# 主力模型链 primary_chain prompt | gpt4_model | parser # 备用模型链 fallback_chain prompt | gpt35_model | parser # 构建带回退的生产级链 robust_chain primary_chain.with_fallbacks([fallback_chain]) 技术提示在流式处理中回退只会在流创建阶段的失败时触发。流已经开始后发生的错误不会触发回退机制。总结LCEL和Runnable系列组件构成了LangChain工作流的基石通过本文的学习我们可以总结出以下要点组件一句话总结典型场景Runnable一切可运行单元的标准化接口统一invoke/batch/stream调用LCEL声明式管道语法用|简化链式组合RunnableSequence顺序执行流水线提示词→模型→解析器RunnableParallel并行执行多个任务多任务并发、多模型对比RunnableLambda自定义函数转换数据清洗、格式转换RunnablePassthrough数据透传与上下文保留RAG中的问题透传RunnableBranch条件分支路由智能客服、多意图识别RunnableWithFallbacks容错与回退机制模型降级、高可用保障最佳实践建议优先使用字典语法创建RunnableParallel代码更简洁且享受LangChain的隐式转换机制合理利用批处理使用batch()替代循环调用invoke()性能提升显著为生产系统添加回退LLM API不可预测with_fallbacks是保障可用性的最佳实践RAG场景善用RunnablePassthrough透传用户输入与检索结果完美融合注意RunnableLambda的流式限制如需流式输出自定义逻辑使用RunnableGenerator掌握这些LCEL核心组件你就掌握了LangChain工作流构建的精髓。无论是简单的问答系统还是复杂的Agent应用都可以通过像搭积木一样灵活组合的方式快速构建出生产级的AI应用。