LangChain LCEL 从入门到实践：用表达式语言构建AI流程管道-北京尧图网络

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(lenlambda 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_fallbacksLCEL 里很多坑来自“参数改不到”比如想临时调低温度、想给模型加停止词、想在某个环节失败后用备用模型兜底。这些需求LCEL都有对应的“链级控制器”。想给模型绑定固定的额外参数用bind。比如你的链路是prompt | llm | parser但你希望这次调用里模型使用temperature0.2同时对输出增加一些停止标记就可以写chain prompt | llm.bind(temperature0.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_instructionsparser.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( summarysummary_chain, keywordskeyword_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, keylambda x: x[score], reverseTrue) 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, flushTrue)这个能力在聊天机器人、打字机效果、边生成边展示的场景里特别重要。你不用自己另外写流式接口的对接代码链路本身通过 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 模板并不会把它传给模型它只会被当成模板变量填充进字符串里最后模型根本收不到。可能你以为模型改了参数其实它只是看到了文案里的“temperature0.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 这套语法带来的确定性。

LangChain LCEL 从入门到实践：用表达式语言构建AI流程管道

相关推荐

OpenClaw 部署全指南：从编译环境搭建到游戏资产提取

Fiddler弱网测试实战：延迟、带宽与网络制式配置详解

BMS电池管理系统深度解析：从SOC估算到安全保护的硬核技术

更多资讯

MCP、Rule、Skill、Agent区别与关系：一张图理清AI智能体核心概念

ArcGIS XY坐标批量转点线面要素：从坐标系避坑到arcpy脚本实战

Claude Code模板工程：用CLAUDE.md与Hooks约束AI，告别随机输出

内省阈值：为什么LLM无法真正自我进化？

NodeMCU网页发中文串口显示：全链路编码与乱码排查

多参数贝叶斯估计的核心：联合概率分布与MCMC后验采样实战

今日更新

NetSpot 实战：用信道分析与频宽优化解决 Wi-Fi 卡顿

C++ split() 字符串分割的五种实现与性能实测

R语言坐标轴范围设置：从xlim/ylim到coord_cartesian的完整指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？