告别嵌套调用的混乱一文吃透 LangChain 表达式语言一、前言为什么你一定要学 LCEL很多同学刚开始写 LangChain 代码时都是这样一步步手动调用的pythonprompt prompt_template.invoke({topic: LangChain}) response model.invoke(prompt) result parser.invoke(response)这种写法写短链还好一旦业务复杂起来——要加检索、加数据处理、加分支判断、加并行调用——代码会迅速膨胀成“意大利面”改一个环节要翻半天。更头疼的是想加流式输出每个组件都要改一遍代码想批量跑多个输入得自己写循环加并发想调试看每一步输入输出得手动打日志替换组件换模型/换解析器牵一发而动全身LCEL 就是 LangChain 官方给出的标准答案。它让你用一行|管道符把所有组件串成一条完整的处理链。二、核心概念Runnable 与 LCEL2.1 Runnable统一的“可运行单元”在 LCEL 的世界里万物皆 Runnable。提示词模板、大模型、输出解析器、检索器甚至你自己写的函数——只要实现了 Runnable 接口就都是标准化的组件拥有统一的调用方法方法作用.invoke()单次执行拿到最终结果.stream()流式输出逐 token 返回.batch()批量并发执行多个输入.ainvoke()异步版本就像所有 USB 设备都能插 USB 接口一样所有 Runnable 都能用同一套规则调用、同一套方式组合。div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-runnable-concept.png altRunnable 统一接口示意图 width600/ br/ em图所有组件都实现 Runnable 统一接口/em /div2.2 LCEL用管道把组件串成流水线LCEL LangChain Expression LanguageLangChain 表达式语言它最核心的设计就是用管道符|连接各个 Runnable 组件左边的输出自动作为右边的输入。类比一下传统写法 手工搬运物料每个工位之间你自己传数据LCEL 写法 搭好一条传送带物料自动顺着流水线往后走python# 传统嵌套写法不推荐 result parser.invoke(model.invoke(prompt.invoke(data))) # LCEL 管道写法推荐✨ chain prompt | model | parser result chain.invoke(data)数据流动方向一目了然text输入数据 | ▼ Prompt 模板 格式化提示词 | ▼ 大模型 生成回复 | ▼ Output Parser解析输出 | ▼ 最终结果div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-lcel-flow.png altLCEL 数据流示意图 width700/ br/ em图LCEL 管道中的数据流向/em /div三、环境准备开始之前先安装依赖并配置模型pythonimport os from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate load_dotenv() # 初始化 DeepSeek 模型 model init_chat_model( modeldeepseek-v4-flash, model_provideropenai, base_urlos.getenv(DEEPSEEK_BASE_URL), api_keyos.getenv(DEEPSEEK_API_KEY), temperature0.5 )为了方便复用可以把模型和模板封装成工具函数python# utils/model_factory.py def get_deepSeek_model(): return init_chat_model(...) # utils/prompt_template.py def getPromptTemplate(system: str, human: str): return ChatPromptTemplate.from_messages([ (system, system), (human, human) ])四、基础用法invoke / batch / stream4.1 invoke单条调用invoke是最常用的方法处理单条输入python# 01_concept_explainer.py from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() parser StrOutputParser() template getPromptTemplate( 你是一名编程讲师擅长用简单语言解释技术概念。, 请解释下面的技术概念 概念{topic} 学习者水平{level} 要求 1. 先给一句话定义 2. 再给一个简单例子 3. 不超过 200 字 ) chain template | model | parser result chain.invoke({ topic: 什么是 RAG, level: 编程小白 }) print(result)适用场景用户提交一次问题、生成一段商品文案、分析一条评论等。4.2 batch批量处理batch可以高效处理多条输入python# 02_batch_product_selling_points.py from langchain_core.output_parsers import StrOutputParser from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() template getPromptTemplate( 你是一名电商运营请根据商品信息提炼 3 个简洁卖点。, 商品名称{product_name} 商品特点{features} 请输出 3 个卖点每个卖点不超过 20 字。 ) chain template | model | StrOutputParser() products [ {product_name: 无线静音鼠标, features: 静音按键、蓝牙连接、续航 30 天}, {product_name: 机械键盘, features: 热插拔、RGB 灯光、全键无冲}, {product_name: 便携显示器, features: 15.6 英寸、1080P、Type-C 一线连接}, ] results chain.batch(products) for product, selling_points in zip(products, results): print(f【{product[product_name]}】) print(selling_points) print(- * 50)适用场景批量处理评论、批量生成商品文案、批量分类工单等。⚠️注意批量调用会消耗更多模型额度测试时不要一次提交太多数据。可以通过config{max_concurrency: 2}控制并发数避免触发频率限制。div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-batch.png altbatch 批量处理示意图 width600/ br/ em图batch 并发处理多条输入/em /div4.3 stream流式输出stream让模型边生成边输出适合聊天、内容生成等场景python# 03_stream_study_advice.py from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() parser StrOutputParser() template getPromptTemplate( 你是一名编程学习规划老师回答要具体、可执行。, 请为下面的学习者提供学习建议 学习目标{goal} 当前基础{level} 每天时间{hours} 小时 请给出 5 条建议。 ) chain template | model | parser data { goal: 掌握 LangChain 并完成一个 RAG 项目, level: 会 Python 和 FastAPI, hours: 2 } # 流式输出逐字打印 for chunk in chain.stream(data): print(chunk, end, flushTrue)小贴士flushTrue可以让内容实时显示而不是等缓冲区满了再一次性输出。div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-stream.png altstream 流式输出示意图 width600/ br/ em图stream 逐 token 流式返回/em /div五、进阶组件5.1 RunnableLambda把普通函数加入链有时我们需要在链中插入自定义的 Python 逻辑比如清洗输入、数据转换等。RunnableLambda可以把普通函数包装成 Runnablepython#04_input_preprocessing.py from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() template getPromptTemplate( 你是一名 Python 讲师请简洁回答学生问题。, 学生的问题是{question} ) parser StrOutputParser() # 自定义清洗函数 def clean_input(data: dict) - dict: question data.get(question, ).strip() return {question: question} # 将普通函数包装成 Runnable cleaner RunnableLambda(clean_input) # 加入链中 chain cleaner | template | model | parser result chain.invoke({ question: 什么是闭包 # 注意前后有空格 }) print(result) # 空格已被自动去除关键点RunnableLambda可以包装任何普通函数也可以直接用 lambda 表达式chain prompt | model | (lambda x: {key: x})div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-runnablelambda.png altRunnableLambda 示意图 width600/ br/ em图RunnableLambda 将普通函数接入链/em /div5.2 RunnableParallel并行执行多个任务RunnableParallel可以把同一个输入同时交给多个 Runnable 并行处理python# 05_review_pipeline.py from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() parser StrOutputParser() # 三条独立的子链 template1 getPromptTemplate( 你是评论情感分析助手只回答正面、中性或负面。, 用户的评价是{review} ) template2 getPromptTemplate( 你是关键词提取助手请提取 3 个关键词用顿号分隔。, 用户的评价是{review} ) template3 getPromptTemplate( 你是电商客服请根据评论生成一段不超过 80 字的礼貌回复。, 用户的评价是{review} ) chain1 template1 | model | parser chain2 template2 | model | parser chain3 template3 | model | parser # 并行执行三条链 analysis_chain RunnableParallel( sentimentchain1, keywordschain2, replychain3 ) result analysis_chain.invoke({ review: 鼠标手感不错也很安静但是滚轮用了两周就有异响。 }) print(result) # 输出 # { # sentiment: 负面, # keywords: 手感、静音、滚轮异响, # reply: 感谢您的反馈很抱歉滚轮异响影响了使用体验…… # }核心优势三条链并发执行总耗时 ≈ 最慢那条链的时间而不是三条链时间相加。div aligncenter img srchttps://img-blog.csdnimg.cn/direct/placeholder-parallel.png altRunnableParallel 并行执行示意图 width700/ br/ em图RunnableParallel 将同一输入并行分发给多个子链/em /div小技巧在 LCEL 管道中遇到字典字面量{key: value}时LangChain 会自动将其包装成RunnableParallel。所以下面两种写法等价python# 显式写法 chain RunnableParallel(contextretriever, questionRunnablePassthrough()) | prompt | model # 字典简写自动转换为 RunnableParallel chain {context: retriever, question: RunnablePassthrough()} | prompt | model5.3 RunnablePassthrough数据透传RunnablePassthrough的作用是原样透传上游数据不做任何修改。用法一RunnablePassthrough()原样透传pythonfrom langchain_core.runnables import RunnablePassthrough # 原样透传什么也不改 chain {question: RunnablePassthrough()} | template | model | parser用法二RunnablePassthrough.assign()追加字段assign方法会保留原始所有字段同时追加新字段python# 06_RunnablePassthrough_demo.py from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableLambda from utils.model_factory import get_deepSeek_model def mock_retriever(query: str): docs [ RunnablePassthrough 用于透传上游数据可搭配 assign 新增字段。, RunnableParallel 并行执行多条子链输出字典。, LCEL 使用 | 符号串联任意 Runnable、普通函数。 ] return \n.join(docs) def clean_question(data: dict): return data[question].strip() model get_deepSeek_model() template PromptTemplate.from_template( 基于下面上下文回答用户问题 上下文{context} 用户问题{question} ) chain ( RunnablePassthrough.assign( clean_questionRunnableLambda(clean_question), contextlambda x: mock_retriever(x[question]) ) | template | model | StrOutputParser() ) result chain.invoke({question: 什么是 RunnablePassthrough }) print(result)执行过程text输入: {question: 什么是 RunnablePassthrough } │ ▼ RunnablePassthrough.assign() │ ┌───────────────┼───────────────┐ │ │ │ ▼ ▼ ▼ 保留 question 清洗生成 检索生成 clean_question context │ │ │ └───────────────┼───────────────┘ ▼ 输出: { question: 什么是 RunnablePassthrough , # 原始字段保留 clean_question: 什么是 RunnablePassthrough, # 新增 context: RunnablePassthrough 用于透传... # 新增 }高频使用场景RAG 应用并行获取检索上下文 用户原始问题数据加工保留原始输入追加清洗、摘要、翻译等衍生字段多分支并行把上游原始数据带到下游避免数据丢失六、完整对比传统写法 vs LCEL对比维度传统嵌套写法LCEL 管道写法代码行数多行分散调用一行链式组合可读性逻辑分散数据流一目了然可复用性难以复用组件可任意替换流式支持需手动改造天然支持.stream()批量处理自己写循环原生支持.batch()并行执行手动实现并发RunnableParallel一行搞定七、总结本章核心要点Runnable 是基石LangChain 所有核心组件Prompt、Model、Parser、Retriever都实现了 Runnable 接口LCEL 是语法用|管道符组合 Runnable形成处理链三大调用方法.invoke()单条处理.batch()批量处理.stream()流式输出三大进阶组件RunnableLambda包装普通函数RunnableParallel并行执行RunnablePassthrough数据透传与追加最常见的 LCEL 结构pythonchain prompt_template | model | parser result chain.invoke(data)掌握了 LCEL你就掌握了 LangChain 现代开发的精髓。从今天开始告别臃肿的嵌套调用用|管道符搭建优雅、可复用、可扩展的 AI 应用链吧