LangChain 的 Expression Language下称 LCEL是这两年在做 LLM 应用编排时绕不开的一套东西。我最早接触它的时候文档里全是|管道符串起来的链式写法看着像 Shell又像函数式编程第一反应是这玩意儿到底比直接写 Python 函数强在哪。真正在几个项目里用下来之后我的结论是LCEL 的价值不在于语法糖而在于它把可组合、可流式、可并行、可观测这四件事变成了默认行为而不是你每写一个功能都要重新搭一遍的脚手架。这篇 CookBook 我打算从实际使用的角度出发把 LCEL 里最常用、最容易踩坑、也最值得反复琢磨的几组写法拆开讲清楚包括管道组合的本质、Runnable 家族的接口契约、流式输出的接入方式、并行与分支的编排技巧以及调试和观测的实操手段。适合已经能跑通一个最小 Demo、但一遇到多个模型串联条件分支流式返回给前端就卡壳的读者也适合想搞清楚 LCEL 底层到底在做什么的人。1. 管道符背后到底发生了什么很多人第一次写 LCEL 是这样的from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template(用一句话解释{topic}) model ChatOpenAI(modelgpt-4o-mini) chain prompt | model chain.invoke({topic: 向量数据库})跑通了但心里没底这个|到底做了什么为什么prompt和model能直接拼起来要回答这个问题得先理解 LCEL 的核心抽象——Runnable 协议。1.1 Runnable 协议一切皆可运行对象LCEL 里所有能被|连接的东西本质上都实现了Runnable接口。这个接口定义了一组标准方法最核心的是四个方法作用返回形态invoke(input)单次调用单个结果batch(inputs)批量调用结果列表stream(input)流式调用结果块迭代器ainvoke/abatch/astream上述三者的异步版本对应异步形态关键在于只要一个对象实现了这套接口它就能和任何其他 Runnable 组合。Prompt 是 RunnableModel 是 RunnableOutputParser 是 Runnable甚至一个普通的 Python 函数用RunnableLambda包一层也变成 Runnable。这就是为什么|能工作——它不是在拼接字符串而是在做对象组合。|运算符在 Python 里对应__or__魔术方法。Runnable 重载了这个方法使得a | b返回一个RunnableSequence对象内部维护一个 steps 列表。当你调用chain.invoke(x)时它做的事情非常朴素把x喂给第一个 step拿到输出再喂给第二个 step依次往下。用伪代码表示就是class RunnableSequence: def invoke(self, input): for step in self.steps: input step.invoke(input) return input理解了这一点很多玄学就消失了。比如为什么prompt | model | parser能跑通因为 prompt 的输出是PromptValuemodel 接受PromptValue或消息列表parser 接受AIMessage——每一步的输入输出类型恰好对得上。LCEL 不做类型检查它靠的是接口契约的隐式约定这也是后面很多报错的根源。1.2 为什么用组合而不是直接写函数有人会问我直接写parser(model(prompt.format(x)))不也一样吗功能上确实一样但 LCEL 的组合方式带来几个直接收益这些收益在项目变大之后会非常明显。第一是流式能力自动继承。当你用chain.stream(x)时RunnableSequence 会尝试让每个 step 都走流式路径。如果 model 支持流式parser 也支持流式比如StrOutputParser那么 token 会一个块一个块地穿过整条链最终你拿到的是增量输出。手写函数调用做不到这一点因为普通函数是全有或全无的。第二是批量与异步的免费获得。chain.batch([x1, x2, x3])会自动对每个输入并行处理在异步场景下你不需要自己写asyncio.gather。这在做批量评估、批量生成时省掉大量样板代码。第三是可观测性。RunnableSequence 的每一步都可以挂回调callbacksLangSmith 之类的追踪工具就是靠这个钩子拿到每一步的输入输出、耗时、token 数的。手写函数调用没有这个统一入口。提示如果你只是写一个一次性脚本手写函数调用完全够用不必强行上 LCEL。LCEL 的收益在链会复用、会变长、需要流式或观测的场景下才显现。1.3 一个容易忽略的细节输入输出的类型流转我在早期踩过一个坑把prompt | model | some_function直接拼起来结果报错说some_function收到了AIMessage而不是字符串。原因是model的输出是AIMessage对象不是纯文本。要么在中间加StrOutputParser()要么在函数里自己取.content。这个细节值得单独强调因为 LCEL 的类型流转是隐式的链越长越容易在中间某一步对不上。我的习惯是每写完一条链先用一个最小输入 invoke 一次把中间每一步的输出类型打印出来确认。具体做法是临时插入一个RunnableLambda(lambda x: (print(type(x)), x)[1])虽然土但排查类型问题极快。from langchain_core.runnables import RunnableLambda debug RunnableLambda(lambda x: (print(f[debug] {type(x)}), x)[1]) chain prompt | debug | model | debug | parser跑一次控制台就会告诉你每一步进来的是什么类型。这个技巧在调试复杂链时比读文档快得多。2. Runnable 家族成员逐个拆解LCEL 的 Runnable 不是单一类而是一个家族。每个成员解决一类特定问题混用的时候要清楚各自的语义。这一节我把最常用的几个成员按使用频率 踩坑密度排序讲。2.1 RunnableLambda把普通函数塞进链里RunnableLambda是最灵活的成员它把任意Callable包装成 Runnable。用途极广数据预处理、字段提取、格式转换、条件判断都能用它。from langchain_core.runnables import RunnableLambda def extract_keywords(text: str) - list[str]: return [w.strip() for w in text.split() if w.strip()] keyword_step RunnableLambda(extract_keywords) chain model | StrOutputParser() | keyword_step这里有个实操要点RunnableLambda 默认只接受一个位置参数。如果你的函数需要多个参数要么用functools.partial预绑定要么在函数内部从 dict 里取。我见过不少人写RunnableLambda(lambda a, b: ...)然后报参数错误就是因为这个。另一个细节是异常处理。RunnableLambda 里抛出的异常会直接冒泡中断整条链。如果你希望某一步失败时降级而不是崩溃可以用.with_fallbacks([...])挂一个备用 Runnablerobust_step risky_step.with_fallbacks([RunnableLambda(lambda x: 默认值)])这个 fallback 机制在调用外部 API、解析不稳定输出时特别有用后面第 4 节会展开。2.2 RunnableParallel并行的正确打开方式当你有多个子任务互不依赖时RunnableParallel能让它们并行执行。它的输入是一个 dict每个 value 是一个 Runnable输出也是一个 dictkey 对应各子任务的输出。from langchain_core.runnables import RunnableParallel parallel RunnableParallel( summarysummary_chain, keywordskeyword_chain, sentimentsentiment_chain, ) result parallel.invoke({text: long_text}) # result {summary: ..., keywords: ..., sentiment: ...}这里的关键理解是RunnableParallel 的每个分支收到的是同一个输入。所以如果你的分支需要不同的输入字段要么在分支内部自己取要么在前面加一步做字段分发。我实测下来RunnableParallel 在同步invoke下其实是顺序执行的Python GIL 限制真正的并行要配合abatch或异步ainvoke。这一点文档里写得比较含蓄很多人以为invoke就并行了结果发现耗时没降。要真正并行用await parallel.ainvoke(...)或者对多个输入用chain.abatch(...)。2.3 RunnableBranch条件分支的声明式写法条件分支是 LCEL 里比较绕的一块。RunnableBranch接受一系列(条件, Runnable)对外加一个默认分支按顺序匹配第一个为真的条件。from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: 代码 in x[question], code_chain), (lambda x: 翻译 in x[question], translate_chain), general_chain, # 默认 )注意条件函数接收的是整个输入 dict不是上一步的输出。这是最容易搞混的地方——很多人以为条件判断的是链的中间结果实际上它判断的是进入 branch 时的输入。如果你需要基于中间结果分支得先把中间结果塞进 dict 再进 branch。另一个坑是RunnableBranch的条件是顺序求值的第一个为真的就短路。所以条件顺序有讲究把最具体、最不可能误判的放前面。我见过把宽泛条件放前面导致后面分支永远走不到的情况。2.4 RunnablePassthrough 与 assign数据透传与增量赋值RunnablePassthrough的作用是原样传递输入听起来没用但在 RAG 和并行场景里是刚需。典型用法是在并行分支里保留原始输入from langchain_core.runnables import RunnablePassthrough rag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | model | StrOutputParser() )这里RunnablePassthrough()把用户的问题原样保留和检索到的 context 一起组成 prompt 的输入 dict。没有它question 字段就丢了。RunnablePassthrough.assign()更进一步它在保留原输入的基础上追加新字段chain RunnablePassthrough.assign( contextretriever | format_docs ) | prompt | model这样输入 dict 里原有的字段全部保留同时多了context。这个写法在需要逐步丰富上下文的场景里非常顺手比手动构造 dict 清晰得多。2.5 成员选择速查表为了不让读者在选型上纠结我把这几个成员的适用场景整理成表成员典型场景注意点RunnableLambda任意函数包装、数据转换单参数异常会中断链RunnableParallel多路独立子任务invoke 下非真并行用 ainvokeRunnableBranch条件路由条件判断的是输入而非中间结果RunnablePassthrough保留原始输入RAG 场景必备Passthrough.assign增量追加字段比手动构造 dict 清晰3. 流式输出从模型到前端的完整链路流式是 LCEL 最实用的能力之一但也是配置最容易出问题的地方。很多人的困惑是我明明调了stream为什么前端还是等半天才一次性出来这一节把流式的链路拆开讲。3.1 流式的三个必要条件要让 token 真正逐块到达最终消费者三个环节必须都支持流式模型层底层模型 API 支持流式返回大多数主流模型都支持。链的每一环中间的 parser、转换步骤不能攒完再吐。StrOutputParser支持流式但如果你中间插了一个RunnableLambda做后处理它默认是攒完整个字符串才输出流式就断了。消费端你用for chunk in chain.stream(x)或异步迭代消费而不是invoke。我遇到最多的断流原因是第 2 条——中间插了一个自定义函数。解决办法是把这个函数改写成生成器或者用支持流式透传的包装。LCEL 对生成器函数的处理是如果 RunnableLambda 包装的是生成器函数它会尝试逐块透传。但更稳妥的做法是把后处理挪到流式消费端做链本身保持纯净。3.2 同步流式与异步流式的写法差异同步流式for chunk in chain.stream({topic: LCEL}): print(chunk, end, flushTrue)异步流式在 Web 服务里更常用async for chunk in chain.astream({topic: LCEL}): await send_to_client(chunk)两者语义一致但异步版本在 FastAPI、aiohttp 这类框架里是必须的否则会阻塞事件循环。这里有个实操经验如果你的链里有任何同步阻塞调用比如同步的 HTTP 请求异步流式会被它卡住表现为前几个 token 很快然后卡住。排查方法是逐个环节替换成异步版本定位阻塞点。3.3 流式下的结构化输出难题流式最麻烦的场景是结构化输出比如让模型返回 JSON。因为 JSON 是不完整的中间状态你不能每来一个 chunk 就json.loads。常见做法有两种第一种是流式收集 结束时解析流式只用于展示进度最终解析在流结束后做。适合展示打字效果但最终要结构化结果的场景。第二种是增量解析用支持部分 JSON 解析的库边流边解析已完整的字段。这种做法实现复杂但在需要边生成边渲染的 UI 里体验最好。我的建议是除非产品明确要求边生成边结构化渲染否则优先用第一种。增量解析的边界情况太多转义字符、嵌套结构、数组未闭合维护成本高。注意流式场景下chain.batch和chain.stream不能混用。批量流式要用chain.batch配合每个元素的stream或者用astream_events做更细粒度的事件消费。3.4 astream_events细粒度事件消费当你的链比较复杂需要知道哪个环节开始、哪个环节结束、哪个环节吐了 token时astream_events是更底层的接口。它吐出的是事件流每个事件带event类型如on_chat_model_stream、on_chain_start和name环节名。async for event in chain.astream_events(input, versionv2): if event[event] on_chat_model_stream: print(event[data][chunk].content, end)这个接口在需要区分模型 token和工具调用结果时特别有用。代价是事件流比较冗长消费端要自己过滤。我一般只在需要精细控制 UI 状态比如正在思考正在调用工具正在生成时才用它普通流式用astream就够了。4. 容错、重试与降级让链在生产环境不崩Demo 里跑通的链到了生产环境会因为网络抖动、模型限流、输出格式不符预期等各种原因失败。LCEL 提供了一套容错原语用好了能省掉大量 try-except 样板代码。4.1 with_retry重试的正确姿势robust_model model.with_retry( stop_after_attempt3, wait_exponential_jitterTrue, )with_retry的关键参数是重试次数和退避策略。wait_exponential_jitterTrue会启用指数退避加随机抖动这在多实例并发调用时能有效避免惊群——所有实例同时重试把服务打垮。我见过不加抖动导致重试风暴的案例加了这个参数后稳定性明显提升。要注意的是重试只对可重试错误有意义。比如限流429、超时、5xx 适合重试而参数错误400、内容审核拒绝这类重试也没用反而浪费配额。LCEL 的with_retry默认会重试大部分异常如果你需要精细控制可以传retry_if_exception_type只对特定异常重试。4.2 with_fallbacks降级到备用方案重试解决的是临时故障fallback 解决的是这个方案彻底不行了换一个。典型场景是主模型不可用时切到备用模型或者结构化解析失败时切到宽松解析。primary ChatOpenAI(modelgpt-4o) backup ChatOpenAI(modelgpt-4o-mini) robust primary.with_fallbacks([backup])fallback 是按顺序尝试的前一个抛异常才走下一个。这里有个容易忽略的点fallback 链的输入输出类型必须兼容否则切换时会报类型错误。我建议 fallback 只用于同类型、同接口的替换不要跨类型降级。4.3 结构化输出的容错组合拳让模型返回结构化数据是失败高发区。我的常用组合是结构化输出 重试 宽松解析兜底。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() chain prompt | model | parser robust_chain chain.with_retry(stop_after_attempt2).with_fallbacks( [prompt | model | StrOutputParser() | RunnableLambda(safe_extract_json)] )safe_extract_json是一个自己写的宽松提取函数用正则从文本里抠出 JSON 片段再解析。这套组合拳的逻辑是先严格解析失败重试一次还失败就退到宽松提取。实测下来结构化输出的成功率能从 85% 左右提到 98% 以上。4.4 超时控制LCEL 本身没有统一的超时参数超时要在模型客户端层配置比如ChatOpenAI(timeout30)。但链级别的超时可以用asyncio.wait_for包一层import asyncio result await asyncio.wait_for(chain.ainvoke(input), timeout60)这个在 Web 服务里很重要避免单个请求卡死拖垮整个服务。注意wait_for取消的是协程底层 HTTP 请求是否真正中断取决于客户端实现所以模型层的 timeout 也要配两层保险。5. 调试与观测链跑不通时怎么定位链一长出错时堆栈信息往往指向 LCEL 内部看不出是哪一步的问题。这一节讲几个我常用的定位手段。5.1 全局调试开关最简单的办法是开 LangChain 的全局调试import langchain langchain.debug True开了之后每一步的输入输出都会打印到控制台。缺点是输出非常啰嗦链长的时候刷屏。我一般只在本地排查时临时开生产环境绝不开。5.2 逐步 invoke 定位断点更精准的做法是把链拆开逐步 invoke。比如a | b | c报错先单独a.invoke(x)看输出再b.invoke(a_output)以此类推。这个方法笨但极其有效能精确定位到哪一步的输入输出不符合预期。配合前面提到的RunnableLambda打印类型技巧基本能在几分钟内定位问题。我的经验是LCEL 的报错 80% 是类型不匹配15% 是字段缺失5% 是真正的逻辑错误。按这个比例去排查效率很高。5.3 用回调做轻量观测如果不想引入完整的追踪平台可以用回调自己记录耗时from langchain_core.callbacks import BaseCallbackHandler import time class TimingHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): self._start time.time() def on_chain_end(self, outputs, **kwargs): print(f耗时 {time.time() - self._start:.2f}s)把 handler 传给invoke(config{callbacks: [handler]})即可。这个轻量方案在需要快速看某条链性能时够用比搭一套完整观测系统省事。5.4 常见报错对照表我把踩过的典型报错整理成表方便对照排查报错关键词大概率原因排查方向Expected mapping type某步期望 dict 但收到字符串检查上一步输出类型missing keyprompt 变量没被填充检查输入 dict 字段名not iterable对流式输出做了非迭代操作检查是否误用 invokerate limit触发限流加重试与退避output parsing error结构化解析失败加 fallback 或宽松解析这张表不是穷举但覆盖了我遇到的大部分情况。遇到新报错时我的习惯是先看报错里提到的类型和字段再对照链的输入输出契约通常能快速缩小范围。6. 组合模式实战几个可以直接抄的链结构理论讲完这一节给几个我在项目里反复用的链结构都是经过实战检验的可以直接改改就用。6.1 检索增强生成RAG标准链from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser rag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | model | StrOutputParser() )这个结构的精髓在于RunnablePassthrough保留了原始问题同时并行地跑了检索。注意retriever | format_docs是一个子链它的输入是原始问题输出是格式化后的文档字符串。整个 dict 构造是并行的所以检索和问题透传同时进行。6.2 多步推理链先规划再执行plan_chain prompt_plan | model | StrOutputParser() execute_chain prompt_execute | model | StrOutputParser() full_chain ( RunnablePassthrough.assign(planplan_chain) | RunnablePassthrough.assign(resultexecute_chain) )这里用两次assign逐步丰富上下文第一次把规划结果塞进 dict第二次基于规划结果执行并把结果塞进去。这种逐步 assign的写法比手动构造嵌套 dict 清晰得多而且每一步的输入都包含之前所有字段方便后续环节引用。6.3 带条件路由的客服链router RunnableBranch( (lambda x: x[intent] 退款, refund_chain), (lambda x: x[intent] 查询, query_chain), fallback_chain, )配合前面一步的意图识别可以用一个小模型或规则把请求路由到不同处理链。这个结构在客服、工单分类场景里很常见。注意意图识别那一步的输出要保证intent字段存在否则 branch 的条件会 KeyError。6.4 并行评估链eval_parallel RunnableParallel( relevancerelevance_chain, fluencyfluency_chain, safetysafety_chain, ) scores await eval_parallel.ainvoke({text: generated})做生成质量评估时多个维度并行打分。这里一定要用ainvoke否则三个评估串行跑耗时翻三倍。评估链本身通常是prompt model 结构化解析每个维度独立天然适合并行。7. 性能与成本的几个实操取舍LCEL 用起来顺手之后下一个问题就是怎么跑得更快、更省。这一节讲几个我在实际项目里验证过的取舍。7.1 批量的两种姿势chain.batch(inputs)和chain.abatch(inputs)的区别不只是同步异步。batch在内部会尽量并行但受限于同步模型客户端的连接池abatch配合异步客户端能真正打满并发。做批量任务时我优先用abatch并配合max_concurrency参数控制并发度避免触发限流results await chain.abatch(inputs, config{max_concurrency: 5})max_concurrency这个参数很关键设太大触发限流设太小浪费吞吐。我的经验值是根据模型的 RPM 限制除以单次调用平均耗时来估算先设一个保守值压测后再调。7.2 缓存重复输入的直接收益LCEL 支持在链级别挂缓存from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache set_llm_cache(InMemoryCache())对于相同输入反复调用的场景比如开发调试、固定 prompt 的批量测试缓存能省掉大量重复调用。生产环境可以用持久化缓存如 Redis 后端但要注意缓存 key 是否包含所有影响输出的因素——如果 prompt 模板变了但缓存没失效会拿到旧结果。我的做法是给缓存加版本前缀模板变更时递增版本号。7.3 小模型优先大模型兜底成本优化最有效的手段是分级调用简单任务用小模型复杂或小模型失败时再上大模型。用 LCEL 表达就是tiered small_model.with_fallbacks([large_model])但这里的失败要定义清楚——小模型不会抛异常它只是答得不好。所以更实际的做法是先用小模型再用规则或小模型自评判断质量质量不达标才升级。这个逻辑用 RunnableBranch 表达tiered RunnableBranch( (lambda x: x[quality_score] 0.7, large_model_chain), small_model_chain, )这套分级策略在成本敏感的场景里收益明显实测能省 40% 以上的调用成本代价是链路复杂度上升。7.4 流式与批量的成本差异一个容易被忽略的点流式调用和普通调用的计费方式可能不同。有些模型对流式请求的计费按 token 计和普通调用一致但也有些场景下流式请求的 overhead 更高。做成本核算时要把这个因素考虑进去别只看单价。我的习惯是面向用户的交互场景用流式体验优先后台批处理用普通调用成本优先。8. 写在最后的一点个人体会LCEL 这套东西刚上手时容易陷入两个极端要么觉得它就是语法糖不如直接写函数要么觉得它什么都能干硬把所有逻辑都塞进链里。我用下来的体会是它最适合的位置是**编排层**——负责把 prompt、模型、解析、检索、分支这些环节串起来并统一提供流式、批量、重试、观测这些横切能力。至于具体的业务逻辑、数据清洗、复杂判断老老实实写在 RunnableLambda 包装的普通函数里反而更清晰、更好测。还有一个我踩过几次的坑值得提醒不要为了用 LCEL 而用 LCEL。如果一个流程只有两三步、不需要流式、不需要复用直接写函数调用完全没问题可读性甚至更好。LCEL 的收益在链变长、需要横切能力、需要被多处复用时才显现。判断标准很简单如果你发现自己在反复写重试 流式 日志的样板代码那就是该上 LCEL 的信号了。最后分享一个我常用的调试习惯每写一条新链先用一个最简单的输入跑通invoke再跑stream确认流式没断最后跑batch确认批量正常。这三步都过了这条链基本就能放心用了。至于更复杂的astream_events和回调观测等真的需要精细控制时再上不必一开始就全副武装。