最近一年多，凡是用LangChain做正经项目的人，基本都从最初那种prompt → model → parse的硬编码写法，慢慢迁到LCEL上来了。LCEL的全称是LangChain Expression Language，说人话就是把Prompt模板、大模型、输出解析器这些组件，用管道符|一个接一个地串成一条链。你写一行chain = prompt | model | parser，背后就是一个完整的可复用、可流式、可并行、可调试的AI处理流水线。这篇文章我把这套链式编程的核心思路、每个环节的选型细节、完整可跑的示例代码，以及我实际踩过的坑全部整理出来，适合刚开始接触LangChain、被“链式调用”这个概念绕晕的开发者，也适合想把手头脚本升级成工程化方案的同学。
1. LCEL到底在解决什么问题：从“胶水代码”到“声明式流水线”
1.1 传统写法为什么难受
先看一段最传统的LangChain调用代码，我相信很多人都写过：
prompt_text = "用{language}写一个{concept}的科普解释，300字以内" formatted_prompt = prompt_text.format(language="Python", concept="装饰器") response = llm.invoke(formatted_prompt) content = response.content print(content)这段代码的问题在只有一两个环节时还不明显，一旦流程复杂起来就非常难受。比如你要先提取用户意图、再检索知识库、再生成回答，你就得自己维护一堆中间变量：intent、retrieved_docs、final_prompt，每一步都要手写“把上一个函数的输出塞给下一个函数”的逻辑。这种代码在项目初期跑得通，但到后期基本不敢改，因为你动任何一环都可能悄悄影响下游输入的形状。我甚至见过一个同事把二十多步的处理流程写在一个函数里，每个步骤靠注释分隔，线上出了问题只能从头print到尾，排查一次要半小时。
LCEL解决的就是这个痛点。它把“上一个输出作为下一个输入”这个动作抽象成管道操作符|，你不再需要手动传参，只需要声明链条：左边执行完，结果自动流到右边。整个流程从一段命令式的“步骤清单”，变成了声明式的“组件流水线”，阅读和理解成本都低了一个量级。
1.2 Runnable协议：|背后的设计哲学
管道符不是简单的运算符重载，它背后是一整套统一的组件协议，在LangChain里叫作Runnable。LCEL里几乎每个组件都是Runnable：Prompt模板是，模型是，输出解析器是，甚至一条由多个组件拼成的链本身也是。
一个Runnable必须实现统一的调用接口，常用的有四个：
invoke(input)：单次输入，返回单个输出batch(inputs)：批量输入，返回输出列表stream(input)：流式输出，逐个token返回ainvoke / abatch / astream：对应的异步版本
当两个Runnable用|连接时，LangChain内部会构造一个RunnableSequence，它本质上也是一个Runnable，所以你可以继续往上拼。管道符的语义很简单：左侧执行结果作为右侧的调用参数。这就跟Unix命令行里的ps aux | grep java一样，前一个程序的输出流变成后一个程序的输入流，只不过这里传递的是内存对象，不是文本流。
需要特别提醒一点：两个组件能不能用|连接，取决于它们的“接口形状”是否匹配。我见过很多新手在这里栽跟头，以为随便两个组件都能串。实际上LCEL在运行时才会真正把上游输出传给下游，如果你把PromptTemplate直接接到StrOutputParser上，大概率会报类型不匹配的错，因为解析器期望的是模型输出对象，而不是PromptValue。所以在设计链的时候，心里要时刻清楚每个环节的输入输出类型。
1.3 一条链等于一个组件
LCEL最具价值的设计在于：组合出来的链和单个组件是同构的，它同样实现了Runnable协议。这意味着你可以把一条链塞进另一条链里，作为其中的一个环节继续使用。
举个例子。我先做一个“清洗链”，专门负责把用户原始输入中的噪声去掉；再做一个“主回答链”，负责生成最终答案；最后把两个链串在一起：
clean_chain = clean_prompt | model | cleaner_parser main_chain = main_prompt | model | main_parser pipeline = clean_chain | main_chain result = pipeline.invoke({"raw_input": user_text})这种组合能力让项目的模块化程度大大提升。你甚至可以把一条链设计成可插拔的插件，比如“摘要链”和“翻译链”互相替换，只要接口形状一致，业务代码一行都不用改。我自己的经验是，一旦你习惯了这种思维，再回到传统写法会非常痛苦，因为你得手动管理每个阶段的状态，而在LCEL里，状态就是链条本身。
2. 链条上的三个核心环节：Prompt、模型、输出解析器
2.1 Prompt模板：变量注入与角色划分
链条的第一环通常都是Prompt模板。它的作用不是“写死一段提示词”，而是把用户输入、上下文、指令模板分离，让代码更干净、让提示词可复用。
LangChain里最常用的是ChatPromptTemplate，它支持多消息结构，可以显式区分system、human、assistant消息。这在对话场景里非常重要，因为Chat模型的输入本质上是一组消息列表，而不是一段纯文本。你手动拼字符串很容易丢失角色信息，ChatPromptTemplate则天然维护了这一点：
from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个{role}，请用{language}回答用户的问题。"), ("human", "{question}") ])from_messages里的大括号变量会在invoke时被统一注入。这里有几个细节值得注意。
第一，变量必须一次性给全，不能漏。给少了，LangChain不会帮你自动补空，而是直接抛异常。我经常看到新手漏传变量，然后一脸懵地翻日志。
第二，可以用partial_variables预先固定部分变量，适合那些每一轮都不变的设定。比如你做了一个专属客服机器人，system消息里要一直强调品牌名和话术风格，就可以把这些内容用partial_variables提前绑定，invoke时只需要传用户问题。
第三，Prompt engineering的核心在这里体现。经验法则就是把“指令”和“内容”分开：指令写在system里，要求模型以什么角色、什么风格、按什么格式输出；内容放在human里，是每一轮真正变化的输入。这样既节省token（因为system部分可以让平台缓存），又让模型的任务边界更清晰。实测下来，把格式要求写清楚、把指令步骤化，对输出质量的提升远比你换一个更大的模型明显。
2.2 模型调用：选型与参数背后的取舍
链条的第二环是模型。在LCEL里，这一步通常是ChatOpenAI或其他Chat模型实例，它的输入是上游传下来的PromptValue，输出是一个AIMessage对象。
选模型这件事没有标准答案，但有几个参数我每次都会认真调。第一是temperature，它控制的是输出的随机性。凡是后续要接解析器的链，我基本都调到0到0.3，因为你需要的是稳定格式，而不是天马行空；只有做创意文案、头脑风暴这类任务才建议调到0.7以上。第二是max_tokens，这个参数决定模型最多生成多少token。很多人忽略它，结果模型生成超长文本，既浪费钱又拖慢速度，尤其在解析任务里，生成几百字的废话对下游毫无价值。
from langchain_openai import ChatOpenAI model = ChatOpenAI( model="gpt-4o-mini", temperature=0.1, max_tokens=1024 )这里必须提一下接入方式。langchain-openai包需要配置API Key，我强烈建议不要硬编码在代码里，而是通过环境变量或者.env文件加载。项目一旦多人协作或者上了CI，把密钥写死在脚本里，迟早会出事。
关于模型输入，有个容易混淆的点：Chat模型接收的不是字符串，而是PromptValue对象。所以你在LCEL链里看到prompt | model能连通，是因为上游Prompt刚好输出的是模型期望的类型。如果你自己写了个函数想直接接到模型上，就得用RunnableLambda包一层，确保输出类型匹配。
2.3 输出解析器：把token流变成字段
链条的最后一环通常是输出解析器，它负责把模型返回的AIMessage内容转成你真正需要的数据结构。这个环节最容易被忽略，但恰恰是它决定了你后端的代码能不能优雅地消费AI产出。
最基础的是StrOutputParser，它只做一件事：把AIMessage里的content取出来，变成纯字符串。适合你只想要一段自然语言答案的场景：
from langchain_core.output_parsers import StrOutputParser parser = StrOutputParser()如果你需要结构化输出，比如让模型返回JSON，那就要用到PydanticOutputParser。配合Pydantic模型，你可以约定输出的字段、类型和含义，模型会按你的schema返回JSON对象，解析器再把JSON反序列化成Pydantic实例：
from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class ArticleMeta(BaseModel): title: str = Field(description="文章标题") summary: str = Field(description="一句话摘要") tags: list[str] = Field(description="推荐标签列表") parser = PydanticOutputParser(pydantic_object=ArticleMeta)这里有个超级常见的坑：写好了解析器，却没把它的格式说明注入到Prompt里，结果模型根本不知道要输出JSON，吐出来一段散文，解析器直接报错。正确做法是把parser.get_format_instructions()的结果拼进Prompt模板里，让模型明确知道输出目标和约束。
还有一个版本兼容性问题需要注意。LangChain内部用的是兼容Pydantic v1的langchain_core.pydantic_v1模块，如果你的项目用的是Pydantic v2，可能会遇到类型不匹配的报错。这不是你的代码写错了，而是两套schema验证逻辑不一致导致的，处理办法就是统一使用LangChain推荐的导入路径。
3. 实操：用管道符搭出可上线的链
3.1 最小可运行链：一句话问答
现在我们从零开始搭一条能跑的最小链。先安装依赖：
pip install langchain langchain-openai python-dotenv然后在项目根目录创建.env文件，把密钥放进去：
OPENAI_API_KEY=sk-你的密钥接下来写主文件：
import os from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser load_dotenv() prompt = ChatPromptTemplate.from_template( "用{language}写一个关于{concept}的通俗解释，控制在200字以内。" ) model = ChatOpenAI( model="gpt-4o-mini", temperature=0.3 ) parser = StrOutputParser() chain = prompt | model | parser result = chain.invoke({ "language": "Python", "concept": "闭包" }) print(result)跑起来之后，你会看到一个字符串形式的回答。这个过程中真正发生的事情是这样的：invoke把字典传给prompt，prompt校验变量并构建出PromptValue；这个PromptValue传给model，模型返回AIMessage；AIMessage传给parser，parser取出content字符串。整条链总共四行代码，却涵盖了从模板渲染到模型推理再到结果提取的全过程。
我建议你拿到最小链之后，先别急着往上加功能，而是把这四行代码的每个环节都打印出来看一遍。比如用prompt.invoke(...)单独看PromptValue长什么样，用model.invoke(...)单独看AIMessage的结构。只有把每个环节的输入输出形状摸透了，后面排错才会快。
3.2 结构化输出链：Pydantic解析器实战
接下来做一个实际项目里更常见的场景：输入一段新闻文本，要求模型抽取标题、来源、时间和涉及人物，输出结构化字段。
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import PydanticOutputParser from langchain_core.pydantic_v1 import BaseModel, Field class NewsItem(BaseModel): title: str = Field(description="新闻标题") source: str = Field(description="新闻来源") published_at: str = Field(description="发布时间，格式YYYY-MM-DD") people: list[str] = Field(description="新闻中涉及的主要人物") parser = PydanticOutputParser(pydantic_object=NewsItem) prompt = ChatPromptTemplate.from_messages([ ("system", "从用户提供的新闻文本中抽取信息。\n{format_instructions}"), ("human", "新闻文本：\n{news_text}") ]) model = ChatOpenAI(model="gpt-4o-mini", temperature=0) chain = prompt | model | parser注意看这里的关键一步：我在system消息里放了一个{format_instructions}变量，invoke时要把解析器的格式说明传进去：
result = chain.invoke({ "news_text": "本报讯（记者张三）昨日，AI科技公司发布新一代大模型，首席科学家李四表示该模型在推理任务上表现优异……", "format_instructions": parser.get_format_instructions() }) print(result.title) print(result.people)运行成功后，result是一个NewsItem实例，你可以直接用result.title这种属性访问方式读取字段，不需要自己写JSON解析逻辑。这在批量处理的场景下特别省事：一次invoke返回一个结构化对象，直接塞给下游数据库或表单渲染逻辑，中间不产生任何脏数据。
我再补充一个实操中的体感：这种链的稳定性高度依赖prompt里的格式说明是否清晰。如果模型偶尔输出了一段“前言”再输出JSON，PydanticOutputParser往往会解析失败。遇到这种情况，可以配合下一节讲的OutputFixingParser做自动修复，或者干脆在Prompt里加一句“只输出JSON对象，不要输出任何其他文字”。
3.3 进阶玩法：流式、并行、分支与回退
最小链跑通之后，你会慢慢发现LCEL真正强大的地方在于它预置了一整套高级组合能力，我这里挑几个最实用的展开。
先说流式输出。把invoke换成stream，模型就会一个token一个token地吐结果，这是做打字机效果的基础：
for chunk in chain.stream({ "language": "Python", "concept": "生成器" }): print(chunk, end="", flush=True)注意，只有在链条末端接的是StrOutputParser这类能逐步解析的解析器时，流式效果才是真正逐token的。如果末端是PydanticOutputParser，它必须等到完整JSON才能解析，流式效果会大幅退化甚至不生效。这个取舍在选解析器的时候就要想清楚。
再说并行。不同链之间可以进行合并，然后用一个统一入口触发：
from langchain_core.runnables import RunnableParallel summary_chain = summary_prompt | model | StrOutputParser() keywords_chain = keywords_prompt | model | StrOutputParser() parallel = RunnableParallel( summary=summary_chain, keywords=keywords_chain ) result = parallel.invoke({"article": article_text}) print(result["summary"]) print(result["keywords"])这个场景在真实项目里非常常见，比如你既想要一篇摘要，又想提取关键词，串行跑两遍太慢，并行就能把时间省下将近一半。
然后是分支路由。LCEL里的RunnableBranch可以根据条件把输入导向不同的链，相当于if-else的声明式写法：
from langchain_core.runnables import RunnableBranch tech_chain = tech_prompt | tech_model | parser general_chain = general_prompt | general_model | parser branch = RunnableBranch( (lambda x: x.get("is_technical") == True, tech_chain), general_chain ) result = branch.invoke({"question": question, "is_technical": True})最后是降级回退。这个功能我要重点推荐。线上跑LLM应用最怕的就是供应商API不稳定，某一段时间疯狂报错。LCEL的with_fallbacks能让你在一把钥匙断了的时候自动换备用钥匙：
fallback_model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) safe_chain = chain.with_fallbacks([fallback_model | parser])这样当主模型因为限流或审查拦截图挂掉时，链会自动切到备用模型继续跑，对用户来说感知不到中断。
4. 常见问题与排查技巧实录
4.1 prompt被判定违规（flagged）怎么办
运行链时可能会遇到类似这样的错误：invalid prompt: your prompt was flagged as potentially violating our usage policy。我第一次遇到时一脸懵，以为是模型出了问题，后来才知道这是模型厂商的内容安全审核机制在起作用。它的本质是：你的输入prompt触发了服务端的敏感内容检测，系统直接拒绝处理请求，返回一个错误码，而不是让模型正常推理。
这里要摆正心态：这不是“绕不过去的障碍”，而是安全红线。作为工程师，我们要做的不是想办法绕过审核，而是设计合规的请求，把用户输入引导到安全、正向、合规的方向上。
实际排查步骤我建议这样走：
- 把prompt拆开，一段一段测试，定位到底哪句话触发了拦截。用二分法切，通常很快就能找到。
- 重点检查system指令，看是否存在越权引导、对抗性指令或让模型扮演不当角色的描述。
- 把命令式写法改成中性任务描述。比如强调“你是一个负责任的助手，只提供合法合规信息”。
- 在进入链之前加一层输入校验，把明显违规的用户输入直接拦截在门外，别让它进入模型。这既保护模型，也保护你的应用。
还有一个关于错误处理的经验：不要因为少数恶意输入就让整条链崩溃。可以在with_fallbacks里指定一个安全回答链，当主链被拦时返回一段预设的说明，而不是直接抛异常吓到用户。
4.2 解析器输出格式不稳定
LCEL链里最脆弱的一环就是输出解析，尤其是PydanticOutputParser。模型是基于概率生成的，哪怕你prompt里写得再清楚“只输出JSON”，它也可能偶尔多输出一段解释文字、把字段名拼错、或者漏掉某个必填字段。这种问题在长文本、复杂schema上尤其明显。
我的处理经验分三层。第一层是在Prompt层面严防死守，把格式指令放system里，并且给出一个few-shot示例。第二层是给Pydantic字段设置合理的默认值，title: str = Field(default="未命名", ...)，这样即使模型漏了字段，解析也不会直接炸。第三层是用LangChain自带的OutputFixingParser做自动修复：
from langchain_core.output_parsers import OutputFixingParser fixing_parser = OutputFixingParser.from_llm( parser=parser, llm=model ) chain = prompt | model | fixing_parser原理是：当原始解析器解析失败时，OutputFixingParser会把报错信息和原始输出一起交给模型，让模型帮忙修正格式再解析。实测下来能挽回大部分格式错误，但它会增加一次额外的模型调用，所以只建议在对实时性要求不高的场景里用。
还有一个小坑值得记一下：如果parser本身解析失败，LangChain的异常信息可能隐藏得很深，排错的时候先打印parser.get_format_instructions()，再拿模型原始输出手动比对，往往能一眼看出问题。
4.3 token超限与prompt精简
链路一长，模型输入很容易逼近上下文窗口上限。最典型的报错是类似“This model's maximum context length is X tokens. However, you requested Y tokens”的信息。这种问题在长文档处理场景里几乎天天见。
要解决，第一步是量化。别靠猜，直接用tiktoken这类工具统计token数：
import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") tokens = enc.encode(prompt_text) print(len(tokens))统计出来之后，看哪部分占大头。按我的经验，系统指令和few-shot示例往往是隐形吞噬token的大户。优化策略有几个方向：把固定指令精简到最核心的动作；few-shot从四个示例砍到两个；用户输入过长就先用摘要链压缩一遍再进主线。
还要学会给模型留“呼吸空间”。不要把你每次请求的token上限撑到极限，因为模型还要生成回复，而生成token也要占用同一个上下文窗口。我一般会预留总窗口的20%以上给输出，这样能避免生成到一半被强行截断的尴尬。
另外，prompt token这个概念要理解清楚：它不是一次性的，是每次请求都要重新计算的。你在prompt里多写100个token，每个请求就多付100个token的钱。所以prompt engineering不只是为了效果，也是为了成本。
4.4 链式调试：如何看清每一环的输入输出
LCEL链写多了之后，你会发现一个问题：链条本质上是一个黑盒，chain.invoke(...)给你最终结果，但中间每一步发生了什么你完全看不到。定位问题全靠猜，效率很低。
我的调试套路是用RunnableLambda在链上插“探针”，打印每一环的中间状态：
from langchain_core.runnables import RunnableLambda from langchain_core.output_parsers import StrOutputParser def debug_step(data, name=""): print(f"=== {name} ===") print(repr(data)[:800]) return data debug_model = RunnableLambda(lambda x: debug_step(x, "model_output")) chain = prompt | model | debug_model | StrOutputParser()这样在模型输出之后、解析器处理之前，你就能看到AIMessage的完整内容。解析器如果报错，直接拿这段原始输出和格式要求对比，问题基本三分钟定位。
另一个技巧是给链和组件命名。chain.invoke时可以通过config传参，也可以用.with_config({"run_name": "my_chain"})给链取名。如果你接入了LangSmith这类追踪平台，命名清楚能极大提升检索效率。即使不接入，你也要养成在每个RunnableLambda里输出清晰的标记字段的习惯，打印出来一目了然。
我还建议在调试阶段把链拆开跑一遍：prompt.invoke(...)看渲染结果，model.invoke(...)看模型原始响应，parser.invoke(...)看解析行为。一条条单独验证，组合时不变量匹配再逐步拼回去。很多看起来诡异的问题，最后发现就是某个环节的输入形状对不上，单独跑一遍立刻现形。
我个人在实际操作中的体会是，LCEL最妙的地方在于它逼着你去思考每个环节的“接口形状”。刚开始我不习惯，总觉得多此一举，但用久了才发现，正是这种契约式的设计让AI应用变得可工程化。管道符那个|看起来简单，背后是Runnable协议带来的组合与复用能力，你可以把任意组件快速拼接、并行、分流、降级，这在传统代码里至少要多写几十行样板。最后再分享一个小技巧：新写的链，先在本地用测试数据跑一遍stream模式，观察每一环的输出形状，这比直接invoke审视一遍更容易发现潜在问题。等你习惯了这种“搭积木”的思考方式，做复杂AI应用的速度会快上一个台阶。
