1. 先从我们最常踩的坑说起为什么会有LCEL我最早接触LangChain的时候还是Chain满天飞的版本。写一个调用链得先建PromptTemplate再建LLMChain再建OutputParser每一个环节都是独立类接口风格还不统一。有的组件支持predict有的支持apply有的支持generate记接口比记业务需求还累。更头疼的是你很难把两个Chain组合在一起想加一个前置步骤就得重写类想流式输出就得换一套API。那段时间社区里讨论最多的话题就是“Chain到底该怎么组织”。后来LangChain团队推出了LCEL全称LangChain Expression Language中文一般叫LangChain表达式语言。它的核心思路很直接把每个处理环节统一抽象成Runnable然后用管道符|把它们串起来。你不需要再纠结某个类是Chain还是Executor只要它实现了Runnable协议就能参与组合。这套设计的价值用一句大实话概括就是把搭流程变成拼管道把各写各的变成统一协议。任何组件都是Runnable任何Runnable都可以互相组合。Prompt是Runnable模型是Runnable解析器是Runnable连你随手写的一个Python函数通过RunnableLambda包装一下也变成了Runnable。如果你已经能跑通基础的LangChain调用但对LCEL和Runnable一直停留在“好像很厉害但不知道厉害在哪”的状态这篇内容就是给你准备的。我会从接口设计、执行机制、实操配置、问题排查几个维度把这条学习路线完整捋一遍。2. Runnable核心接口与执行机制拆解2.1 Runnable协议到底定义了哪些能力Runnable本质上是一个抽象基类定义了一批标准方法。只要实现了这些方法一个对象就能接入LCEL的流水线。以下几个是必须掌握的懒加载执行入口方法同步/异步作用典型场景invoke同步执行单次调用返回完整结果普通请求、调试ainvoke异步异步执行单次调用FastAPI接口、异步爬虫batch同步批量执行输入列表离线批量处理abatch异步异步批量执行高并发批量任务stream同步流式返回结果片段打字机效果、长文本输出astream异步异步流式返回WebSocket实时推送建议读者用一个生活类比来理解。Runnable就像一条流水线上的工作站。invoke是来一个工件加工一次batch是一次铺一排工件逐个加工stream是加工完一部分就立刻从传送带上传出去不等整个工件完工。为什么LangChain用的是invoke而不是run我理解是因为run这个词在历史版本里已经被用滥了不同组件里的run含义五花八门有的返回字符串有的返回字典有的返回消息对象。统一叫invoke再加上RunnableConfig参数语义就清晰多了。2.2 管道符做了什么事组合背后的执行逻辑LCEL最引人注目的语法就是|。它背后是怎么工作的其实每两个Runnable用管道符相连LangChain内部就会构建一个RunnableSequence对象。比如chain prompt | llm | parser这里发生的事是先执行prompt把它的输出作为llm的输入再把llm的输出给parser最后把parser的输出作为整个链的结果返回。管道的左侧输出格式必须能匹配右侧的输入格式这一点和Unix命令行的管道设计非常像。但有一个关键区别构建这条链的时候什么都不会执行。LCEL是延迟执行的构建链只是一个组装描述的过程只有你真正调用invoke()等执行方法时数据才开始流动。我遇到过不少朋友在构建链时试图打印中间结果发现什么都没输出以为代码写错了其实就是没有触达执行阶段。另外RunnableSequence不仅会顺序执行还会做一些额外的工作。它会把RunnableConfig逐层透传给每一个子组件会尝试合并流式事件流会在批处理时自动把输入拆分再汇总。这就是为什么LCEL组合出来的流程天然拥有比手写代码更强的可观测性和调度能力。2.3 配置透传RunnableConfig是隐藏的调度控制台很多人在学LCEL时会忽略RunnableConfig这个参数。它可以说是整个执行机制的调度控制台。from langchain_core.runnables import RunnableConfig config RunnableConfig( max_concurrency5, recursion_limit20, run_namequery_sql_chain, tags[production], metadata{user_id: demo_user} ) chain.invoke({question: 按部门统计本周工单数量}, configconfig)这个config会被透传到链里的所有Runnable组件。你可以在这里切换回调处理器、设置缓存、限制并发度、给运行打标签。LangSmith的追踪信息也会读取这个config里的run_name和tags方便你在后台定位是哪条链出了问题。我见过很多项目里排查线上问题时只能翻日志手算调用关系就是因为没有在config里埋标记。其实只要在入口处传一个带run_name的config整个链的每一步都会带上这个标识性能分析和错误定位的效率会明显提高。3. 实操用LCEL搭一条可观测、可扩展的LLM调用链3.1 从零搭一个结构化输出链光说概念不过瘾我们直接写一个能跑的例子。假设我们要做一个“用户需求转技术方案”的小工具用户输入一句话需求系统返回一个结构化的技术方案JSON。先定义输出结构from typing import List from pydantic import BaseModel, Field class TechnologyOption(BaseModel): name: str Field(description技术方案名称) difficulty: int Field(description实现难度1到5越大越难) timeline_days: int Field(description预计工期天数) class SolutionResult(BaseModel): summary: str Field(description方案摘要) options: List[TechnologyOption] Field(description可选技术方案列表)然后构造Prompt、模型和解析器。这里重点提一下最新版本里推荐用with_structured_output它可以直接要求模型按Pydantic模型输出避免自己写解析逻辑from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深技术架构师。请把用户的需求拆解为可执行的技术方案输出必须符合给定格式。), (human, 用户需求{requirement}) ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) structured_llm llm.with_structured_output(SolutionResult)接下来用LCEL串起来chain prompt | structured_llm调用result chain.invoke({requirement: 我需要一个内部工具让运营同学可以自助查询订单数据不需要写SQL。}) print(result)运行之后你会得到一个SolutionResult对象不需要再自己写正则抽取JSON。这里的关键点在于with_structured_output返回的也是一个Runnable而且它内部做了很多脏活自动生成JSON Schema、约束模型输出格式、解析响应并完整校验。更妙的是它对外暴露的接口和其他Runnable完全一致你依然可以用管道符把它接到链里。3.2 给链加上分支、并行和兜底逻辑实际生产环境里不可能只做一条链从头走到尾。输入不同、模型报错、结果格式不对都是常态。LCEL提供了几个非常实用的组合Runnable来应对这些情况。场景一根据输入类型走不同分支。比如输入中包含“紧急”字样时走快速通道否则走标准通道。可以用RunnableBranchfrom langchain_core.runnables import RunnableBranch, RunnableLambda def is_urgent(input_dict): return 紧急 in input_dict.get(requirement, ) quick_prompt ChatPromptTemplate.from_template(用户需求紧急请快速给出最小可行方案。需求{requirement}) standard_prompt ChatPromptTemplate.from_template(请给出完整技术方案。需求{requirement}) quick_chain quick_prompt | llm standard_chain standard_prompt | llm branch_chain RunnableBranch( (is_urgent, quick_chain), standard_chain )场景二多个模型并行调用谁先返回谁先用或者全部返回再合并。这里用到RunnableParallelfrom langchain_core.runnables import RunnableParallel llm_fast ChatOpenAI(modelgpt-4o-mini) llm_deep ChatOpenAI(modelgpt-4o) parallel_chain RunnableParallel( fast_answerprompt | llm_fast, deep_answerprompt | llm_deep ) out parallel_chain.invoke({requirement: ... }) print(out[fast_answer]) print(out[deep_answer])场景三模型触发限流或者网络抖动时自动重试或回退。LCEL提供了with_retry和with_fallbackssafe_llm llm.with_retry(retry_if_exception_type(TimeoutError,), stop_after_attempt3) backup_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) final_llm llm.with_fallbacks([backup_llm])这两个方法返回的依然是Runnable所以可以放心地继续往链后面接。3.3 完整链的落地形态把上面这些组合起来一条稍微完整的链长这样from langchain_core.runnables import RunnableParallel, RunnableBranch, RunnablePassthrough from langchain_core.output_parsers import StrOutputParser def route_by_tone(inputs): if 紧急 in inputs.get(requirement, ): return quick return standard prompt_std ChatPromptTemplate.from_template(用户需求{requirement}\n请给出技术方案概述) prompt_quick ChatPromptTemplate.from_template(用户需求{requirement}\n需求紧急请给出最简可行方案概述) llm_main ChatOpenAI(modelgpt-4o-mini, temperature0.3) llm_backup ChatOpenAI(modelgpt-3.5-turbo, temperature0.1) def no_op(x): return x chain ( RunnablePassthrough.assign(routeroute_by_tone) | RunnableBranch( (lambda x: x[route] quick, prompt_quick), prompt_std ) | llm_main.with_fallbacks([llm_backup]) | StrOutputParser() ) response chain.invoke({requirement: 我需要一个自动汇总每日销售数据的脚本紧急})补充说明一下RunnablePassthrough.assign的作用是让输入字典原样通过并在其上增加新的字段。这里它先根据requirement计算出route后续分支再依据route选择不同的Prompt。这个模式在真实项目里非常常用相当于在管道的中间位置插入了一个轻量计算节点。4. 流式输出、异步执行和批处理的正确打开方式4.1 打字机效果的实现路径LCEL真正让我觉得“值回票价”的功能是流式输出。同一套链不需要额外改造直接切到stream方法就能拿到生成器for chunk in chain.stream({requirement: 给我一个自动清洗Excel数据的方案}): print(chunk, end, flushTrue)这里有几个坑值得提醒。第一只有链路中的模型本身支持流式输出流式效果才能生效。如果你在链里写了一个自定义的RunnableLambda它对输入输出做的是整体处理那流式往往会把整块结果一次性返回分组变得没有意义。第二如果解析器是结构化解析器比如StrOutputParser它可以随流式逐步解析但如果用了with_structured_output加Pydantic模型通常要等完整结果才能解析成对象流式体验会打折。如果你做的是Web应用配合FastAPI的StreamingResponse可以做出一打字机效果的交互界面from fastapi.responses import StreamingResponse async def generate_solution(requirement: str): async for chunk in chain.astream({requirement: requirement}): yield chunk app.post(/solution) async def solution_endpoint(payload: dict): return StreamingResponse(generate_solution(payload[requirement]), media_typetext/plain)4.2 异步与并发batch接口的效率收益batch不是简单地for循环调用invoke。它会根据配置里的max_concurrency控制并发度所有输入一起执行结果按输入顺序返回。对于需要同时处理十几条文本标注任务的场景这个接口能显著缩短耗时。inputs [ {requirement: 自动生成周报模板}, {requirement: 搭建一个用于客户反馈分类的模型}, {requirement: 把PDF合同转为结构化字段}, ] results chain.batch(inputs, config{max_concurrency: 3})有一点必须注意batch在底层可能会开启多个线程或异步任务。如果你的链里有非线程安全的共享状态比如全局计数器、数据库连接很容易出现数据错乱。我自己的习惯是给每条输入在config的metadata里带上唯一请求ID所有日志和埋点都带这个ID排查问题时能少掉很多头发。异步接口ainvoke和abatch适合在协程环境中使用。FastAPI的接口天然支持async直接在路由函数里用await chain.ainvoke(...)没问题。但在Jupyter Notebook里跑asyncio.run()往往会报“event loop is already running”之类的错误需要用nest_asyncio或者直接换同步接口。4.3 RunnableLambda把普通函数变成流水线工位不是所有处理环节都来自LangChain官方组件。业务中经常需要插入自定义的辅助逻辑比如清洗文本、格式化日期、计算字段。RunnableLambda就是干这个的。from langchain_core.runnables import RunnableLambda def normalize_requirement(requirement: str) - str: return requirement.strip().replace(\n, ) chain ( RunnableLambda(normalize_requirement) | prompt | llm | StrOutputParser() )这里有一个容易踩的坑RunnableLambda包装的函数如果接收的是一个字典那它必须接收一个且仅接收一个参数返回值会直接传给下一个Runnable。如果想要函数接收多个命名参数可以在字典展开后传给函数或者用RunnableLambda的input_schema做更严格的参数映射。我一般建议在RunnableLambda里只做轻量操作重量级IO操作最好拆到独立的异步任务里。因为RunnableLambda默认是同步阻塞的如果函数内部执行了一个5秒的HTTP请求整个流水线都会卡住即使你调用的是astream也不会自动变成异步。5. 常见问题与排查技巧实录5.1 链能构建但invoke报错最典型的报错是AttributeError: RunnableSequence object has no attribute invoke。这通常是因为装的是非常古老的版本。LCEL在0.0.2xx时代才逐步稳定部分中间版本接口调整过。建议检查版本pip show langchain-core langchain把langchain-core升级到0.2以上或更新版本后这个问题基本消失。如果公司内部环境锁了版本那你需要确认当前版本的Runnable协议覆盖了哪些方法有可能只能用__call__或者run。5.2 流式输出得到的是整段文本出现这种情况先自查链里有没有RunnableLambda。很多自定义函数整体吞掉了流式事件。另一种可能是用了with_structured_output前面说过它为了解析成Pydantic对象需要完整结果。如果需要既流式又结构化可以把推理和结构化拆成两步先用普通模型流式输出完整文本再用一个单独Runnable做结构化解析。5.3 并发数配置不生效max_concurrency需要在RunnableConfig的max_concurrency字段里设置或者放在batch的config参数里。有些组件对并发策略有自己的实现比如某些向量存储或数据库检索器并发控制是在组件内部实现的外部config只对LCEL调度器生效。遇到这种问题最直接的办法是到组件的源码里确认invoke实现有没有使用config[max_concurrency]。还有一个高频错误在异步上下文里batch的行为可能不同步。batch内部默认使用线程池如果你传入的非线程安全对象出了问题可以先试abatch把并发交给事件循环。5.4 RunnableConfig的优先级问题当你定义链时给某个组件单独传了参数比如llm ChatOpenAI(temperature0)又在config里传temperature到底听谁的LCEL的规则是显式构造组件时传入的参数优先级最高config里的配置作为补充透传不会覆盖组件构造参数。所以如果你想动态调整温度不能只改config而是要把温度变成prompt模板变量或者用RunnableLambda构造一个可动态修改参数的包装器。我在实际项目中做过多模型切换和动态参数调整最后的统一方案是把模型实例放进一个缓存字典然后在RunnableLambda里根据业务字段选择缓存里的模型调用invoke。这样既保留了LCEL的组合能力又绕开了参数透传的语义限制。5.5 LangSmith追踪里的隐性问题排查如果使用了LangSmith发现某一步耗时异常不要只盯着大模型。先看prompt模板的计算耗时再看解析器的复杂度。有一次我排查一个链里的慢问题发现耗时全在一个看似简单的RunnableLambda里原因是它内部拼了一个巨大的列表推导式几万条数据循环。这种问题从追踪页面的step耗时分布里一眼就能看出来但如果你没打run_name和tags跨服务追踪会非常困难。现象可能原因排查建议invoke报“no attribute”langchain-core版本过旧升级langchain-core确认Runnable协议包含invoke流式无效果链路中有RunnableLambda整体返回去除Lambda、改用生成器函数或接受一次性结果batch结果顺序错乱自定义组件内部状态非线程安全给每条输入加唯一ID检查组件线程安全性异步报event loop错误Notebook环境事件循环冲突使用nest_asyncio或改用同步接口动态参数不生效config配置不覆盖构造参数用RunnableLambda包装模型实例动态选择结构化解析偶发失败模型输出格式不稳定增加降级解析器或with_fallbacks兜底5.6 多版本兼容的处理思路很多团队项目还运行在LangChain 0.0.x的老代码上新代码又引入了LCEL。这种过渡期最忌讳直接重写所有链。我用过的平稳迁移方式是新写功能尽量用LCEL老链包装成RunnableLambda接入新管道逐步替换。因为LCEL对Runnable的兼容性足够好一条老链可以用RunnableLambda(old_chain)快速封装然后和新的LCEL链并行运行对比输出结果后再下线旧逻辑。这个方法听起来简单实际执行时最大的坑是回调函数和老版本的事件系统不兼容。老链里注册的回调逻辑在LCEL管道里可能不触发需要在包装时把回调逻辑替换成以RunnableConfig传递的新格式。这类替换往往比改业务逻辑更费时间。6. 我的一点实操体会学了LCEL之后再看以前写的LangChain代码最感慨的其实是它的可组合性和接口一致性给我省下了大量“粘合代码”。以前每加一个功能都要新写一个类或者在一个函数里塞各种条件判断现在直接用管道符拼接每一条链都是一个Runnable对象可以被其他链继续镶嵌也可以被其他服务远程调用这种“处处皆组件”的体验非常舒服。我自己的习惯是拿到一个任务先画出数据流转的草稿标注每个环节的输入输出格式再决定哪些环节用官方组件哪些环节用RunnableLambda哪些环节需要并行或分支。这个前置设计步骤花不了多少时间却能避免很多返工。如果你刚开始学LCEL我建议别急着看API文档先把自己手头一个跑通的普通LangChain调用改写成LCEL风格。改完之后再试着加一个stream输出、一个batch调用、一个with_fallbacks兜底。这四步走完你对Runnable执行机制的理解会上升一个台阶之后再回头去研究那些高级组件会顺很多。