很多人在学习LangChain时第一个卡住的地方就是LCELLangChain Expression Language。恰好我也在这个坑里爬过一阵子今天把学习笔记整理出来聊聊LCEL是什么、为什么要用它、怎么用它把大模型应用拼装起来以及在调试过程中踩过的那些坑。这篇文章适合两类人一是刚开始接触LangChain、被官方文档绕晕的新手二是已经写过一堆裸的LLM调用、想规范工程结构的开发者。LCEL的核心很简单用管道符|把不同的处理环节串起来让数据像自来水一样流过各个组件。这个思路和Unix管道很像但底层又有一整套统一的Runnable协议在支撑。你不需要定义一个又一个链类也不用操心流式、异步和批处理这些副作用LCEL把这些能力全部打包好了。1. 为什么要有LCEL从链式调用说起1.1 传统PromptLLM调用的痛点最早我写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函数包装成RunnableRunnablePassthrough则原样传递输入通常用来在链中插入“侧路”或调试节点。3. 实操用LCEL搭建一个可运行的问答链3.1 最小可用示例PromptLLM解析器直接看代码。下面是一个用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(modelgpt-4o-mini, temperature0.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_keyquestion, history_messages_keyhistory, ) 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( intentintent_chain, emotionemotion_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, flushTrue)注意这里的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( renderedprompt ) | model | parser )这样做最后result里不会出现rendered但链的执行过程能看到生成后的完整提示词。另外如果你已经在接LangSmith直接在平台里看每一步的耗时和输入输出比本地日志更直观。结尾学LCEL最大的体会是别被“表达式语言”这个名字吓到它本质上就是一个用管道符组织Runnable对象的语法糖背后是LangChain对整个AI应用工程化的沉淀。我刚开始学时也经历过对着官方文档发呆、按部就班抄代码却跑不通的时期。后来我调整了思路先用最小示例把prompt | model | parser跑通再逐个加记忆、并行、流式。每加一个功能就打印一次中间结果很快就把整套模型串起来了。如果你也在学建议找一个真实的业务场景比如客服助手或文档问答练手LCEL的模块化设计会让你越写越顺。最后分享一个小技巧RunnableParallel是提升链性能的利器但别忘了在这种组合链上增加超时控制防止某个分支卡死拖累整个请求。用with_config({timeout: 10})可以给链统一设置超时时间这个细节我在上线前才想到差点就出事故了。