051、LangChain Expression Language (LCEL) 入门今天下午调了个线上报警LangChain 的 chain 返回空字符串可单独测 LLM 时响应非常正常。我盯着管道符看了十分钟心想这|难道还能把 prompt 吞了最后发现是模板里少了个花括号——变量名写错LLM 拿到的是空模板它当然给你空答案。这事儿不怪 LangChain怪我没理解 LCEL 到底在背后干了什么。先别急着写代码LCEL 不是什么玄学它只是把“可运行的东西”统一成一种协议。这里的“可运行”包括模型、提示模板、解析器甚至任意函数。管道符|也不是 Python 的位或它被重载成了“把左边输出接到右边输入”的粘合剂。你写prompt | model | parser实际上构造了一个RunnableSequence这个过程是惰性的——你拿着这个 chain 对象以为它是结果其实它是一张待执行的流程图。所以第一个坑就是别把 chain 当结果打印。很多人会这样chainprompt|model|StrOutputParser()print(chain)# 这打印出来的是一堆 Runnable 信息不是你的答案我当时就想这 chain 是不是坏了后来才反应过来你还没invoke呢。LCEL 的每个 Runnable 都有统一的接口invoke、batch、stream还有异步的ainvoke、abatch、astream。你调chain.invoke({question: ...})的时候它才真正开始跑。再聊一个我踩过的坑中间过程看不见。LCEL 的好处是简洁坏处是太简洁了中间变量全被藏起来。你想知道 prompt 填完之后长什么样model 返回原始内容又是什么光靠打印 chain 是不行的。这时候要用RunnableLambda或者RunnablePassthrough当探针。比如这样fromlangchain_core.runnablesimportRunnableLambdadefdebug_step(x):print(这一步的输入是,x)# 别怕乱开发期就该这么干returnx chainprompt|RunnableLambda(debug_step)|model|StrOutputParser()这里有个细节RunnableLambda接收一个函数函数的返回值会作为下一个节点的输入。如果你的函数只打印不返回那下一步就成None了。我当时就干过这种蠢事输出一行日志结果模型收到的全是None报错参都没提。还有一个容易绕晕的地方管道符只能串行不能并行。你如果想让多个任务同时跑比如一个分支做摘要另一个分支做关键词提取最后合并结果怎么办用RunnableParallel。fromlangchain_core.runnablesimportRunnableParallel summarize_chainprompt_summary|model|StrOutputParser()keywords_chainprompt_keywords|model|StrOutputParser()parallel_chainRunnableParallel(summarysummarize_chain,keywordskeywords_chain)resultparallel_chain.invoke({text:...})# result[summary] 和 result[keywords] 就都齐了注意RunnableParallel的输出是一个字典字典的 key 就是你给它的参数名。这里容易踩坑的是如果你两个分支都用同一个 prompt 模板变量名冲突后写的会覆盖先写的。我当时写了两段 prompt一个叫prompt1一个叫prompt2结果变量名都用了question最后两个分支收到的其实是同一个值查了半天。再来说说条件分支。LCEL 不是流程图工具它默认没有 if-else。但你可以用RunnableBranch或者直接包一个RunnableLambda做路由。我一般用后者因为更直观而且不需要记额外的 API。defroute(x):ifx.get(type)summary:returnsummarize_chain.invoke(x)else:returnkeywords_chain.invoke(x)routerRunnableLambda(route)这里有个性能陷阱RunnableLambda内部如果再调用其他 chain 的invoke会阻塞式等待也没法流式。如果你要真·流式得用RunnableBranch去构造静态分支而不是在函数里动态 invoke。不过说实话大多数后端场景里一次性 invoke 就够了流式只是锦上添花。说到流式我这次排查问题的时候还注意到一个现象model.stream()明明是一个 token 一个 token 往外蹦但接了StrOutputParser之后chain.stream()却像是攒了一整句才输出。乍一看以为解析器吞了流。其实不是StrOutputParser是支持流式的它会逐个 token 透传。问题往往出在你自定义的解析器上——如果你在 parser 的parse方法里做了聚合处理比如把内容存到 list 里那流式就会被缓存直到 parse 结束才吐一个完整结果。解决方法是让 parser 继承BaseTransformOutputParser或者干脆别在 parser 里做状态聚合把它放到最后一个RunnableLambda里。这属于流式设计的细节不讲太多但遇到“模型明明在流程序却不流”的时候首先怀疑你的自定义解析器别怀疑框架。再补一个更隐蔽的坑RunnableParallel里的分支如果有一个抛异常整个 chain 就全停了其他分支拿不到结果。这也是 LCEL 的默认行为。我一开始以为它类似asyncio.gather能容错后来看源码才发现它内部是concurrent.futures的ThreadPoolExecutor异常会直接穿透。所以你要做容错得在每个分支外面套RunnableLambda在里面 try-except返回一个默认值。调试 LCEL 真的需要一些耐心。我不会一上来就写复杂 chain而是先用普通函数把逻辑理清楚确认每一步的输入输出类型然后再搬进 LCEL。有一次我花了一下午调一个并行 chain最后发现是RunnableParallel的输入需要是字典我传了个字符串进去它倒是没报错只是把字符串当成了字典的 key 去遍历结果每个字符都被当作一次请求直接把 API 打爆了。这不是虚构是真事。所以我的个人经验是LCEL 适合线性流程和少量分支千万别为了“优雅”去构建一棵复杂的 Runnable 树。生产环境里可维护性远比简洁重要。在你把逻辑翻译成 LCEL 之前先写一堆普通 Python 函数再考虑哪一段需要并行、哪一段需要流式最后才用管道符组装。组装时尽量每个节点都用RunnableLambda显式命名哪怕它只是传个参数这样日志里能看到数据流到哪一步断了。还有个建议开发期在关键节点加RunnableLambda(debug_step)打印输入输出。但上线前记得删掉或者换成logging。你不想在日志里看到一堆 token 级的花花绿绿除非你在写流式演示 demo。关于 LCEL 的入门先掌握这些就够。它不复杂但确实需要适应这种“声明式”的写法。以后碰到 chain 不干活别先怀疑 AI 造反按我说的拆平它一步步看数据。祝你的管道永远通畅。