1. 项目概述为什么我们需要LangChain表达式语言如果你最近在折腾大语言模型应用开发大概率已经听过LangChain的大名。作为一个旨在简化LLM应用构建的框架LangChain确实提供了丰富的模块和工具。但真正上手后很多开发者包括我自己都曾有过这样的困惑链条Chains的构建虽然直观但随着业务逻辑复杂度的提升代码会迅速变得冗长且难以维护。调试一个多步骤的链就像在迷宫里找路输入输出在哪一步变了形中间状态是什么经常让人一头雾水。这正是LangChain Expression LanguageLCEL要解决的核心痛点。它不是一个全新的框架而是LangChain内部一套声明式的、用于组合链式组件的“语法糖”和底层规范。你可以把它理解为构建LangChain应用的“乐高说明书”和“通用接口”。在LCEL出现之前我们可能用SequentialChain硬编码流程或者写一堆嵌套的回调函数。而LCEL通过引入Runnable协议将链中的每一个步骤无论是调用LLM、查询向量数据库还是执行一个Python函数都抽象成统一的可执行对象然后用管道符|像连接Linux命令一样将它们优雅地组合起来。简单来说LCEL让构建复杂AI工作流变得像搭积木一样清晰、灵活且易于调试。它解决了传统链式编程的三大难题组合的灵活性、中间过程的可见性以及生产级特性如流式输出、并行处理的内置支持。无论你是想快速搭建一个检索增强生成RAG系统还是设计一个包含条件判断和循环的智能体AgentLCEL都能提供一套简洁而强大的表达方式。接下来我将从一个实践者的角度带你深入LCEL的肌理看看它如何改变我们构建LLM应用的方式。2. LCEL核心设计哲学与核心接口解析2.1 万物皆可“运行”理解Runnable协议LCEL的基石是Runnable协议。这是一个关键的理念转变在LCEL的世界里一切可执行单元都是Runnable对象。这包括LLM模型如ChatOpenAI,ChatAnthropic提示词模板PromptTemplate输出解析器StrOutputParser,JsonOutputParser工具Tool甚至是你自定义的Python函数Runnable协议定义了标准的输入输出方法最主要的是.invoke()同步调用和.ainvoke()异步调用。这意味着无论底层是调用一个API、查询数据库还是执行计算你都可以用统一的方式去“运行”它。这种抽象带来了巨大的好处。首先它提供了极致的组合性。因为接口统一任何Runnable的输出都可以作为另一个Runnable的输入组合变得无比自然。其次它带来了强大的可观测性。由于每个步骤都是独立的Runnable框架可以更容易地追踪数据流为调试和监控铺平道路。2.2 管道操作符LCEL的“灵魂语法”LCEL最直观、最优雅的特性就是使用管道操作符|来组合Runnable。这借鉴了Unix shell和现代函数式编程的思想让数据流的方向一目了然。# 一个经典的LCEL链模板 - 模型 - 解析器 from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser prompt ChatPromptTemplate.from_template(讲一个关于{topic}的笑话) model ChatOpenAI(modelgpt-4) output_parser StrOutputParser() # 使用管道符组合 chain prompt | model | output_parser # 运行链 result chain.invoke({topic: 程序员}) print(result)这段代码的可读性极高。从左到右清晰地展示了数据流向一个包含topic的字典输入到prompt生成格式化后的提示词传递给model生成AI的原始响应最后经由output_parser提取出我们需要的字符串文本。这种声明式的写法比用一系列函数调用嵌套要清晰得多尤其是在链的长度增加时。2.3 核心接口方法详解invoke, batch, streamRunnable协议不仅统一了调用方式还内置了对常见生产场景的优化支持。除了基础的.invoke()你还需要熟悉另外两个核心方法.batch(): 批量处理当你需要处理一组输入时使用.batch()可以显著提升效率特别是底层模型API支持批量调用时。LCEL会智能地优化执行过程。# 批量处理多个主题 inputs [{topic: 程序员}, {topic: 科学家}, {topic: 猫}] results chain.batch(inputs) # 结果是一个包含三个笑话的列表.stream(): 流式输出这是LCEL相比传统链的一个巨大优势。对于需要实时显示AI生成内容的场景如聊天应用流式输出至关重要。LCEL原生支持从链中的任何一步开始流式输出。# 流式输出模型生成的内容 for chunk in chain.stream({topic: 哲学家}): print(chunk, end, flushTrue) # 你会看到文字一个词一个词地“流”出来注意流式输出的粒度取决于你从链的哪一步开始stream。model.stream()会流式返回原始的AI响应块如OpenAI的delta而chain.stream()链末端是StrOutputParser则会流式返回已解析的文本字符串。理解这一点对实现正确的用户体验很重要。这些方法在所有的Runnable组合中都保持一致这意味着你为简单链编写的流式或批量处理代码在迁移到复杂链时依然可以正常工作极大地提升了代码的可复用性和可维护性。3. 构建复杂工作流超越线性管道LCEL的强大远不止于线性管道。它提供了一系列“原语”让你能够构建带有分支、合并、条件判断甚至循环的复杂逻辑。3.1 分支与路由根据条件选择执行路径想象一个场景用户输入一个问题你需要先判断它是需要查询知识库的“技术问题”还是可以直接回答的“通用问题”。这需要分支逻辑。LCEL通过RunnableBranch来实现这一点。它接收一个由条件可运行对象组成的列表。条件是一个接收输入并返回布尔值的函数或Runnable。from langchain.schema.runnable import RunnableBranch # 定义条件判断函数 def is_technical(query: dict) - bool: # 简单的关键词判断实际应用中可能使用一个分类模型 tech_keywords [错误, API, 配置, 安装, 代码] return any(keyword in query[question] for keyword in tech_keywords) # 定义不同分支的处理链 general_chain ChatPromptTemplate.from_template(请友好地回答这个通用问题{question}) | ChatOpenAI() | StrOutputParser() technical_chain ChatPromptTemplate.from_template(你是一个技术专家请解答{question}) | ChatOpenAI() | StrOutputParser() # 构建分支 branch RunnableBranch( (is_technical, technical_chain), # 如果是技术问题走技术链 general_chain # 否则走通用链 ) # 运行分支 result branch.invoke({question: 如何解决Python的ModuleNotFoundError?}) print(result) # 会走technical_chain3.2 并行处理与结果合并RunnableParallel很多时候我们需要同时执行多个任务然后合并它们的结果。例如在RAG系统中我们可能并行查询多个不同的数据源。RunnableParallel在LCEL中常用字典语法{}表示就是用于此目的。from langchain.schema.runnable import RunnableParallel # 模拟两个不同的信息检索过程 def fetch_wiki_info(topic: dict): return f来自百科的{topic[topic]}信息... def fetch_news_info(topic: dict): return f关于{topic[topic]}的最新新闻... # 构建并行任务 parallel RunnableParallel({ wiki_info: fetch_wiki_info, # 注意函数会被自动包装为RunnableLambda news_info: fetch_news_info }) # 执行并行任务返回一个字典 info_dict parallel.invoke({topic: 人工智能}) # 输出: {wiki_info: 来自百科的人工智能信息..., news_info: 关于人工智能的最新新闻...} # 将并行结果传递给下游链 summary_chain ChatPromptTemplate.from_template( 基于以下信息进行总结 百科信息{wiki_info} 新闻信息{news_info} 请生成一份关于{topic}的摘要。 ) | ChatOpenAI() | StrOutputParser() # 组合先并行获取信息再总结 full_chain parallel | summary_chain result full_chain.invoke({topic: 人工智能})RunnableParallel会并发地执行所有任务如果使用异步调用并将结果收集到一个字典中字典的键就是你指定的名称。这个字典可以直接作为后续链的输入后续链的提示词模板可以通过相同的键名来引用这些值。3.3 动态配置与上下文传递在复杂应用中我们经常需要根据运行时信息来动态配置链的某一部分。例如根据用户选择来切换不同的LLM模型或配置参数。LCEL通过RunnableConfig和with_config方法支持这一点。更常见的一个需求是“上下文传递”在链的早期步骤中产生的某个值需要在很下游的步骤中被使用。LCEL通过RunnablePassthrough来优雅地解决这个问题。RunnablePassthrough像一个传送带它原封不动地传递输入但同时也可以被用来复制或添加新的数据到流中。from langchain.schema.runnable import RunnablePassthrough # 假设我们有一个链先提取用户查询中的关键词然后用这个关键词去检索最后生成回答 extract_keyword_chain ChatPromptTemplate.from_template(从以下问题中提取核心关键词{question}) | ChatOpenAI() | StrOutputParser() def retrieve_docs(keyword: str): # 模拟检索过程 return [f文档1关于{keyword}, f文档2关于{keyword}] # 使用RunnablePassthrough来保留原始的question同时添加检索到的文档 chain ( RunnablePassthrough.assign( # .assign() 用于添加新字段 keywordextract_keyword_chain, # 这一步的输出会赋值给keyword字段 docslambda x: retrieve_docs(extract_keyword_chain.invoke(x)) # 使用lambda进行复杂操作 ) | ChatPromptTemplate.from_template(基于以下文档回答{question}\n{docs}) | ChatOpenAI() | StrOutputParser() ) # 输入中需要包含question answer chain.invoke({question: LangChain有什么优势})在这个例子中RunnablePassthrough.assign()创建了一个新的字典它包含了原始输入的所有字段即question并新增了keyword和docs两个字段。这样在最终的提示词模板中我们就可以同时访问到原始的question和后续生成的docs。4. 生产化特性流式、回退与监控LCEL在设计之初就考虑了生产环境的需求内置了许多开箱即用的高级特性。4.1 深度解构流式输出如前所述.stream()方法提供了流式能力。但生产级应用往往需要更细粒度的控制。LCEL允许你订阅链中特定Runnable的流式事件这通过with_listeners()或更底层的astream_events()Beta API实现。astream_events可以让你在生成token、链开始/结束、工具调用等关键节点收到事件是实现复杂UI交互如实时显示“正在思考…”、“正在调用工具”的基础。# 这是一个Beta API接口可能变化但概念很重要 async for event in chain.astream_events({topic: AI}, versionv1): kind event[event] if kind on_chat_model_stream: # 收到一个新的token token event[data][chunk].content print(token, end, flushTrue) elif kind on_tool_start: print(f\n[正在调用工具: {event[name]}]...)4.2 优雅降级与回退机制在分布式系统中服务不可用是常态。LCEL的RunnableWithFallbacks允许你为链设置一个或多个备选方案。from langchain.schema.runnable import RunnableWithFallbacks primary_chain ChatOpenAI(modelgpt-4, temperature0) fallback_chain ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 备用模型 chain_with_fallback RunnableWithFallbacks( runnableprimary_chain, fallbacks[fallback_chain] ) try: # 如果primary_chain因速率限制或宕机失败会自动尝试fallback_chain result chain_with_fallback.invoke(你好) except Exception as e: # 如果所有备选都失败才会抛出异常 print(f所有链都失败了: {e})你可以为整个复杂链设置回退也可以只为其中某个脆弱的环节如调用特定API的工具设置。这大大增强了应用的鲁棒性。4.3 链路追踪与调试调试一个多步LCEL链比调试传统代码化的链要直观。因为每个步骤都是独立的Runnable你可以轻松地单独测试任何一步。此外利用LangSmithLangChain的官方监控平台你可以自动记录每一次链执行的详细轨迹包括每个步骤的输入输出、耗时、token使用量等。即使不使用LangSmith你也可以通过with_config为链注入自定义的回调函数来打印日志或收集指标。from langchain.callbacks.tracers import ConsoleCallbackHandler # 在调用时传入回调在控制台看到详细的执行步骤日志 result chain.invoke( {topic: 测试}, config{callbacks: [ConsoleCallbackHandler()]} )5. 从传统Chain迁移到LCEL实战指南与避坑如果你已有一些基于旧版Chain类构建的代码迁移到LCEL不仅能获得更好的性能和维护性还能解锁新特性。迁移过程通常是渐进的。5.1 迁移策略与模式对比旧模式自定义Chain类from langchain.chains import LLMChain prompt PromptTemplate(...) llm ChatOpenAI(...) old_chain LLMChain(llmllm, promptprompt) result old_chain.run({input: ...})LCEL模式new_chain prompt | llm | StrOutputParser() result new_chain.invoke({input: ...})迁移的关键在于识别旧链中的每个步骤并将其转化为Runnable对象。LLMChain通常对应PromptTemplate | LLM。TransformChain用于数据转换可以很方便地用RunnableLambda将Python函数包装为Runnable替代。5.2 常见问题与解决方案实录在实际迁移和开发中我遇到过不少典型问题这里分享一些排查技巧问题1输入输出格式不匹配症状链在某个步骤抛出验证错误如“期望字典得到字符串”。根因LCEL对类型要求更严格。每个Runnable都有预期的输入和输出模式。排查使用.input_schema.schema()和.output_schema.schema()打印中间步骤的输入输出模式。确保上一步的输出类型与下一步的输入类型兼容。解决使用RunnableLambda进行简单的格式转换。例如如果模型输出的是AIMessage而下一个普通函数需要字符串可以插入RunnableLambda(lambda x: x.content)。问题2流式输出不工作或格式混乱症状调用.stream()后没有输出或者输出的是难以解析的对象。根因链的末端可能不是一个能产生“可流”内容的组件或者流出的粒度不是你想要的。排查从模型层开始流式输出逐步向后排查。model.stream()是否正常然后尝试(prompt | model).stream()。解决确保最终输出解析器如StrOutputParser支持流式。对于自定义解析你可能需要实现一个也支持astream方法的Runnable。问题3并行任务未真正并发症状使用了RunnableParallel但感觉执行速度没有提升。根因如果你使用同步的.invoke()方法调用包含RunnableParallel的链LCEL默认可能不会并发执行取决于具体实现和I/O类型。对于真正的并发必须使用异步。解决始终对包含I/O操作如API调用、数据库查询的并行任务使用异步。async def run_parallel(): result await chain.ainvoke(input_data)问题4配置如API Key无法传递症状在链中使用了with_config动态设置模型参数但似乎没生效。根因配置的传递是链式的但某些自定义的RunnableLambda函数可能不会自动继承配置。解决在需要配置的RunnableLambda内部通过context参数显式获取配置。from langchain.schema.runnable.config import RunnableConfig def my_func(input, config: RunnableConfig): api_key config.get(configurable, {}).get(api_key) # ... 使用api_key5.3 性能优化要点善用.batch()对于批量任务优先使用.batch()而非循环调用.invoke()。LCEL和底层模型如OpenAI API都可能对批量处理进行优化。异步是王道任何涉及网络I/OLLM调用、工具调用、检索的链都应使用异步接口.ainvoke(),.abatch(),.astream()以充分利用现代Python的异步并发能力避免阻塞。缓存中间结果对于计算成本高或结果稳定的步骤可以考虑集成像langchain.cache支持内存、SQLite、Redis等这样的缓存层避免重复计算。精简提示词在LCEL链中提示词模板可能被频繁渲染和传递。确保模板本身是高效的避免不必要的复杂字符串操作。6. 高级模式与自定义扩展当你熟练掌握LCEL基础后可以探索更高级的模式来构建真正强大和灵活的应用。6.1 构建有状态的会话链LCEL链本质上是无状态的函数式组合。要实现多轮对话状态记忆需要将历史消息作为输入的一部分进行管理。通常的模式是使用RunnableWithMessageHistory它包装你的链并自动管理一个“历史记录”对象。from langchain.memory import ChatMessageHistory from langchain.schema.runnable.history import RunnableWithMessageHistory # 1. 定义核心链无状态 core_chain prompt | model | output_parser # 2. 用RunnableWithMessageHistory包装 chain_with_history RunnableWithMessageHistory( core_chain, get_session_historylambda session_id: ChatMessageHistory(), # 根据session_id获取历史存储 input_messages_keyinput, # 输入中用户消息的键 history_messages_keyhistory # 输入中历史消息的键 ) # 3. 调用时传入config指定session_id config {configurable: {session_id: user_123}} # 第一次调用历史为空 response1 chain_with_history.invoke({input: 你好我叫小明}, configconfig) # 第二次调用链的输入会自动包含上一次的对话历史 response2 chain_with_history.invoke({input: 我刚才说我叫什么}, configconfig)6.2 创建自定义Runnable组件虽然LCEL提供了丰富的内置组件但总有一天你需要创建自己的。只要遵循Runnable协议即可。from typing import Any, Optional from langchain.schema.runnable import Runnable class MyCustomRetriever(Runnable): def __init__(self, database_url): self.client connect_to_db(database_url) def invoke(self, query: str, config: Optional[RunnableConfig] None) - List[Document]: 同步调用接口 # 你的检索逻辑 docs self.client.search(query) return docs async def ainvoke(self, query: str, config: Optional[RunnableConfig] None) - List[Document]: 异步调用接口可选但推荐实现 # 异步检索逻辑 return await self.client.async_search(query) # 通常还需要实现stream, batch, astream_events等方法但invoke/ainvoke是最低要求 # 现在你的自定义检索器可以无缝接入LCEL管道 chain MyCustomRetriever(sqlite:///data.db) | format_docs | prompt | model | parser6.3 实现条件循环与智能体逻辑LCEL本身不包含显式的while循环语法但可以通过组合RunnableBranch和递归来实现复杂的、有条件终止的循环这正是构建智能体Agent的核心。一个典型的ReAct模式智能体循环可以抽象为根据当前状态问题、历史、工具结果决定下一步行动思考、调用工具、结束。执行行动。更新状态。如果未结束回到步骤1。在LCEL中你可以将“单步决策与执行”定义为一个Runnable然后通过一个外部循环控制器可能是另一个Runnable或普通Python函数来反复调用它直到满足终止条件。LCEL的优雅之处在于这个“单步决策”链本身可以由prompt | model | output_parser加上Tool调用组合而成保持了高度的模块化。7. 生态整合与最佳实践最后谈谈LCEL如何与LangChain生态的其他部分协同工作以及我总结的一些最佳实践。7.1 与LangChain工具Tools和智能体Agents的集成LCEL与LangChain的工具系统是天作之合。工具本身就是Runnable。这意味着你可以轻松地将工具插入到任何LCEL链中。from langchain.tools import Tool def search_api(query: str) - str: # 模拟搜索 return f搜索结果: {query} search_tool Tool.from_function( funcsearch_api, nameSearch, description用于搜索网络信息 ) # 一个简单的“搜索后回答”链 chain ( RunnablePassthrough.assign( search_resultsearch_tool # 直接将工具作为Runnable使用 ) | ChatPromptTemplate.from_template(基于{search_result}回答{query}) | ChatOpenAI() | StrOutputParser() )而新一代的LangChain智能体如create_react_agent其内部实现也大量依赖LCEL来构建可预测、可调试的执行图。7.2 测试与可维护性建议单元测试每个Runnable得益于LCEL的声明式和模块化特性你可以且应该对链中的每个Runnable组件进行独立的单元测试。Mock掉LLM和外部API的调用测试你的提示词模板、解析逻辑和自定义函数。集成测试完整链使用固定的输入和Mock的LLM如langchain.chat_models.FakeChatModel来测试整个链的数据流是否符合预期。版本化提示词模板将提示词模板存储在代码之外如数据库、配置文件便于A/B测试和迭代更新而无需重新部署代码。利用LangSmith进行全链路监控在生产环境中强烈建议集成LangSmith。它能可视化每次调用帮助你快速定位性能瓶颈哪个步骤最耗时和逻辑错误输入输出在何时何地出现了偏差。7.3 何时使用LCEL何时使用传统方式坚定选择LCEL当你构建新的、涉及多个步骤的、可能需要流式输出、批量处理或复杂逻辑分支/并行的链时。对于绝大多数应用场景LCEL都是更优解。传统Chain或自定义类可能仍有用武之地对于一些极其简单、一次性的脚本比如就调用一次LLM直接使用LLMChain也未尝不可。或者当你需要封装一个非常复杂、有大量内部状态和方法的“黑盒”模块时将其实现为一个自定义类然后在LCEL链中通过RunnableLambda来调用也是一种混合策略。从我个人的实践经验来看一旦适应了LCEL的思维模式就很难再回到过去那种命令式、面条式的链构建方式了。它带来的代码清晰度、可组合性和可维护性提升是实实在在的。刚开始学习时可能会觉得那些RunnableBranch、RunnableParallel的语法有些陌生但请坚持练习。你可以从改造一个旧的、简单的链开始逐步尝试添加流式输出、错误回退等特性很快就能体会到它的强大与优雅。
