LangChain 真正让我觉得“顺手”的转折点，是彻底理解 LCEL 的那一天。LCEL 的全称是 LangChain Expression Language，一套用来组合大模型应用组件的表达式语法。我在连续做了几个基于 LangChain 的文本处理小项目之后，越来越确定，LCEL 不是简单的语法糖，而是整个框架里最值得先花时间摸透的设计。这篇笔记不打算讲什么高深理论，就是记录我从“用 Chain 类硬撑着写流程”到“用 LCEL 重新梳理所有链路”的过程。适合正在用 LangChain 但每次自定义流程都靠手工拼代码的人，也适合被各种 Chain 名称绕晕的初学者。你读完应该能搞清楚 LCEL 怎么用、为什么好用、出问题的时候又该怎么排查。
1. 为什么要有LCEL：它治好了我的手工拼装恐惧症
1.1 没有LCEL之前，一条链是怎么拼出来的
早期常见的做法，就是老老实实手工调：先执行prompt.format(...)，把格式化后的字符串丢给模型接口，拿到返回值之后自己再解析、校验、过滤。如果中间还要记录日志、处理超时、判断要不要重试，那就只能在函数里不断叠加if/else和try/except。一两步的时候很清晰，一旦超过三步，这段逻辑就开始膨胀。
当时用 LangChain 的LLMChain也舒服不到哪去。这个类把“模板、模型、输出解析”封装成了三步走，但它是典型的面向对象继承风格，想加一个前置处理步骤，常常得再套一层子类，想在某个环节插入调试打印，又得改结构。更麻烦的是，多个链之间要共享某些步骤时，往往只能复制代码过去改。整个项目里的流程逻辑分散在各处，听起来都有点“古典”。
其实最让人难受的不是代码量大，而是没有一种统一的抽象来管理Component之间怎么连接。每个组件只负责一件事，一旦需要两件事以上，你就要写胶水代码。真正难维护的，恰恰是这些胶水代码：它不业务、不模型、不逻辑，只是用来搬运数据，但出错和难读的往往就是它们。LCEL 出现之后，我最大的感受就是胶水代码被语法本身替代了，你用一条管道符号，就把组件的输入输出接在了一起。
1.2 LCEL的立意：把“流程”变成“表达式”
LCEL 的核心设计很简单：它定义了一个标准接口，叫作 Runnable，任何组件只要实现了这个接口，就可以用竖线操作符|来组合。于是，一条链路可以写成像下面这样直观的表达式：
chain = prompt | llm | output_parser这个写法很像 Unix 管道：上一步的输出，自动变成下一步的输入，数据像水流一样从左往右走。你不再需要手动把上一步的结果取出来、塞给下一步，LCEL 会在运行时自动完成整个传递过程。
我早期看这个语法，觉得它只是写法简洁。后来才发现，真正的价值是组合之后的“能力自动获得”。一个被|串起来的链路，自动支持同步调用、异步调用、批量调用、流式输出、重试、回退等等，不需要每个组件自己实现这些功能。你只要保证每个零件符合 Runnable 协议，系统就会帮你把整套运行能力接好。
这和人装修时用标准水管接头很像：过去每根管子形状都不同，接起来要用生料带、胶水甚至铁丝，现在大家统一成标准接口，插上就能用。LCEL 里的 Runnable 就是这个标准接口。组件内部怎么做是它自己的事，但对外都暴露一样的组合方式，整条链路因此变得可预测、可复用、可替换。
2. LCEL的核心概念拆解：先搞懂这些再写代码
2.1 Runnable 协议的六个关键方法
LCEL 里所有组件，本质上都实现了一个 Runnable 接口。理解了这个接口，基本就理解了 LCEL。这个接口提供了六个最常用方法，前三个是同步版本，后三个是异步版本：
| 方法 | 作用 | 什么时候用 |
|---|---|---|
invoke | 单次同步调用 | 最常见，输入一个参数，返回一个结果 |
batch | 批量同步调用 | 一次性处理多个输入，自动并发 |
stream | 同步流式调用 | 一边生成一边返回结果，适合打字机效果 |
ainvoke | 单次异步调用 | 在异步事件循环里调用 |
abatch | 批量异步调用 | 异步场景下批量处理 |
astream | 异步流式调用 | 异步场景下流式输出 |
这些方法最大的价值是统一。你拿到任何一个 Runnable 组件，不管是 PromptTemplate、LLM、OutputParser 还是一个自定义函数包装器，都可以用同样的方式去调用它。这意味着，你可以先放心地组合链路，完全不用关系它内部是怎么实现的。
举个容易理解的例子：stream方法。过去要流式输出，得看模型接口支不支持、返回协议怎么处理、怎么一点点拼字符串。但在 LCEL 里，你只要对最终的 chain 调用.stream(...)就行，如果你的模型支持流式，而且后面挂的输出解析器也是流式的，那么结果会像打字一样一点点冒出来。如果某个环节不支持流式，LCEL 会退化为一次性返回，而不会报错。
2.2 Prompt模板、模型和输出解析器怎么连
一条最典型的 LCEL 链，就是“Prompt 模板 -> 模型 -> 输出解析器”。如果你从一个字符串模板开始，写出来通常是：
from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话") llm = your_chat_model # 这里换成你自己的 ChatModel 实例 chain = prompt | llm | StrOutputParser() result = chain.invoke({"topic": "程序员"}) print(result)这段代码背后发生了几件容易被忽略的事。首先，chain.invoke()的输入是一个字典，因为 PromptTemplate 需要填充topic这个变量。其次，经过prompt之后，输出已经是一个填充好的消息结构，而不是普通字符串；数据传进llm，模型返回的是一个 AIMessage 对象；最后交给StrOutputParser，它负责把 AIMessage 抽成字符串。如果你去掉最后的输出解析器，拿到的就是那个 AIMessage 对象，打印出来会带上类型标签，处理起来就没那么直接。
这个三步结构是所有 LCEL 应用的地基。你后面看到的所有花活，比如并行、分支、自定义函数、数据库缓存，本质上都是在这些基础组件之间加节点。先把这三步消化掉，后面理解什么都轻松。
2.3 RunnablePassthrough、RunnableLambda、RunnableParallel
这三个辅助类大概是我用得最多的，值得一个个说清楚。
第一个是RunnablePassthrough，中文叫透传。它本身不干活，输入是什么，输出就是什么。你可能觉得一个“不干活”的组件没用，但它在两种场景下很好使。一种是如果一条链的输入需要同时保留原始数据又要增加新字段，可以用RunnablePassthrough.assign(...)给数据字典新增一个 key。例如：
from langchain_core.runnables import RunnablePassthrough chain = ( RunnablePassthrough.assign(len=lambda x: len(x["article"])) | prompt | llm | StrOutputParser() )这里原始输入里面有article，经过assign之后，新字典里多了一个len字段，而原始字段不会动。这种“保留原上下文，同时追加计算结果”的思路，在写 RAG 类链路时特别常用。
第二个是RunnableLambda。它的作用是把你手写的 Python 函数包装成 Runnable。你可以把一个普通的同步函数、异步函数直接变成组件挂到链路上。我常用的写法是这样：
from langchain_core.runnables import RunnableLambda def filter_short_texts(texts): return [t for t in texts if len(t) > 20] chain = retriever | RunnableLambda(filter_short_texts) | prompt | llm | StrOutputParser()这里retriever可以是任何返回文本列表的组件，过滤函数被包进去之后，后续环节拿到的就是过滤后的结果。写业务校验、后处理、异常兜底，都是RunnableLambda的活。
第三个是RunnableParallel，它的作用是并行执行多条子链，最后把结果汇总成一个字典。例如同时做摘要和关键词提取，就可以让模型的两套调用同时跑，而不是串行等完一个再跑另一个。后面实操部分我会具体演示。
2.4 链的运行时配置：bind、with_config、with_fallbacks
LCEL 里很多坑来自“参数改不到”，比如想临时调低温度、想给模型加停止词、想在某个环节失败后用备用模型兜底。这些需求LCEL都有对应的“链级控制器”。
想给模型绑定固定的额外参数，用bind。比如你的链路是prompt | llm | parser，但你希望这次调用里模型使用temperature=0.2，同时对输出增加一些停止标记，就可以写：
chain = prompt | llm.bind(temperature=0.2, stop=["\n\n"]) | parserbind的作用是把参数“焊死”在模型这个组件上，每次调用都自动携带，不用你在 invoke 的时候手动干预。
想给整条链路挂标签、元数据、回调，用with_config。这个功能我最初完全忽略，直到排查生产问题的时候才发现它的价值：你可以给不同的链打上不同的 tags，比如["客服", "v1.2"]，这样在链路追踪日志里一眼就能分辨是哪条链在哪次请求中跑的。
想给链路加容错，用with_fallbacks和with_retry。前者可以指定备用组件，比如主模型超时后自动切到备用模型或者缓存；后者可以自动重试异常，尤其适合处理临时限流。以前这些逻辑都要自己用 try/except 包一层，现在声明式地挂到组件上就行。我第一次用with_fallbacks是在一个文本抽取 Demo 里，主模型不稳定，挂了之后自动切到备用模型，那一刻才觉得链路真正有了“工程感”。
3. 实操记录：三条链子把LCEL玩明白
3.1 最基础的结构化提取链：从一段反馈里抽字段
我们先做一个最常见的练手 Demo：从一段用户反馈里提取结构化信息。比如输入一句话，要求模型输出三个方面的一个 JSON，包括“方面”、“情绪”和“关键句”。用 LCEL 写起来非常简洁：
from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser parser = JsonOutputParser() prompt = ChatPromptTemplate.from_template( """从下面的用户反馈中提取字段： - 方面 - 情绪（只能取 positive / negative / neutral） - 关键句 用户反馈： {feedback} {format_instructions} """ ).partial(format_instructions=parser.get_format_instructions()) # your_chat_model 替换为你实际初始化的模型 chain = prompt | your_chat_model | parser result = chain.invoke({"feedback": "物流太慢了，包装倒是很结实，好评"}) print(result)这里重要的一个点，是JsonOutputParser会生成一段“格式说明”，告诉模型应该输出什么样的 JSON。通过.partial()把这段说明固定在 Prompt 里，模型就能稳定地输出 JSON 结构。最终返回结果是 Python 字典，可以直接拿给后续逻辑用，不用自己写json.loads。
我踩过的坑是，最初我用的是字符串输出再加json.loads解析，偶尔会因为模型返回的 JSON 里带多余的 Markdown 代码块而解析失败。后来换成JsonOutputParser，这类错误明显减少。当然它也不是万能，如果模型输出彻底不符合 JSON 格式，它该报错还是报错，但至少省掉了我自己写容错解析的一堆代码。
3.2 并行链：让两条子链同时跑，最后拼成一个结果
在实际项目里，经常需要针对同一段内容做多个任务。比如一篇文章进去，既让模型写摘要，又让模型提取关键词。如果串行做，总耗时是两次模型调用的时间之和；用RunnableParallel并行做，总耗时就是两次的较大值。
from langchain_core.runnables import RunnableParallel summary_prompt = ChatPromptTemplate.from_template("用一句话概括下面的文章：\n{article}") keyword_prompt = ChatPromptTemplate.from_template("提取这篇文章的5个关键词，用逗号分隔：\n{article}") summary_chain = summary_prompt | your_chat_model | StrOutputParser() keyword_chain = keyword_prompt | your_chat_model | StrOutputParser() parallel_chain = RunnableParallel( summary=summary_chain, keywords=keyword_chain, ) result = parallel_chain.invoke({"article": "这里是文章正文..."}) print(result["summary"]) print(result["keywords"])这里RunnableParallel会把原始输入字典同时传给两条子链，然后等它们都完成之后，把各自的输出按 key 拼起来。返回的result是一个字典，包含summary和keywords两个字段，取用特别方便。
需要注意的是，并行调用会放大限流风险。如果你一下子挂十几条子链同时打同一个模型接口，大概率会触发频率限制。我通常控制在三到五条以内，再多的任务就分批处理。别为了贪快，把自己账号的调用配额打爆。
3.3 在链路里插入自定义逻辑：RunnableLambda 的实际用法
真实项目很少是“输入 -> 模型 -> 输出”这么干净，中间一般有校验、过滤、查数据库、重试之类的逻辑。这些逻辑并不依赖模型，但又要跟模型调用一起组成一条完整流程。用RunnableLambda把它们包成 Runnable 就行了。
拿我做过的一个问答项目举例，大致流程是：用户问题进来，先判断是不是空字符串；如果非空，通过检索模块拿候选段落；候选段落经过长度过滤和排序，拼成 Prompt 上下文；最后调模型生成回答。用 LCEL 写出来的结构大概是：
from langchain_core.runnables import RunnableLambda def validate_query(query): if not query or len(query.strip()) == 0: raise ValueError("query is empty") return query.strip() def rank_documents(docs): # docs 假设已经是带分数的候选结果，这里做一次排序和过滤 docs = sorted(docs, key=lambda x: x["score"], reverse=True) return [d for d in docs if d["score"] >= 0.3][:5] def format_context(docs): return "\n\n".join(d["content"] for d in docs) preprocess = ( RunnableLambda(validate_query) | retriever | RunnableLambda(rank_documents) | RunnableLambda(format_context) ) chain = preprocess | prompt | your_chat_model | StrOutputParser()这种写法最大的好处是，整条链路里每一步都能被单独测试。你甚至可以临时拆出preprocess这条链单独跑一遍，看看到底是检索问题、过滤问题还是模型问题。而且RunnableLambda里面抛出的异常会被上层捕获，你可以再给它接with_retry或者自定义兜底逻辑，不用把所有异常处理都堆在函数里。
3.4 流式输出与异步批量：从串行到并发
LCEL 在流式输出上的体验，是我最欣赏的部分之一。只要模型支持流式，你对链调用.stream()，就能拿到逐段生成的结果：
for chunk in chain.stream({"topic": "程序员笑话"}): print(chunk, end="", flush=True)这个能力在聊天机器人、打字机效果、边生成边展示的场景里特别重要。你不用自己另外写流式接口的对接代码，链路本身通过 Runnable 协议把流式能力传下去了。
异步调用则适合 IO 密集型的批量任务。比如要处理 50 条文本，用batch方法可以自动并发；如果是在异步代码里，就用abatch。写法几乎不需要额外学习成本：
inputs = [{"feedback": f"反馈文本{i}"} for i in range(20)] results = await chain.abatch(inputs)说实话，过去自己手动实现批量并发，又要创建线程池又要收集结果，代码写一大坨。现在一行方法调用，链路自己负责并发调度。前提是你要理解.batch()不等同于无限并发，实际并发量取决于底层模型接口的限制，必要时还是要分批。
4. 排查实录：LCEL运行时容易踩的坑
4.1 流式输出为什么总是一下子全部返回
如果你用.stream()时发现结果等了很久之后一次性全出来，而不是一个一个冒出来，大概率是链路里某个环节不支持流式。最常见的嫌疑是输出解析器或某个RunnableLambda。
我在初学阶段就遇到过：链是prompt | llm | parser，模型本身支持流式，但只要加了一个自定义RunnableLambda做后处理，流式就“失效”了。原因很简单：RunnableLambda是一个一次性接收全部输入的函数，它要么看到完整结果后执行处理，要么等待上游全部结束后才返回。这就打断了流式传递。排查思路是把链路从中间拆开，分别测试prompt | llm和prompt | llm | parser的流式行为，定位是哪一层引起的问题。
另外StrOutputParser和较新版本的JsonOutputParser是支持流式的，但一些专门做格式校验的 Pydantic 解析器通常不是。如果你必须用强校验的输出格式，又想保留流式效果，一般做法是先用StrOutputParser做流式展示，最后在处理阶段再做严格校验。
4.2 bind 和 prompt 变量别搞混
我见过不少人把模型参数放在 Prompt 变量里传，比如这样：在 invoke 时传一个temperature进字典。但 Prompt 模板并不会把它传给模型，它只会被当成模板变量填充进字符串里，最后模型根本收不到。可能你以为模型改了参数，其实它只是看到了文案里的“temperature=0.2”。
正确的做法就是用bind。你可以理解成，bind是在调用真正模型之前，把一些固定参数事先“注入”进去。比如：
chain = prompt | your_chat_model.bind(stop=["\n用户："])这样每次调用模型时，都会附带stop序列。把它和 Prompt 变量区分开的关键原则是：变量是给模板填字的，bind是给模型调用传参数的。实在分不清，就记住一句口诀：用户输入走 invoke 字典，固定参数走 bind。
4.3 RunnableParallel 的返回顺序和并发控制
RunnableParallel返回的是一个字典，这个字典要等所有子链都完成之后才会产生，所以不要指望能早取到某个字段。另外，并行执行的子链之间是竞态关系，最终返回字典的 key 顺序不一定跟定义顺序完全一致，取结果时用 key 而不是下标。
还有并发控制的问题。RunnableParallel定义十条子链，确实意味着十个并发调用，但底层模型接口如果限流，就会导致大面积失败。我踩过的坑是一次挂十条链去提取不同字段，结果 6 个成功、4 个报限流错误。后面学老实了，控制在三条链以内，或加上重试。本质上 LCEL 给出的是能力，不是银弹，并发量上限还是要看你的模型服务允许多少。
4.4 调试链路：在每个环节加入口打印
LCEL 写到最后，最大的难点不是写出来，而是出问题时怎么定位。我的技巧是写一个小工具函数，用来在任意两个环节之间插一个“窥视孔”：
from langchain_core.runnables import RunnableLambda def debug_step(name): def _debug(x): print(f"[{name}] type={type(x).__name__} content={x}") return x return RunnableLambda(_debug) chain = prompt | debug_step("after_prompt") | your_chat_model | debug_step("after_model") | parser这样你可以非常直观地看到，数据在每一步到底是什么类型、什么内容。尤其当你需要确认传给模型的内容对不对时，看after_prompt那一步打印出来的消息内容就够了，不用再猜。
除了打印，我也推荐给链路挂上 debug 用途的 tags 或 metadata，并把回调数据接进你们团队的可观测平台。LCEL 的 Runnable 天然支持这类追踪，省去了自己手动埋点。早期项目里我完全没用这个能力，结果线上排查问题时两眼一抹黑，后来给每条链都打了 tag，定位问题速度提升明显。
4.5 版本变迁带来的一些“坑”
LangChain 的版本迭代速度比较快，LCEL 之后，很多旧概念都变了。我在学习过程中遇到过几个特别误导人的点。
一个是RunnableMap这个名字。它在某些老版本的文章里大量出现，其实就是RunnableParallel的前身。新代码里写RunnableMap会提示过时，最好直接统一用RunnableParallel。
另一个是模块导入路径。早期很多组件从langchain.chains导入，现在更推荐从langchain_core.runnables或对应领域模块导入。如果你查到的资料和当前环境对不上，优先看官方文档里的“入包路径迁移”说明，别硬着头皮改代码。
还有一点，官方虽然还保留了不少旧版Chain类，但维护重心明显在 LCEL 这套语法上。新项目建议直接用 LCEL，别为了省事继续走旧路子。你现在写的链路，未来升级、扩展、接入追踪都要容易得多。
5. 按我的经验，LCEL什么时候该用，什么时候别硬上
学习 LCEL 到一定阶段之后，我反而开始有意识克制自己。不是所有流程都要套 LCEL，也不是所有地方都适合用管道风格。
什么时候用 LCEL 非常舒服？一是流程属于明显的“顺序数据流”，输入从一头进入，经过若干处理，从另一头出来；二是你希望组件能够复用，同一个模型调用在多个链路里出现，拆成 Runnable 后可以被不同链共享；三是并发、流式、重试这些工程能力，你想在同一条链里一步到位。满足其中之一，LCEL 就是很好的选择。
什么时候别硬上？如果逻辑非常简单，只是一次 prompt 一次模型调用，那直接写普通 Python 函数反而更直接干净。硬套 LCEL 只会多出一层抽象，阅读成本反而上升。另外，如果你需要非常复杂的条件跳转、人工确认、多轮来回、状态保存，LCEL 虽然提供了RunnableBranch一类能力，但表达式风格写多了，可读性会显著下降。这种场景更适合用图式工作流框架来编排，把状态和跳转都显式表达出来，而不是硬塞进管道里。
还有一个经验是不要把 LCEL 链路排得太长。管道写法带来的流畅感很容易让人上头，一个链串上七八个RunnableLambda，出问题的时候定位成本会变大，每插一个调试节点也让链路变得臃肿。更好的做法是把长流程拆成几个短链，每个短链职责单一，再在上层做组装。短链之间也方便单独测试、单独复用。
我个人最终的心得是，把 LCEL 当作一种“接口协议”来理解，而不是单纯的写代码技巧。所有核心组件遵守 Runnable 接口，业务逻辑用RunnableLambda包裹，路线由管道清晰表达。只要守住这个习惯，后面无论是做流式展示、并行抽取，还是批量处理，改造起来都非常省事。这套笔记里的每个例子，我都建议你亲手敲一遍，然后试着把其中一条链改成异步调用，再试着加一层调试输出。跑通的那一刻，你才能真正感受到 LCEL 这套语法带来的确定性。
