1. 为什么LCEL值得专门花时间学
1.1 从传统调用方式到LCEL的痛点转变
先说个背景。我最初用LangChain的时候，写一条链路是这样的：手动定义一个Prompt模板，拿去生成文本，塞给模型调用，再用一个函数去解析模型返回的内容。如果中间要加个记忆组件，或者要插入一个检索步骤，逻辑就开始绕了。代码里到处都是if判断和临时变量，链路越长越难维护。尤其是当你需要在多个不同场景里复用同一套逻辑时，复制粘贴就成了常态，改一版就要同步改好几个地方，非常痛苦。
LCEL（LangChain Expression Language）给我的感觉是，它把上面那套手工拼接的操作变成了一种声明式的表达式。你用管道符|把组件连起来，整条调用链路就是一条清晰的数据流。这条数据流可以被保存、被复用，甚至可以被内省（introspect），你可以清楚地看到每一步的输入和输出。对于经常跟LLM打交道的开发者来说，这种变化有点像从过程式调用切换到管道式组装，代码量和心智负担都降了不少。
当然，我一开始也觉得LCEL是不是为了炫技搞出来的语法糖。实际用下来发现，它解决的核心问题不是写法好不好看，而是链路的组合能力和可维护性。LCEL把每个环节都抽象成Runnable接口，只要实现这个接口的东西就能被拼接。换句话说，你可以把模型、提示模板、输出解析器、外部工具调用，甚至数据库查询都统一成同一种处理单元，随意组合。
1.2 LCEL在LangChain整体架构里的位置
LangChain本身的组件很多：模型、提示模板、记忆、检索器、Agent、工具调用、回调系统等。LCEL并不是其中某一个功能模块，而是把它们粘合在一起的骨架。我自己的理解是，如果没有LCEL，LangChain更像一个工具集，你想要哪块就单独拿哪块用；有了LCEL之后，它才真正成为一个可编排的框架。
这里有个关键点：LangChain里几乎所有核心组件都实现了Runnable协议。也就是说，提示模板是可运行的，模型是可运行的，输出解析器是可运行的，检索器也是可运行的。LCEL的管道符本质上是在把不同Runnable的输入输出首尾对接起来。这个设计带来的最大好处是，你不需要关心每个组件内部怎么实现，只需要约定好数据格式就行。
我记得有一次要改动某个内部项目的调用链路，原先是先做文档检索，再拼提示词，再调用模型，最后解析结果。如果用传统函数式写法，改动任何一个环节的返回格式，后续所有环节都要跟着调整。但用LCEL后，每个环节在管道里只是一个小节点，节点的输入输出格式清晰，想替换一个节点只需要保证新的节点接得住上一步的返回数据。这种松耦合的感觉在项目迭代中帮了大忙。
2. 从零开始拆解LCEL的核心语法
2.1 一条最小管道里发生了什么
入门LCEL的第一行示例几乎都是这样的：
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("用一句话解释{concept}是什么") model = ChatOpenAI(model="gpt-4o-mini", temperature=0) parser = StrOutputParser() chain = prompt | model | parser result = chain.invoke({"concept": "LCEL"}) print(result)别看这段代码短，里面有三层东西值得展开。
第一层是Pipeline机制。prompt | model之所以能成立，是因为左侧的prompt对象实现了__or__方法。严格说这里其实是LangChain针对Runnable对象的运算符重载，不是为了Python语言的接口规范，就是为了你写起来顺手。管道执行时，invoke传入的字典先被prompt接收，prompt用字典里的concept变量渲染出一个消息列表，然后把这个列表交给model调用。model返回一个AIMessage对象，再传给parser，parser从消息对象里抽出字符串内容。
第二层是类型流转。很多人一开始容易忽略：管道中间流转的并不是纯字符串。prompt的输出是消息列表，model的输出是消息对象，parser的输出才是干净字符串。如果你自己实现一个自定义的Runnable插在管道中间，就必须清楚上一个节点给自己传进来的是什么类型、自己要传给下一个节点的又是什么类型。这是排查问题的关键。展开说，ChatPromptTemplate返回的是一个包含HumanMessage的列表，ChatOpenAI的invoke接收消息列表并返回AIMessage，StrOutputParser接收任何带content属性的消息对象，解析出文本。这三个组件的类型是天然吻合的。
第三层是惰性执行。管道构建时并不实际调用任何东西，只是组装结构和验证部分配置。一旦调用了invoke，数据才会真正流起来。这个设计的妙处在于，你可以把管道当成普通对象到处传递、组合、条件分支，完全不会有副作用。我在实际项目里经常把定义好的chain作为参数传给函数，或者直接嵌入另一个更大的管道中，非常灵活。
2.2 输入格式与偏函数的处理细节
LCEL的管道第一个节点如果是PromptTemplate，那么invoke时需要传入的是包含所有模板变量的字典。这是初学者最容易踩的坑，比如模板里写了两个变量，调用时只传了一个，就会报KeyError。
解决办法比较常见的是用partial预填字段。假设你的提示模板是：
prompt = ChatPromptTemplate.from_template( "请用{style}的风格，为{product}写一句广告语" )但你在业务中很多地方都用同一个style，那就可以这样固定：
prompt_with_style = prompt.partial(style="简洁有力") chain = prompt_with_style | model | parser chain.invoke({"product": "智能水杯"})注意partial仍然返回一个Runnable对象，所以管道里可以继续拼接。这跟Python标准库functools.partial的思路类似，都是在调用前预先绑定一部分参数。用在不同场景下复用同一条提示模板，比如有的场景固定要求语气轻松，有的场景要求正式，你就可以为每个场景配置一个partial后的变体，而不用重新定义模板。
另一个常见输入场景是模型输入直接就是字符串。如果管道起始节点不是PromptTemplate而是Model本身，比如：
chain = model | parser chain.invoke("你好")这时候invoke参数直接给字符串也是可以的，因为多数模型Runnable支持字符串输入。但如果后续节点期望接收字典，你就要注意转换。实践中我一般建议还是保留一个PromptTemplate作为入口，哪怕它只是原样透传，因为这样链条的数据契约最清晰。
2.3 分支、合并与并行：结构化管道的核心
LCEL真正变强的时刻，是你开始用RunnableParallel、RunnableBranch、RunnablePassthrough组合复杂逻辑的时候。前面说的小管道只是一条直线，但实际业务里链路往往有分支、有合并、有条件跳转。
RunnableParallel的语义是：输入同一个数据，同时发给多个子Runnable，把各自结果合并成一个字典。举个例子：
from langchain_core.runnables import RunnableParallel chain = RunnableParallel( task1=prompt1 | model1 | parser, task2=prompt2 | model2 | parser ) result = chain.invoke({"topic": "人工智能"}) # result = {"task1": "...", "task2": "..."}这个能力特别适合"同一份资料，做两个不同角度的分析"这类场景。两个子管道接收同样的初始输入，并行执行，最后在一个字典里汇总结果。关键是，其中一个子管道还能继续作为当前管道的一部分往下流转，比如再把两个结果拼接成一个汇总提示词传给下一轮模型。
RunnablePassthrough的用途更隐蔽，也更实用。它的行为是"不做任何事，把输入原样传下去，并顺手把指定字段也带进字典"。当你配合RunnableParallel用时，它经常被用来保留原始输入。比如你想让模型回答问题时既基于检索到的上下文，又保留原始问题，就可以这样：
from langchain_core.runnables import RunnableParallel, RunnablePassthrough retrieval_chain = ( RunnableParallel( context=retriever, question=RunnablePassthrough() ) | prompt | model | parser )这里retriever接收原始question字符串去检索文档，RunnablePassthrough把question原封不动放进字典，两者合并后正好组成prompt需要的两个变量。这是LCEL里非常经典的一个写法，代码既短又直观。
RunnableBranch则实现条件路由，用法类似elif链：
from langchain_core.runnables import RunnableBranch branch = RunnableBranch( (lambda x: len(x["question"]) < 10, short_chain), (lambda x: len(x["question"]) < 50, medium_chain), long_chain )每个分支由条件和Runnable组成，按顺序判断，先命中的就用哪个。注意最后一个兜底分支不写条件。这种结构在聊天机器人意图分流时很常用，比在业务代码里堆if要整洁得多。
3. 实战：用LCEL搭建一个可用的问答链路
3.1 环境准备与最小可运行示例
开始写一个稍微完整一点的例子。假设我们要做一个基于文档检索的问答系统，用户输入一个问题，系统先从文档库里找出相关片段，再让模型基于片段回答。这是RAG最典型的形式，也最能体现LCEL的组合优势。
环境上我用的依赖版本大概是：
pip install langchain-core langchain-openai langchain-community如果想要本地向量检索方便一点，可以再装langchain-chroma和langchain-text-splitters。我自己平时会把这些依赖写进requirements.txt，避免项目换机器后版本漂移。
先说检索侧。一个经典的在线示例往往用到内置的文档加载器和向量库，但为了专注LCEL本身，我这里用一个比较简化的写法，用Chroma作为向量存储：
from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma loader = TextLoader("./knowledge.txt") documents = loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50) chunks = splitter.split_documents(documents) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") db = Chroma.from_documents(chunks, embeddings) retriever = db.as_retriever(search_kwargs={"k": 3})这段代码里有个容易忽略的细节：RecursiveCharacterTextSplitter的chunk_overlap参数。很多人不设这个值，导致一个问题被切断后检索不到完整语义。重叠区域一般设为块大小的10%到20%，这样上下文衔接自然一些。chunk_size也不能随便设置，它取决于模型窗口和业务问题的粒度，我这边先用300做演示，生产环境一般要针对语料做调节分析。
3.2 完整的RAG链路与并行传递的关键
接下来把检索和生成连接起来。这是整个LCEL链路里最容易让人困惑的地方：retriever的输入是什么？prompt需要的输入是什么？怎么让二者对齐？
我用自己的写法来拆解：
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel, RunnablePassthrough system_prompt = ( "你是智能助手。请基于以下资料回答问题，" "如果资料中没有答案，就如实说明。\n\n" "资料：\n{context}\n\n" "问题：{question}\n" "回答：" ) prompt = ChatPromptTemplate.from_template(system_prompt) model = ChatOpenAI(model="gpt-4o-mini", temperature=0) parser = StrOutputParser() chain = ( RunnableParallel( {"context": retriever, "question": RunnablePassthrough()} ) | prompt | model | parser ) answer = chain.invoke("这份文档里提到了哪些技术栈？")重点看RunnableParallel这一步。初始输入是一段字符串，即用户问题。retriever会接收这个字符串并调用相似度搜索，返回拼接好的文档片段；RunnablePassthrough会把原始问题原样保留。合并后的字典把context和question两个变量喂给prompt模板，后面就交给你选的模型了。
这里必须提一个很常见的坑：如果RunnableParallel里没有RunnablePassthrough来携带原始问题，那么prompt的question变量就会缺失。很多人看到报错"Missing required input variables"，第一反应是检查模板，实际上往往是前面的并行结构漏传了字段。
然后说一下as_retriever的搜索参数。search_kwargs={"k": 3}代表取最相关的3个片段。片段太少可能信息不全，太多可能引入噪音。我在项目里维护过一个脚本，专门对k取2、3、4、5做对比，最后根据回答准确性选值。这种参数调优没什么捷径，就是结合你自己的语料反复测。
3.3 流式输出与异步：生产环境的关键一环
很多初学者把LCEL跑通了就收工，但到了生产环境，交互体验和并发能力都得另算。好消息是，LCEL对这两种场景几乎不用改管道结构，只需要换调用方法。
流式输出用stream方法，而异步用ainvoke、astream。以我上面那个RAG链路为例：
async def async_answer(question: str): async for chunk in chain.astream(question): print(chunk, end="", flush=True)这段代码里的astream同时做了两件事：一是异步执行，二是逐块产出结果。你写Web接口时可以直接把异步流式响应接给前端，用户能实时看到生成的文字。如果用传统同步方式一次性返回，大模型生成时间长的时候，前端很容易觉得像卡死了。
流式输出能够工作，要感谢管道里每个环节都实现了stream接口。但有个细节要注意：如果你的自定义环节不实现stream，LCEL会自动退回到invoke，这会破坏整个链路的流式效果。比如你在管道里加了一个RunnableLambda做后处理，它默认不支持流式渲染子级，模型生成结果会积攒到最后一次性输出。所以我平时在后处理逻辑里尽量选用支持流式透传的组件，或者干脆让后处理步骤用astream单独拼接，避免把流式体验断掉。
异步并发这块，LCEL管道本身的线程安全性还是可以的。只要你不在多个协程之间共享可变状态，直接用同一个chain对象处理多个请求没有问题。我在实际项目中用FastAPI承载服务，路由函数里直接await chain.ainvoke(...)，压测下来稳定，没发现共享冲突。
4. 常见问题与排查技巧实录
4.1 输入变量不匹配与输出解析失败
最常遇到的报错有两类。第一类是模板变量缺失，报错信息类似于：
Exception: Prompt is missing input variables: ['question']这类问题前面提过，多半是管道上游传给prompt的字典里没有包含全部模板变量。排查方式很简单，在报错前插入一个临时调试节点，把上一步的返回值打印出来：
from langchain_core.runnables import RunnableLambda def debug_print(x): print("DEBUG:", x) return x chain = RunnableParallel( context=retriever, question=RunnablePassthrough() ) | debug_run | prompt | model | parser注意这里要确保debug_print返回原值，否则下游数据就断了。实际项目里我不会长期保留这种打印，但排查时它比反复猜哪里少了字段要高效得多。
第二类是模型输出的格式与预期不符。比如你写了一个PydanticOutputParser来解析结构化结果，但模型返回了多余的文字说明，导致解析器直接抛错。这种问题常见于小模型或者提示词不够强的时候。我建议把输出解析器的格式说明完整写进提示词里，比如要求只输出JSON，并且用format_instructions自动注入。LCEL中你可以把解析器和提示词拼接：
from langchain_core.output_parsers import PydanticOutputParser from langchain_core.pydantic_v1 import BaseModel, Field class Movie(BaseModel): name: str = Field(description="电影名称") rating: float = Field(description="评分") parser = PydanticOutputParser(pydantic_object=Movie) prompt = ChatPromptTemplate.from_template( "根据输入内容提取电影信息。\n输入：{input}\n{format_instructions}" ).partial(format_instructions=parser.get_format_instructions()) chain = prompt | model | parser这里把format_instructions用partial固定住，模板始终携带要求模型只输出结构化JSON的指令。实测下来，配合temperature调低，解析失败率大幅度下降。
4.2 并行分支的数据依赖与执行机制
用RunnableParallel时，不少朋友以为它真的是多线程并行执行。严格意义上说，LCEL对并行任务的实现方式在同步场景下并没有说一定使用多线程，具体取决于底层执行器是否配置。我的经验是，如果你想让多个独立分支真正并发执行以获得性能提升，可以配合RunnableParallel的并发执行配置，比如在组件里显式设置executor。不过大多数应用场景下，并发收益不是决定性的，更重要的是先保证每个分支自身逻辑正确。
分支之间还有一个容易让新手懵的点：一个分支想使用另一个分支的结果。RunnableParallel里各分支是接收同一份输入并行执行的，天然不能访问彼此的输出。很多人会尝试在分支A的lambda里直接引用分支B，这在构建阶段就埋下了隐患。正确的做法是把依赖关系变成链式结构：
first_step = RunnableParallel(result1=chain_a) second_step = RunnableParallel( result2=chain_b, result1_plus=first_step | RunnableLambda(...) )或者更直接的，如果你需要把两个分支结果合并后一起给下一个模型，那就在分支结束后加一个聚合节点：
from langchain_core.runnables import RunnableLambda def merge(context, generated): return f"根据以下资料：{context}，模型生成了：{generated}，请进行总结。" final_chain = RunnableParallel( context=retriever, generated=generation_chain ) | RunnableLambda(lambda x: merge(x["context"], x["generated"])) | model | parser这个模式很好用，整体结构还是LCEL的风格，但数据依赖逻辑则显式地表达出来了。
4.3 调试与观测手段：从打印到追踪
LCEL管道的黑盒感其实是可以通过回调机制打破的。LangChain提供了callbacks参数，可以在invoke时传入。常用的是把所有中间步骤的回调事件打印出来：
from langchain_core.callbacks import StdOutCallbackHandler handler = StdOutCallbackHandler() chain.invoke(input_data, config={"callbacks": [handler]})这样模型调用的请求、响应、耗时等关键事件都会输出到控制台。比起自己在每个节点插打印，回调的方式干净得多，而且不会干扰管道结构。
如果是更大的生产系统，建议接入LangChain生态里自带的可观测性平台，或者至少把trace输出到本地。你可以在环境变量里开启trace功能，这样每次调用的完整链路都会被记录下来，包括检索到的片段内容、模型的提示词、原始输出等。定位问题的时候直接看trace，比靠日志盲猜高效太多。
至于性能问题，我记得有一次一个管道在数据量大时响应慢，排查了半天发现是检索器每次都在内存里去重排序，重复计算严重。后来把向量检索做了持久化缓存，响应时间立刻降了下来。这类问题跟LCEL本身关系不大，更多是具体组件的实现设计，但LCEL把所有节点串成一体的好处就在这里：任何一步慢，你都能在链路观测里一眼定位。
5. 实操心得与后续建议
我花了一段时间把LCEL的文档翻完，又在几个项目里反复折腾之后，最大的感受是它不值得被当作一个"新语法"去背，而是应该当成一种组织LLM应用的方式去理解。你写一百个不同的prompt模板和模型调用，不如先把一条管道的流转方式学透，后面的组合能力都是水到渠成的事。
给正准备上手的朋友一个建议：先别急着套高大上的Agent框架，从一行prompt | model | parser开始，体会一下什么叫做数据在管道里流。然后刻意练习把RunnableParallel和RunnablePassthrough用到一个真实场景里，比如做一次带检索的问答。走过这一步，LCEL的大部分概念就都有了落地的载体。
我个人踩坑最多的地方反而是"过度设计"。一开始什么链路都想用RunnableBranch和并行结构，代码是炫了，但调试难度直线上升。后来学乖了，只在逻辑确实需要分支或同时取多路数据时才使用并行结构，其他时候老老实实写直线管道。LCEL的优点是表达能力丰富，但不是代码越复杂越好，够用且清晰才是最关键的。
最后提一句，这个内容后续还可以怎么变？如果你对LCEL已经有感觉了，可以试试把自定义的工具调用、动态路由或者多轮记忆接入管道里，每加一块都会加深对Runnable协议的理解。慢慢你会发现，LangChain里的大部分复杂功能，本质上都是在用LCEL做组合，理解它之后看别的源码会轻松很多。
