LCEL：LangChain表达式语言，AI应用开发的工程化实践

1. 从概念到实践为什么我们需要 LCEL如果你最近在折腾大模型应用开发尤其是基于 LangChain 这类框架那么“LCEL”这个词大概率已经在你眼前晃过无数次了。LCEL全称 LangChain Expression Language官方文档会告诉你它是一种声明式的、用于组合链路的语言。但说人话就是它是一套让你用写 Python 代码一样直观的方式去“组装”和“描述”复杂 AI 应用流程的语法糖和规范。在 LCEL 出现之前我们是怎么写链路的无非是面向对象那一套定义一个类在__init__里初始化各种组件LLM、提示词模板、输出解析器然后在run或invoke方法里手动组织调用逻辑。代码写起来冗长链路一复杂各种if-else和临时变量就让代码可读性急剧下降。更头疼的是像流式输出、异步调用、并行处理、错误处理这些高级功能每加一个都得写一堆“胶水代码”不仅容易出错而且不同人写出来的风格迥异后期维护简直是噩梦。LCEL 就是为了解决这些工程化痛点而生的。它的核心思想是“组合优于继承”。你把 LLM、提示词、工具、条件判断等都看作是一个个独立的“可调用对象”然后通过|运算符在 LCEL 里读作“管道”或“然后”把它们像乐高积木一样连接起来。这样构建出来的链路本身也是一个可调用对象可以继续被组合。这种声明式的写法让链路的意图变得异常清晰代码简洁到有时让你怀疑“这就完了”更重要的是LCEL 不是空架子它为这种声明式链路“免费”带来了强大的运行时特性开箱即用的流式支持、异步调用、并行执行、重试机制、回调函数集成等等。你不用再为这些通用能力写一行代码LCEL 底层都帮你处理好了。这才是它被称为“工程化写法”的关键——它把最佳实践固化到了语言层面。所以当你在热搜里看到“RAG工程化”、“AI应用开发”、“AI工程实践”这些词时LCEL 往往是背后那个关键的实现手段。它回答的不是“能不能做”的问题而是“怎么做得更优雅、更健壮、更易维护”的问题。接下来我们就抛开理论直接进入实战看看在四种最典型的 AI 应用场景里如何用 LCEL 写出专业级的代码。2. 四类典型链路的 LCEL 工程化写法详解2.1 基础问答链从“能跑”到“好用”的蜕变最简单的链路莫过于用户输入问题模型给出答案。用传统写法你可能需要初始化模型、组装提示词、调用、解析输出。用 LCEL三行代码就能定义核心链路。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 1. 定义提示词模板 prompt ChatPromptTemplate.from_template(你是一个专业的助手。请回答以下问题\n\n问题{question}) # 2. 初始化模型 model ChatOpenAI(modelgpt-4o, temperature0) # 3. 使用 LCEL 组合链路模板 - 模型 - 输出为字符串 basic_chain prompt | model | (lambda x: x.content) # 等价于basic_chain prompt | model | StrOutputParser()这里prompt | model是 LCEL 的核心。它创建了一个RunnableSequence对象。当你调用basic_chain.invoke({question: 什么是量子计算})时数据流会从左到右执行先将question变量填入模板生成完整的提示词然后将提示词传给ChatOpenAI模型调用最后通过一个简单的 lambda 函数或官方的StrOutputParser提取模型返回消息中的content字段。工程化要点1善用输出解析器上面的 lambda 函数在简单场景可行但不规范。生产环境应使用 LangChain 内置的OutputParser它们功能更强大比如能处理模型可能返回的AIMessage、ToolCall等多种复杂结构。from langchain_core.output_parsers import StrOutputParser basic_chain prompt | model | StrOutputParser()StrOutputParser()会智能地提取出最终的文本内容代码更清晰也便于未来替换为其他解析器如解析成 JSON 的JsonOutputParser。工程化要点2配置化与参数传递LCEL 链路本身是惰性的只有在invoke或stream时才会执行。你可以通过with_config方法或在调用时传入config参数动态调整链路行为比如为某次调用切换模型或添加特定元数据。# 方法1创建时绑定配置 chain_with_config basic_chain.with_config(run_namemy_qa_chain, metadata{version: 1.0}) # 方法2调用时传入配置 result basic_chain.invoke( {question: 解释一下Transformer。}, config{configurable: {model_name: gpt-3.5-turbo}} # 可被某些组件读取 )这种配置化能力使得同一条链路可以轻松适配开发、测试、生产等不同环境。2.2 检索增强生成链构建高效 RAG 系统的核心RAG 是当前最火的应用模式之一其核心链路是检索 - 组织上下文 - 生成。用传统方式写检索器、向量数据库、上下文拼接等逻辑交织在一起。LCEL 能让这条链路变得模块化和清晰。假设我们有一个已加载文档的向量库检索器retriever。from langchain_core.runnables import RunnablePassthrough # 定义提示词模板它期待两个输入变量context 和 question template 基于以下上下文回答用户的问题。如果你不知道答案就说你不知道不要编造。 上下文 {context} 问题{question} prompt ChatPromptTemplate.from_template(template) # 定义 RAG 链路 rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | StrOutputParser() )这段代码是 LCEL 组合艺术的体现。我们使用一个字典来定义链路的输入映射context: retriever表示context这个变量的值是通过retriever这个可调用对象生成的。retriever.invoke(question)会返回相关文档片段。question: RunnablePassthrough()表示question这个变量的值直接来自上游传递下来的输入即用户原始问题。RunnablePassthrough是一个特殊的“传递”节点它不做任何改变只是把输入原样传递到指定位置。当调用rag_chain.invoke(LangChain 是什么)时LCEL 会先并行执行或按需执行这个字典里的每个可调用对象用问题去检索retriever同时把问题本身传递下去。然后将执行结果一个包含context和question键的字典喂给prompt模板。之后的流程就和基础链一样了。工程化要点3优化检索与上下文处理上面的基础 RAG 链存在“上下文过长”或“信息冗余”的风险。在生产中我们常需要更精细的控制。from langchain_core.runnables import RunnableLambda def format_docs(docs): 将检索到的文档列表格式化为一个连贯的字符串。 return \n\n.join([f来源 {i1}: {doc.page_content} for i, doc in enumerate(docs)]) def reciprocal_rank_fusion(retrieved_results, k60): 一个简单的 RRF 重排序算法示例用于融合多个检索器的结果。 # 简化实现实际应用需更完整 fused_scores {} for results in retrieved_results: for rank, doc in enumerate(results): doc_id doc.id # 假设文档有唯一ID fused_scores[doc_id] fused_scores.get(doc_id, 0) 1.0 / (rank k) # 根据融合分数排序并返回文档 sorted_docs sorted(...) return sorted_docs # 假设我们有两个不同的检索器一个基于向量一个基于关键词 vector_retriever ... bm25_retriever ... # 构建更复杂的 RAG 链 advanced_rag_chain ( { question: RunnablePassthrough(), vector_docs: vector_retriever, keyword_docs: bm25_retriever, } | RunnableLambda(lambda x: { question: x[question], context: format_docs(reciprocal_rank_fusion([x[vector_docs], x[keyword_docs]])) }) | prompt | model | StrOutputParser() )这里我们展示了如何并行多路检索同时使用向量检索和关键词检索。自定义处理节点使用RunnableLambda包装自定义函数format_docs和reciprocal_rank_fusion对检索结果进行格式化、融合、重排序。这是 LCEL 与自定义业务逻辑对接的关键。结构化数据流清晰地展示了数据如何从原始问题经过多个处理阶段最终形成给模型的上下文。这种写法将复杂的 RAG 流程分解为一个个可测试、可替换的组件极大地提升了代码的可维护性。2.3 条件判断与路由链实现动态工作流很多复杂的 AI 应用不是一条直线走到底的需要根据模型输出或用户输入动态决定下一步做什么。这就是“智能路由”。LCEL 通过RunnableBranch和RunnableLambda让路由逻辑变得直观。一个经典场景是根据用户问题类型决定是调用通用模型、搜索工具还是查询数据库。from langchain_core.runnables import RunnableBranch, RunnableLambda from langchain_core.prompts import PromptTemplate # 第一步分类判断 classifier_prompt PromptTemplate.from_template( 请将用户问题分类。类别包括 1. general_qa: 通用知识问答如“天为什么是蓝的” 2. search: 需要实时信息的问题如“今天北京的天气如何” 3. query_db: 需要查询内部知识库的问题如“我们公司Q3的销售目标是多少” 用户问题{question} 只返回类别名称不要有其他任何文字。 ) classifier_chain classifier_prompt | model | StrOutputParser() # 第二步定义各分支的处理链路 general_qa_chain (PromptTemplate.from_template(回答{question}) | model | StrOutputParser()) # 假设 search_chain 和 query_db_chain 是预先定义好的工具调用链 search_chain ... query_db_chain ... # 第三步使用 RunnableBranch 构建路由 branch RunnableBranch( (lambda x: search in x[topic], search_chain), # 条件1如果 topic 包含 “search”走搜索链 (lambda x: query_db in x[topic], query_db_chain), # 条件2 general_qa_chain # 默认分支 ) # 第四步组合完整链路先分类再根据分类结果路由 full_router_chain ( { question: RunnablePassthrough(), topic: classifier_chain # 这里 classifier_chain 的输入是上游的 question } | branch )执行过程当你调用full_router_chain.invoke(“今天股市行情怎么样”)时输入question同时传递给RunnablePassthrough()保留原问题和classifier_chain进行分类。classifier_chain输出分类结果比如“search”。此时组合成一个字典{“question”: “今天股市...” “topic”: “search”}传递给branch。RunnableBranch按顺序评估每个条件。第一个条件lambda x: “search” in x[“topic”]为真因此执行search_chain。search_chain会接收到这个完整的字典作为输入。工程化要点4路由条件的设计与调试路由条件可以是任何返回布尔值的函数。对于复杂逻辑建议将条件判断也封装成独立的 LCEL 链或RunnableLambda确保可测试性。另外一定要设置一个合理的默认分支general_qa_chain作为“安全网”处理未匹配到任何明确条件的情况。调试路由链时可以单独测试分类器和每个分支链确保它们按预期工作。LCEL 链的.invoke()方法非常方便进行单元测试。2.4 多步骤协作链串联与并联的复杂编排对于需要模型多次调用、工具交替使用的复杂任务如规划、执行、反思的 ReAct 模式或需要多专家协作的任务LCEL 的管道和组合能力更能发挥威力。假设我们要写一个链先让模型根据复杂任务制定一个计划步骤列表然后并行执行所有步骤最后总结结果。from langchain_core.runnables import RunnableParallel # 步骤1计划生成链 planner_prompt ChatPromptTemplate.from_template( 任务{task} 请将这个复杂任务分解为不超过3个清晰的子步骤。以JSON格式输出包含一个steps字段其值为字符串列表。 例如{{“steps”: [“步骤1描述”, “步骤2描述”]}} ) planner_chain planner_prompt | model | JsonOutputParser() # 步骤2定义一个处理单个步骤的链假设是个简单的问答 def execute_step(step_info): # step_info 可能是一个字典包含步骤描述和上下文 step step_info[step] # 这里可以调用工具、查询数据库等 # 简化起见我们让模型“模拟”执行 execution_prompt ChatPromptTemplate.from_template(模拟执行步骤{step}。请给出执行结果摘要。) execution_chain execution_prompt | model | StrOutputParser() return execution_chain.invoke({step: step}) # 步骤3总结链 summarizer_prompt ChatPromptTemplate.from_template( 原始任务{task} 已执行的步骤及结果 {step_results} 请基于以上信息给出任务的最终总结。 ) summarizer_chain summarizer_prompt | model | StrOutputParser() # 组合完整的多步骤协作链 def map_steps(steps): 将步骤列表映射为可并行执行的任务列表 return [RunnableLambda(lambda sstep: execute_step({step: s})) for step in steps] multi_step_chain ( { task: RunnablePassthrough(), steps: planner_chain # 输出是 {steps: [...]} } | RunnableLambda(lambda x: { task: x[task], step_results: RunnableParallel( **{fstep_{i}: chain for i, chain in enumerate(map_steps(x[steps][steps]))} ) }) | RunnableLambda(lambda x: { task: x[task], step_results: \n.join([f- {k}: {v} for k, v in x[step_results].items()]) }) | summarizer_chain )这个链看起来复杂但结构清晰计划阶段planner_chain生成步骤列表。并行执行阶段这是关键。我们使用RunnableParallel来并行执行所有子步骤。RunnableParallel(**dict)接受一个字典其中每个值都是一个Runnable。它会并行调用所有Runnable并返回一个字典键是原字典的键值是对应的执行结果。结果汇总与总结将并行执行的结果格式化为字符串传递给总结链。工程化要点5并行化与错误处理RunnableParallel极大地提升了链路的效率。但要注意并行执行的任务之间应尽可能独立避免资源竞争。对于可能失败的任务如网络调用LCEL 链支持通过.with_retry()方法添加重试机制或者通过try-except封装在RunnableLambda中返回一个兜底结果避免整个链路因一个步骤失败而崩溃。from langchain_core.runnables import RunnableConfig import asyncio # 为链添加重试 robust_chain some_chain.with_retry(stop_after_attempt3, wait_exponential_jitterTrue) # 在 RunnableLambda 中处理错误 safe_executor RunnableLambda(lambda x: { result: execute_step(x) if some_condition else Step skipped due to condition., error: None })这种将复杂流程分解为可并行、可容错的组件的模式是构建鲁棒性强的生产级 AI 应用的基础。3. LCEL 的高级特性与生产级调优掌握了基本链路的写法我们还需要了解 LCEL 如何支撑起一个健壮的生产系统。这涉及到流式、异步、监控、调试等方方面面。3.1 流式输出与异步调用流式输出是提升用户体验的关键。LCEL 链天生支持流式只需将invoke改为stream。# 同步流式逐词输出 for chunk in basic_chain.stream({question: 讲一个故事}): print(chunk, end, flushTrue) # 模拟打字机效果 # 异步流式在异步框架如 FastAPI 中非常有用 async for chunk in basic_chain.astream({question: 讲一个故事}): # 通过 WebSocket 发送 chunk 到前端 await websocket.send_text(chunk)LCEL 保证了流式过程中每个中间步骤如提示词填充、模型调用产生的令牌都能被逐步 yield 出来而不是等到最后才一次性输出。异步调用则能更好地利用 I/O 等待时间提高吞吐量。# 异步调用单个链 result await basic_chain.ainvoke({question: ...}) # 并行异步调用多个输入批量处理 tasks [basic_chain.ainvoke({question: q}) for q in question_list] results await asyncio.gather(*tasks)工程化要点6流式中间结果有时我们不仅想流式模型的最终输出还想在过程中知道当前处于哪个步骤例如显示“正在检索...”、“正在生成...”。LCEL 通过回调系统或直接访问流的中间节点来实现。from langchain_core.runnables import RunnableConfig chain_with_metadata (prompt | model | StrOutputParser()).with_config( run_nameMyStreamingChain, metadata{version: 2.0} ) # 在 stream 或 astream 时可以配置 callbacks 来捕获事件 # 更简单的方式是利用 astream_events APILangChain 较新版本提供它能提供更细粒度的事件流。 async for event in chain_with_metadata.astream_events( {question: ...}, versionv1 ): kind event[event] if kind on_chain_start: print(fStarting chain: {event[name]}) elif kind on_llm_start: print(fCalling LLM...) elif kind on_llm_stream: print(fToken: {event[data][chunk].content}, end)这对于构建交互式前端或实现复杂的进度提示至关重要。3.2 链路监控、日志与调试当链路部署到生产环境可观测性就成了必须。LCEL 与 LangSmith 等追踪平台深度集成可以无缝记录每次链路的执行详情、输入输出、耗时、token 使用量等。import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] your-api-key os.environ[LANGCHAIN_PROJECT] my-production-project # 现在每次调用 .invoke() 或 .stream()都会自动记录到 LangSmith result basic_chain.invoke({question: ...})在 LangSmith UI 上你可以看到链路的可视化图谱、每个节点的输入输出、耗时和成本。这是调试复杂链路、定位性能瓶颈、分析异常情况的利器。对于本地调试可以使用.invoke()的config参数传入一个callbacks列表使用ConsoleCallbackHandler在终端打印执行日志。工程化要点7链路版本管理与 A/B 测试利用 LCEL 链的可组合性可以轻松实现链路版本管理。例如将链路的定义提示词、模型选择、参数等抽取到配置文件中。通过切换配置就能快速上线新版本或进行 A/B 测试。import yaml # 从配置文件加载链路配置 with open(chain_config_v1.yaml) as f: config_v1 yaml.safe_load(f) def build_chain_from_config(config): prompt ChatPromptTemplate.from_template(config[prompt_template]) model ChatOpenAI(**config[model_params]) return prompt | model | StrOutputParser() chain_v1 build_chain_from_config(config_v1) # 加载另一个配置就是 chain_v2结合部署平台的路由能力可以将不同版本的链路分配给不同比例的用户流量从而科学地评估效果。3.3 性能优化与成本控制AI 应用的成本主要来自大模型 API 调用。LCEL 提供了几种优化思路缓存对确定性高的步骤如提示词填充、某些工具查询结果进行缓存。可以使用RunnableLambda配合functools.lru_cache或外部缓存如 Redis。模型选择在链路中根据问题复杂度动态选择模型。例如简单问题用便宜快速的gpt-3.5-turbo复杂问题再用gpt-4。这可以通过RunnableBranch实现。提前退出在某些检查点判断是否还有必要继续执行后续昂贵步骤。例如在 RAG 链中如果检索器返回的文档为空或相关性极低可以直接返回“未找到相关信息”而无需调用大模型。from langchain_core.runnables import RunnableBranch def check_context_sufficiency(docs): 检查检索到的文档是否足够回答问题 if not docs or len(docs[0].page_content) 50: # 简单示例 return insufficient return sufficient # 带提前退出的 RAG 链 smart_rag_chain ( {context: retriever, question: RunnablePassthrough()} | RunnableLambda(lambda x: { data: x, sufficiency: check_context_sufficiency(x[context]) }) | RunnableBranch( (lambda x: x[sufficiency] insufficient, RunnableLambda(lambda x: 抱歉知识库中未找到相关信息。)), (lambda x: { context: x[data][context], question: x[data][question] } | prompt | model | StrOutputParser()) ) )这个链在检索后立即评估上下文是否充足如果不充足就直接返回预设回复避免了不必要的模型调用费用。4. 常见“坑点”与最佳实践实录在实际项目中踩过一些坑后我总结出以下经验这些在官方文档里不一定会强调。坑点1RunnableLambda中状态管理混乱RunnableLambda包装的函数应该是纯函数或无副作用的。避免在其中修改外部变量或依赖全局状态因为链可能被并发调用。如果需要共享状态如计数器、缓存应使用依赖注入或将其封装在具有适当锁机制的单例服务中。坑点2过度嵌套导致可读性下降LCEL 虽然强大但过度使用管道和嵌套字典会让代码难以阅读。当一个链的步骤超过 5-6 步时考虑将其拆分为多个子链并通过.with_config(run_name...)为子链命名这样在 LangSmith 追踪中也会更清晰。# 不推荐一长串管道 complex_chain step1 | step2 | step3 | step4 | step5 | step6 # 推荐拆分子链并命名 sub_chain_a (step1 | step2).with_config(run_namedata_preprocessing) sub_chain_b (step3 | step4).with_config(run_namecore_logic) sub_chain_c (step5 | step6).with_config(run_namepost_processing) complex_chain sub_chain_a | sub_chain_b | sub_chain_c坑点3错误处理粒度太粗默认情况下链中某个节点抛出异常整个链就会失败。对于生产系统我们需要更细粒度的容错。除了前面提到的.with_retry()还可以使用RunnableBranch创建错误处理分支或者用try-except包裹RunnableLambda返回一个表示失败的特定值让下游节点决定如何处理。from langchain_core.runnables import RunnableBranch def safe_tool_call(tool_input): try: return some_unreliable_tool(tool_input) except Exception as e: return {error: str(e), result: None} safe_tool_chain RunnableLambda(safe_tool_call) # 在后续链路中检查错误 error_aware_chain ( safe_tool_chain | RunnableBranch( (lambda x: x.get(error) is not None, RunnableLambda(lambda x: f工具调用失败{x[error]})), (lambda x: process_successful_result(x[result])) ) )最佳实践1为所有链添加配置和元数据养成使用.with_config()的习惯为链设置run_name和metadata。这不会影响功能但在追踪和监控时它是区分不同链路、统计不同版本性能的生命线。最佳实践2编写单元测试LCEL 链是纯函数式的组合非常适合单元测试。你可以单独测试每个Runnable组件也可以测试整个链在特定输入下的输出。使用unittest或pytest模拟外部依赖如模型、检索器确保链路的逻辑正确性。def test_basic_chain(): # 使用模拟模型 mock_model Mock(specChatOpenAI) mock_model.invoke.return_value AIMessage(contentMocked answer) test_chain prompt | mock_model | StrOutputParser() result test_chain.invoke({question: test}) assert result Mocked answer mock_model.invoke.assert_called_once()最佳实践3关注输入输出模式LCEL 链的输入和输出有明确的模式。使用chain.input_schema.schema()和chain.output_schema.schema()可以打印出链期望的输入和输出的 JSON Schema。这在构建前后端接口、文档化 API 时非常有用能确保数据格式的正确性。LCEL 不是银弹但它为 LangChain 应用开发提供了一套强大、优雅且符合工程学规范的范式。从简单的提示词链到复杂的多智能体工作流LCEL 的声明式组合思想都能让你更专注于业务逻辑本身而不是繁琐的流程控制代码。掌握它意味着你掌握了构建可维护、可观测、高性能 AI 应用的关键工具。