很多人在学习LangChain时，第一个卡住的地方就是LCEL（LangChain Expression Language）。恰好我也在这个坑里爬过一阵子，今天把学习笔记整理出来，聊聊LCEL是什么、为什么要用它、怎么用它把大模型应用拼装起来，以及在调试过程中踩过的那些坑。这篇文章适合两类人：一是刚开始接触LangChain、被官方文档绕晕的新手，二是已经写过一堆裸的LLM调用、想规范工程结构的开发者。
LCEL的核心很简单：用管道符|把不同的处理环节串起来，让数据像自来水一样流过各个组件。这个思路和Unix管道很像，但底层又有一整套统一的Runnable协议在支撑。你不需要定义一个又一个链类，也不用操心流式、异步和批处理这些副作用，LCEL把这些能力全部打包好了。
1. 为什么要有LCEL：从链式调用说起
1.1 传统Prompt+LLM调用的痛点
最早我写LangChain应用，基本就是“重造轮子”的状态。比如我要实现一个翻译链，需要先定义一个类，内部维护提示词模板和模型实例，再写一个run()方法接收输入、组装消息、调用模型、解析输出。整个过程也不算复杂，但当应用里有多条不同业务逻辑的链时，问题就来了：每个链的调用方式五花八门，有的返回字符串，有的返回消息对象，有的还要做后处理。如果想要流式输出，还需要立刻去翻模型厂商的SDK文档，然后再封装一层。代码里充斥着各种if和try-except，维护成本直线上升。
更麻烦的是错误处理和可观测性。一个链跑挂了，我只能靠打印日志去定位到底是在Prompt组装阶段出错，还是模型调用阶段出错，又或者是解析阶段出错。这些都是裸调用的通病，但LCEL出现后，LangChain官方把这些通用能力抽成了标准协议。
1.2 LCEL的设计目标与核心优势
LCEL的设计目标，用一句话说就是：让大模型应用的构建变成“搭积木”而非“写泥水匠”。每个积木都是一个Runnable对象，它们遵循同一个调用协议。这样你用|拼接出来的东西，本身也是一个Runnable，可以被继续拼接，也可以被独立执行。
其核心优势可以归纳成四点。第一是声明式：你告诉系统“数据从哪里流向哪里”，而不是一步步写过程代码。第二是可组合：小到单次Prompt调用，大到多路并行的复杂工作流，都能用同一种语法组织。第三是自动支持流式、异步、批处理和重试，这一点对生产环境尤为重要。第四是自带追踪：LCEL会生成调用链的事件，可以接入监控平台，每条链的输入、输出、耗时都一目了然。
2. LCEL核心语法与基础组件
2.1 管道操作符 | 的底层逻辑
LCEL里最显眼的符号就是|。在Python里，|是运算符重载的产物。LangChain给Runnable对象实现了__or__方法，让它能返回一个包含左边和右边两个组件的RunnableSequence对象。这个Sequence内部维护一个有序列表，运行时从上往下依次调用每个step，把上一级的输出作为下一级的输入。
举个例子：
chain = prompt | model | parser这行代码创建了一个链条，执行chain.invoke({"topic": "AI"})时，实际顺序是：
- 把
{"topic": "AI"}传给prompt，得到PromptValue； - 把PromptValue传给model，得到AIMessage；
- 把AIMessage传给parser，得到纯文本结果。
这里最容易踩的坑是输入输出格式不匹配。左边的组件输出的类型，必须和右边组件期望的输入类型对得上。比如PromptTemplate需要dict输入，但如果你在它前面接了一个RunnableLambda返回的是字符串，就会报错。理解这条数据流，比背API重要得多。
2.2 Runnable接口：LCEL的通用协议
LCEL中所有组件都被包装成Runnable对象，这是整套体系的基石。一个Runnable必须实现若干方法，最关键的就是invoke、batch、stream，以及对应的异步版本ainvoke、abatch、astream。
invoke(input)：同步执行单次调用，返回结果。batch(inputs)：批量执行，传入一组输入，返回一组结果。stream(input)：流式执行，返回一个生成器，可以逐个拿到增量数据。ainvoke(input)/abatch(inputs)/astream(input)：异步版本。
这意味着，只要你把所有业务逻辑封装成Runnable，它就能天然获得流式、异步和批处理能力，而不用自己从头实现。还有一个容易被忽略的点：Runnable对象自带with_config、with_fallbacks和with_retry方法，这些是生产级应用里非常重要的工程能力。比如你希望某个模型调用失败时自动切换备用模型，直接在原链上with_fallbacks(...)就可以了。
2.3 常用Runnable组件：PromptTemplate、ChatModel、OutputParser
实际开发中，你会反复用到几类Runnable组件。
第一类是PromptTemplate。它的作用是把dict类型的变量渲染成模板，输出通常是PromptValue。你可以在模板里写变量占位符，然后在运行时填充。它的子类ChatPromptTemplate专门用来组织多轮对话消息列表，支持SystemMessage、HumanMessage等角色。
第二类是各种模型封装，比如ChatOpenAI、ChatAnthropic，这些都被做成了Runnable。它们接收消息列表，输出AIMessage对象。注意不要直接拿裸模型跑，最好让它作为链的中间环节，方便后续解析和扩展。
第三类是输出解析器，常见的有StrOutputParser（直接把内容转成字符串）和PydanticOutputParser（把模型输出解析成结构化数据）。它们负责把模型返回的信息“洗”成你真正需要的格式。
除了上面三类，还有两个常被忽略但极其好用的通用组件：RunnableLambda可以把任意Python函数包装成Runnable；RunnablePassthrough则原样传递输入，通常用来在链中插入“侧路”或调试节点。
3. 实操：用LCEL搭建一个可运行的问答链
3.1 最小可用示例：Prompt+LLM+解析器
直接看代码。下面是一个用LCEL构建的最小问答链：
from langchain.prompts import ChatPromptTemplate from langchain.chat_models import ChatOpenAI from langchain.schema.output_parser import StrOutputParser prompt = ChatPromptTemplate.from_template( "你是一名耐心、专业的助手。请用通俗的语言回答用户问题。\n问题：{question}" ) model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) parser = StrOutputParser() chain = prompt | model | parser result = chain.invoke({"question": "什么是LCEL"}) print(result)这个例子里的知识链就串联起来了。执行后你会得到一个纯字符串答案，而不是AIMessage对象。整个过程清晰又省事，不用自己写消息构造和输出提取。
值得注意的是，chain.invoke的参数必须和PromptTemplate里的变量对齐。如果模板用了{question}，你却传了一个不带这个键的dict，甚至直接传字符串，就会立刻报错。这是新手最常犯的问题。
3.2 加入记忆与上下文
聊天场景里，你必须记住历史对话。裸调用时我会自己维护一个消息列表，再每轮拼进去。LCEL下可以用RunnableWithMessageHistory做包装，自动完成历史消息的存储和注入。
from langchain.memory import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] chain = prompt | model | parser with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="question", history_messages_key="history", ) result = with_history.invoke( {"question": "我叫小明"}, config={"configurable": {"session_id": "user-1"}}, ) print(result)这里有两个细节要注意。第一，get_session_history返回的是一个ChatMessageHistory对象，它是消息历史的容器。第二，config里的session_id是区分不同会话的关键。你既可以按用户ID分，也可以按设备ID分，完全取决于业务设计。如果忽略history_messages_key的配置，模板里的{history}就渲染不了。
3.3 并行与分支：LCEL的组合能力
LCEL真正强大的是它的组合能力。假设一个客服机器人需要同时做“意图识别”和“情绪分析”，再汇总结果。如果串行执行，会很慢；如果用RunnableParallel，就能并行跑。
from langchain_core.runnables import RunnableParallel intent_chain = intent_prompt | model | parser emotion_chain = emotion_prompt | model | parser parallel_chain = RunnableParallel( intent=intent_chain, emotion=emotion_chain ) full_chain = ( {"question": lambda x: x["question"]} | parallel_chain | answer_prompt | model | parser ) result = full_chain.invoke({"question": "我想退货，气死我了"})这段代码里，前面加了一个映射步骤，保证进入parallel的数据格式是字典。parallel_chain会同时执行两个子链，产出{"intent": ..., "emotion": ...}，再交给后面的汇总提示词。本质上，RunnableParallel内部用的是线程池，所以多个分支之间是并发而非阻塞串行。如果你的分支里调用了资源竞争的操作，就得特别注意线程安全。
4. 进阶功能：流式输出、异步与回调
4.1 流式输出stream的实现机制
LLM应用里，流式输出几乎是刚需。用户在等待答案时，如果屏幕上是一段一段蹦出来的文字，体验感会好很多。LCEL里调用链的stream()方法，能拿到逐块生成的结果。
for chunk in chain.stream({"question": "讲一个关于编程的笑话"}): print(chunk, end="", flush=True)注意，这里的chain是一个LCEL链，模型本身必须支持流式输出。底层实现是LCEL从最后一个组件反向接管生成器：parser接收AIMessage的流式块，然后逐一输出最终文本块。如果你自定义了RunnableLambda，记得也要让函数返回迭代器或确保它能消费生成器，否则流式会被截断。
4.2 异步运行async与并行调用
现代Web框架普遍依赖异步机制。LCEL提供了完整的异步接口，你可以用ainvoke、astream把链嵌入到异步应用里。
async def handle_question(question: str) -> str: result = await chain.ainvoke({"question": question}) return result # 在FastAPI路由里直接调用当你需要同时处理一批输入时，batch方法非常高效。它内部会对输入列表做并发调度，而不是顺序循环。实测下来，用batch处理8条输入，耗时通常仅为串行调用的1/3到1/5。批量并发时要注意：如果你在代码里对全局变量做了写操作，并发下极可能产生数据竞争。
4.3 回调与追踪：让链可视化
LCEL的每个步骤都会触发回调事件。你可以实现一个自定义回调处理器，在链的各个生命周期打日志或上报指标。
from langchain_core.callbacks import BaseCallbackHandler class SimpleHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f"开始执行: {serialized['name']}, 输入: {inputs}") def on_chain_end(self, outputs, **kwargs): print(f"执行结束: {outputs}")把回调挂到链上：
chain = prompt | model | parser handler = SimpleHandler() chain.invoke({"question": "你好"}, config={"callbacks": [handler]})如果项目规模大，LangChain自家的监控平台也是不错的选择。它会自动把LCEL的每一步链路、token用量和延迟时间记录下来。在本地调试时，我喜欢用with_config({"run_name": "my_chain"})给链命名，这样日志里就能清晰区分不同业务链。
5. 常见问题与排查技巧实录
5.1 接口不匹配问题
这一块是踩坑重灾区。LCEL里最典型的报错就是“Expected a Runnable or a callable”。这通常是因为你把一个非Runnable对象（比如一个字符串）放到了|后面。还有一种是dict和字符串混用，比如某个组件需要dict输入，你却给了字符串。
我自己总结了一张排查表：
| 常见现象 | 可能原因 | 解决办法 |
|---|---|---|
| 调用invoke时提示缺少输入键 | 传入的dict和PromptTemplate变量不一致 | 打印模板，确认变量名，检查大小写 |
| 链中间某个节点输出字符串，下个节点要dict | 中间节点没有返回正确结构 | 插入RunnablePassthrough或映射dict |
| 输出是AIMessage，不是文本 | 输出解析器没加或加错了 | 在链尾部加上StrOutputParser |
| 异步环境下直接用invoke | 没有用异步接口 | 换成ainvoke或astream |
| 流式什么都没有输出 | 某些自定义函数没有消费生成器 | 确保自定义函数是生成器友好的 |
5.2 并行执行时的线程安全
RunnableParallel内部是并行执行多个子链的，线程模型会共享内存。如果你在自定义的RunnableLambda里修改模块级的全局变量，就会出现数据错乱，甚至偶发报错。
我遇到过一个问题：并行调用两个子链，都要往同一个列表里追加结果，结果最后列表里多了重复项。原因就是两个线程同时写同一个内存地址。解决办法是避免共享可变状态，或者把需要共享的数据存到上下文中，用RunnableConfig传递，不要直接引用全局变量。如果实在需要全局缓存，请加锁。
5.3 调试LCEL链的实用方法
LCEL写起来很快，但调试时往往无从下手。我的习惯是在关键位置插入RunnableLambda打印中间结果：
def debug_question(x): print(f"当前输入: {x}") return x chain = debug_q | prompt | model | parser用内部函数包装，不影响数据流转，还能实时观察。另一个技巧是利用RunnablePassthrough.assign()在链中附加调试信息：
from langchain_core.runnables import RunnablePassthrough chain = ( RunnablePassthrough.assign( rendered=prompt ) | model | parser )这样做，最后result里不会出现rendered，但链的执行过程能看到生成后的完整提示词。另外，如果你已经在接LangSmith，直接在平台里看每一步的耗时和输入输出比本地日志更直观。
结尾
学LCEL最大的体会是：别被“表达式语言”这个名字吓到，它本质上就是一个用管道符组织Runnable对象的语法糖，背后是LangChain对整个AI应用工程化的沉淀。我刚开始学时，也经历过对着官方文档发呆、按部就班抄代码却跑不通的时期。
后来我调整了思路，先用最小示例把prompt | model | parser跑通，再逐个加记忆、并行、流式。每加一个功能就打印一次中间结果，很快就把整套模型串起来了。如果你也在学，建议找一个真实的业务场景（比如客服助手或文档问答）练手，LCEL的模块化设计会让你越写越顺。
最后分享一个小技巧：RunnableParallel是提升链性能的利器，但别忘了在这种组合链上增加超时控制，防止某个分支卡死，拖累整个请求。用with_config({"timeout": 10})可以给链统一设置超时时间，这个细节我在上线前才想到，差点就出事故了。
