1. 项目概述为什么我们需要 LCEL如果你最近在折腾大模型应用开发尤其是用 LangChain 这类框架大概率已经听过 LCEL 这个名字了。LCEL全称 LangChain Expression Language翻译过来是“LangChain 表达式语言”。乍一听可能有点唬人感觉又是要学一门新语言。但说穿了它其实就是一套用 Python 写链路的“语法糖”和“最佳实践模板”。在 LCEL 出现之前我们是怎么写链路的我早期踩过不少坑。比如用最原始的 Python 函数把一个个提示词模板、大模型调用、输出解析器手动串起来代码里充斥着大量的if-else、临时变量和嵌套回调一个简单的问答链都能写出几十行“面条代码”。后来 LangChain 提供了LLMChain这类基础组件情况好了些但一旦链路复杂起来比如要并行调用、条件分支、动态路由代码又会变得臃肿且难以调试。更头疼的是像流式输出、异步支持、错误处理这些工程化必备的特性自己从头实现不仅费时费力还容易出 bug。LCEL 就是为了解决这些痛点而生的。它通过重载 Python 的管道操作符|让链路的组合变得像搭积木一样直观prompt | model | parser。但这只是表面它的核心价值在于任何用 LCEL 写的链路天生就自带了流式输出、异步支持、批量处理、并行执行、错误重试、中间结果访问等一整套生产级特性。你不用再为这些通用能力写一行代码只需要关注业务逻辑本身。所以LCEL 不是一门需要深度学习的语言而是一种声明式的链路编写范式。它强迫你以“数据流”的视角来设计应用让代码更清晰、更模块化、也更容易维护。接下来我会结合 4 类最典型的 AI 应用场景拆解 LCEL 的工程化写法让你看完就能直接用到自己的项目里。2. 核心设计思路从“过程式”到“声明式”的转变要理解 LCEL 的好处得先看看没有它的时候我们是怎么做的。假设我们有一个最简单的需求用户输入一个问题模型根据问题生成 SQL 语句我们再对 SQL 进行格式化。2.1 传统写法的困境一种典型的“过程式”写法可能是这样的from langchain.llms import OpenAI from langchain.prompts import PromptTemplate import sqlparse def generate_sql(question: str, api_key: str) - str: # 1. 定义提示词 prompt_template PromptTemplate( input_variables[question], template请根据以下问题生成对应的 SQL 查询语句\n问题{question} ) prompt_str prompt_template.format(questionquestion) # 2. 调用模型 llm OpenAI(api_keyapi_key, temperature0) raw_response llm(prompt_str) # 3. 解析并格式化输出 # 假设模型返回的文本里SQL 在 sql 代码块中 import re match re.search(rsql\n(.*?)\n, raw_response, re.DOTALL) if match: sql_code match.group(1).strip() else: # 没找到代码块尝试直接提取 sql_code raw_response.strip() # 4. 格式化 SQL formatted_sql sqlparse.format(sql_code, reindentTrue, keyword_caseupper) return formatted_sql这段代码工作正常但问题很明显结构僵化所有步骤线性耦合在一起如果想在生成 SQL 后增加一个“语法校验”步骤就得修改函数内部结构。缺乏复用prompt_template、llm的初始化逻辑和业务逻辑绑死了其他地方想用同样的模型和提示词组合得重新写。没有流式输出用户必须等待所有步骤网络请求、解析、格式化完成才能看到结果。错误处理简陋模型调用失败、解析失败都没有专门的恢复机制。难以测试因为函数内部状态复杂单元测试需要 mock 多个环节。2.2 LCEL 的声明式哲学LCEL 引导我们用另一种方式思考。我们将上述流程视为一个数据转换管道输入用户问题。节点1将问题填入提示词模板输出格式化后的提示词字符串。节点2将提示词字符串发送给大模型输出原始响应文本。节点3从原始文本中提取 SQL 代码块。节点4格式化 SQL 代码。在 LCEL 中每个节点都是一个实现了Runnable协议的对象。Runnable可以是一个提示词模板、一个大模型、一个输出解析器甚至是一个自定义函数。我们用|把它们连接起来形成一个RunnableSequence可运行序列。from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI # 推荐使用新的集成包 from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough import sqlparse import re # 定义各个可运行节点 prompt PromptTemplate.from_template(请根据以下问题生成对应的 SQL 查询语句\n问题{question}) model ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 自定义一个 SQL 提取和格式化的 Runnable def extract_and_format_sql(text: str) - str: match re.search(rsql\n(.*?)\n, text, re.DOTALL) sql_code match.group(1).strip() if match else text.strip() return sqlparse.format(sql_code, reindentTrue, keyword_caseupper) # 使用 LCEL 声明链路 sql_chain prompt | model | StrOutputParser() | extract_and_format_sql这个sql_chain本身就是一个Runnable对象。你可以调用它的invoke、batch、astream等方法。关键优势立刻显现流式输出调用astream你可以看到提示词生成、模型逐字输出、解析器工作的全过程。异步原生直接调用ainvoke或astream即可。易于组合这个sql_chain可以作为一个更大的链路的子模块。标准接口所有Runnable都有统一的输入输出规范便于调试和监控。这种声明式的写法将“做什么”业务逻辑和“怎么做”执行引擎解耦了。你负责用|定义数据流LangChain 负责以最高效、最稳定的方式执行它。这是 LCEL 工程化价值的核心。3. 四类典型链路的 LCEL 工程化写法理解了核心思想我们来看实战。下面这四类链路几乎覆盖了 80% 的 AI 应用场景。3.1 基础问答链不仅仅是 prompt model最简单的链是问答但生产环境的要求不止于此。我们需要考虑系统提示词、聊天历史、以及安全的输出解析。场景构建一个客服助手需要遵守安全规范并以 JSON 格式返回。传统写法痛点手动拼接聊天历史字符串操作容易出错输出解析需要额外步骤流式输出实现麻烦。LCEL 工程化写法from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain_core.output_parsers import JsonOutputParser from langchain_core.pydantic_v1 import BaseModel, Field from typing import List # 1. 定义严格输出的数据结构Pydantic Model class AssistantResponse(BaseModel): 助手的标准化响应 answer: str Field(description对用户问题的直接回答) confidence: float Field(description回答的置信度0-1之间, ge0, le1) needs_human: bool Field(description是否需要转接人工客服, defaultFalse) # 2. 构建声明式提示词。MessagesPlaceholder 用于动态插入聊天历史。 system_prompt 你是一个专业的客服助手。请根据对话历史回答用户问题。 你的回答必须友好、准确。如果问题超出你的能力范围或涉及敏感内容请明确表示无法回答并建议转人工。 请始终以以下 JSON 格式输出 { answer: 你的回答内容, confidence: 0.95, needs_human: false } prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history), # 动态占位符 (human, {user_input}) ]) # 3. 初始化模型和解析器 model ChatOpenAI(modelgpt-4, temperature0.1) parser JsonOutputParser(pydantic_objectAssistantResponse) # 绑定 Pydantic 模型进行强解析 # 4. 使用 LCEL 组装链路 # 注意输入需要是一个包含 chat_history 和 user_input 的字典 customer_service_chain prompt | model | parser # 使用示例 from langchain.schema import AIMessage, HumanMessage chat_history [ HumanMessage(content你们的产品保修期多久), AIMessage(content我们的产品标准保修期是12个月。) ] current_input {user_input: 那保修范围包括哪些, chat_history: chat_history} try: response customer_service_chain.invoke(current_input) print(f回答{response[answer]}) print(f置信度{response[confidence]}) if response[needs_human]: print(建议转接人工客服。) except Exception as e: # JsonOutputParser 会在模型输出不符合 JSON 或 Pydantic 模型时抛出异常 print(f解析输出时出错{e}) # 这里可以添加降级逻辑例如 fallback 到一个普通文本链工程化要点解析使用 Pydantic 定义输出模式JsonOutputParser配合 Pydantic Model能强制模型输出结构化的 JSON并自动进行类型验证和约束检查如confidence必须在 0-1 之间。这比用正则表达式提取稳定得多。MessagesPlaceholder动态管理历史这是处理多轮对话的优雅方式。LCEL 链的输入是一个字典你可以自由地构造和更新chat_history这个键对应的消息列表链本身不需要做任何修改。统一的错误处理由于解析器可能失败整个invoke调用应该被try-except包裹。在生产系统中这里通常是配置重试或降级策略的关键点。流式支持如果你想实现打字机效果只需将invoke改为astream解析器会流式地生成并解析 JSON 的各个字段。实操心得对于关键业务链强烈建议使用JsonOutputParserPydantic。这相当于为模型的“自由发挥”加了一个牢笼能极大提升输出结果的稳定性和可预测性。调试时如果解析失败可以先去掉解析器看看模型的原始输出到底偏离了多少。3.2 检索增强生成RAG链管理复杂上下文RAG 是当前最主流的 AI 应用范式之一。其核心挑战在于如何将检索到的文档片段高效、合理地整合到提示词中并防止上下文过长。场景基于内部知识库的智能问答系统。传统写法痛点检索器、提示词组装、模型调用、后处理逻辑分散处理多文档时上下文窗口容易爆炸难以对检索步骤进行监控和调优。LCEL 工程化写法from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import RunnablePassthrough, RunnableLambda from langchain.schema.output_parser import StrOutputParser from langchain.text_splitter import RecursiveCharacterTextSplitter # 假设我们已经有了一个向量数据库 vectorstore embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 这里仅为示例实际应从持久化存储加载 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 检索4个相关片段 # 1. 定义提示词模板明确要求模型基于上下文回答 template 你是一个知识库助手。请严格根据以下提供的上下文信息来回答问题。 如果你在上下文中找不到答案就明确说“根据已知信息无法回答此问题”不要编造信息。 上下文信息 {context} 问题{question} 请给出基于上下文的回答 prompt ChatPromptTemplate.from_template(template) # 2. 定义格式化检索结果的函数 def format_docs(docs): 将检索到的 Document 列表格式化为一个单一的上下文字符串。 return \n\n.join([f来源 {i1}: {doc.page_content} for i, doc in enumerate(docs)]) # 3. 使用 LCEL 组装 RAG 链 # 关键使用 RunnablePassthrough 来传递初始问题并并行地触发检索 rag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | ChatOpenAI(modelgpt-3.5-turbo-16k, temperature0) # 使用长上下文模型 | StrOutputParser() ) # 使用示例 answer rag_chain.invoke(LangChain 中 LCEL 的主要优点是什么) print(answer) # 如果你想获得检索到的源文档用于引用或评估 from langchain.callbacks import get_openai_callback with get_openai_callback() as cb: # 先检索 question LangChain 中 LCEL 的主要优点是什么 retrieved_docs retriever.invoke(question) print(f检索到 {len(retrieved_docs)} 个文档片段) for doc in retrieved_docs: print(f- 片段摘要: {doc.page_content[:200]}...) # 再执行完整链这里 retriever 会被再次调用生产环境可优化 answer rag_chain.invoke(question) print(f\n最终答案{answer}) print(f\n本次消耗{cb})工程化要点解析使用字典组合输入{“context”: retriever | format_docs, “question”: RunnablePassthrough()}是 LCEL 的精华之一。它创建了一个并行分支question直接传递而context分支则执行retriever检索然后将结果通过format_docs函数格式化。这两个分支的结果最终会合并成一个字典作为prompt的输入。这比手动调用检索器并拼接字符串清晰得多。RunnablePassthrough的作用它像一个透明的管道将输入原封不动地传递到指定位置。在这里它确保了原始问题question能流入到最终的提示词模板中。上下文管理format_docs函数让你能精细控制如何将多个文档片段呈现给模型。例如你可以加上元数据来源、分数或者进行截断、摘要防止超出模型上下文限制。可观测性通过get_openai_callback或集成 LangSmith可以方便地追踪检索步骤返回了哪些文档、消耗了多少 Token这对于调试和优化检索策略如search_kwargs至关重要。注意事项在真正的生产环境retriever可能会被调用两次一次在链中一次在监控中。为了避免重复计算可以将检索步骤的结果缓存起来或者使用更高级的Runnable分支控制。此外对于超长文档考虑在format_docs中加入智能截断或摘要逻辑。3.3 条件判断与路由链构建智能工作流很多应用需要根据用户输入的内容决定走哪条处理路径。比如判断用户是想查询天气、订餐还是聊天然后分发给不同的子链处理。场景一个多功能的 AI 助手能处理查询、创作、分析等不同类型任务。传统写法痛点需要写一堆if-elif-else语句判断逻辑和业务逻辑耦合严重增加新功能需要修改核心路由函数违反开闭原则。LCEL 工程化写法from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnableBranch, RunnableLambda from typing import Literal # 0. 定义路由判断链让模型自己判断意图 class RouteQuery(BaseModel): 路由判断的模型 destination: Literal[query, creative, analysis, small_talk] Field(description根据问题判断应路由到的处理模块) reason: str Field(description做出此判断的理由) from langchain_core.output_parsers import JsonOutputParser route_parser JsonOutputParser(pydantic_objectRouteQuery) route_prompt ChatPromptTemplate.from_messages([ (system, 你是一个请求分类器。请分析用户的问题判断其最适合由哪个处理模块处理。 可选模块 - query: 事实性查询如“珠穆朗玛峰多高” - creative: 创意生成如“写一首关于春天的诗” - analysis: 逻辑分析或比较如“比较Python和Java的优缺点” - small_talk: 闲聊或问候如“你好吗” 请以 JSON 格式输出包含 destination 和 reason 字段。), (human, {question}) ]) # 路由判断链 route_chain route_prompt | ChatOpenAI(modelgpt-3.5-turbo, temperature0) | route_parser # 1. 定义各个子处理链 # 查询链 query_prompt ChatPromptTemplate.from_template(你是一个百科全书。请准确、简洁地回答这个事实性问题{question}) query_chain query_prompt | ChatOpenAI() | StrOutputParser() # 创作链 creative_prompt ChatPromptTemplate.from_template(你是一个创意作家。请根据以下主题进行创作{question}) creative_chain creative_prompt | ChatOpenAI(temperature0.8) | StrOutputParser() # 分析链 analysis_prompt ChatPromptTemplate.from_template(你是一个分析师。请从多个角度严谨地分析以下问题{question}) analysis_chain analysis_prompt | ChatOpenAI(temperature0.2) | StrOutputParser() # 闲聊链 small_talk_prompt ChatPromptTemplate.from_template(你是一个友好的伙伴。请进行轻松愉快的闲聊{question}) small_talk_chain small_talk_prompt | ChatOpenAI(temperature0.9) | StrOutputParser() # 2. 使用 RunnableBranch 定义路由逻辑 # RunnableBranch 接收一个 (condition, runnable) 对的列表 branch RunnableBranch( (lambda x: x[destination] query, query_chain), (lambda x: x[destination] creative, creative_chain), (lambda x: x[destination] analysis, analysis_chain), (lambda x: x[destination] small_talk, small_talk_chain), # 默认分支 RunnableLambda(lambda x: f抱歉我暂时无法处理‘{x.get(question, )}’这类请求。) ) # 3. 组装完整的工作流链 # 第一步判断意图。输入是原始问题输出是包含 destination 和原问题的字典。 # 第二步根据 destination 路由到对应的子链。 full_chain { # 保留原始问题并添加路由结果 question: RunnablePassthrough(), destination: route_chain | (lambda x: x[destination]) # 只提取 destination 字段 } | branch # 将合并后的字典输入 branch # 使用示例 questions [ 巴黎铁塔有多高, 帮我写一个关于人工智能的短故事开头。, 从就业市场和学习曲线分析前端和后端开发哪个更适合初学者, 今天天气真好, 如何制造一台永动机 ] for q in questions: print(f\n问题{q}) response full_chain.invoke(q) print(f回答{response}\n{-*40})工程化要点解析用模型做路由判断这是更鲁棒的方式。相比基于关键词的硬编码规则让大模型理解用户意图更准确也能处理更模糊的请求。RouteQuery这个 Pydantic 模型确保了路由判断的输出是结构化的。RunnableBranch是核心它类似于if-elif-else但每个分支都是一个Runnable。条件是一个接收输入字典并返回布尔值的函数。这使得路由逻辑本身也成为了可组合、可测试的模块。数据流的精心设计注意full_chain的组装。我们构造了一个字典同时包含原始问题question和路由结果destination。这个字典正是branch所期望的输入。branch会根据destination的值将整个字典包含question传递给对应的子链。子链的提示词模板只使用question字段忽略了destination。默认分支RunnableBranch的最后一个元素是默认分支用于处理未匹配任何条件的情况是必不可少的错误兜底。实操心得对于路由链一定要记录日志。记录下每个请求的原始问题、模型判断的destination和reason以及最终执行的子链。这对于评估路由准确性、发现 Bad Case 并迭代优化路由提示词至关重要。可以将route_chain的输出完整地传递下去而不仅仅是destination方便日志记录。3.4 并行与汇总链处理复杂任务分解有些任务需要同时做多件事然后汇总结果。比如用户给了一篇文章需要同时总结摘要、提取关键词、分析情感倾向。场景文档多维度分析。传统写法痛点需要手动管理多线程/异步处理结果收集和异常代码结构复杂。LCEL 工程化写法from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnableLambda, RunnableParallel from typing import Dict, Any # 1. 定义三个并行分析任务的子链 summary_prompt ChatPromptTemplate.from_template(请用一段话总结以下文本的核心内容\n文本{document}) keyword_prompt ChatPromptTemplate.from_template(请从以下文本中提取3-5个核心关键词\n文本{document}) sentiment_prompt ChatPromptTemplate.from_template(请分析以下文本的情感倾向积极、消极或中性并简要说明理由\n文本{document}) model ChatOpenAI(modelgpt-3.5-turbo, temperature0) summary_chain summary_prompt | model | StrOutputParser() keyword_chain keyword_prompt | model | StrOutputParser() sentiment_chain sentiment_prompt | model | StrOutputParser() # 2. 使用 RunnableParallel 进行并行执行 # RunnableParallel 会同时执行其内部的多个 Runnable并将结果合并到一个字典中。 parallel_analysis_chain RunnableParallel({ summary: summary_chain, keywords: keyword_chain, sentiment: sentiment_chain }) # 注意每个子链的输入都需要是包含 document 键的字典。 # 3. 定义汇总链将并行结果整合成一份报告 def generate_report(analysis_results: Dict[str, Any]) - str: 根据并行分析的结果生成最终报告 report f文档分析报告 1. 内容摘要 {analysis_results[summary]} 2. 核心关键词 {analysis_results[keywords]} 3. 情感分析 {analysis_results[sentiment]} --- 分析完成 --- return report report_chain RunnableLambda(generate_report) # 4. 组装完整链先并行分析再生成报告 # 输入需要是一个字典如 {document: 一些很长的文本...} full_analysis_chain parallel_analysis_chain | report_chain # 使用示例 sample_document 人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。 人工智能领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。 人工智能从诞生以来理论和技术日益成熟应用领域也不断扩大可以设想未来人工智能带来的科技产品 将会是人类智慧的“容器”。人工智能可以对人的意识、思维的信息过程的模拟。人工智能不是人的智能 但能像人那样思考、也可能超过人的智能。 result full_analysis_chain.invoke({document: sample_document}) print(result) # 5. 进阶带流式输出的并行观察哪个任务先完成 print(\n--- 流式执行演示 ---) async def stream_analysis(): async for chunk in parallel_analysis_chain.astream({document: sample_document}): # chunk 是一个字典键是 summary, keywords, sentiment # 值可能是部分结果如果模型支持流式 for key, value in chunk.items(): if value: # 只打印有更新的部分 print(f[{key.upper()} 更新]: {value[:100]}...) import asyncio asyncio.run(stream_analysis())工程化要点解析RunnableParallel是并行利器它接受一个字典字典的值是各个Runnable子链。当执行时它会并发地运行所有这些子链如果后端支持异步。这比手动写asyncio.gather简洁安全得多。输入输出的统一RunnableParallel要求所有子链的输入格式一致。这里每个子链的提示词都期望一个包含document字段的字典。full_analysis_chain的输入也是同样的格式。结果聚合RunnableParallel的输出是一个字典键是之前定义的如summary值是对应子链的输出。这个字典可以直接传递给下一个环节如report_chain进行处理。流式输出的并行观察当调用astream时RunnableParallel会流式地输出每个子链的进度。这在处理耗时不同的任务时非常有用你可以看到哪个分析任务最先返回结果提升用户体验。注意事项并行执行会同时消耗多个模型的 Token。请务必注意成本控制和速率限制。对于免费或低限额的 API可能需要加入限流或队列机制。RunnableParallel本身不提供限流功能你需要在外层进行控制或者使用 LangChain 的RunnableWithFallbacks等组件进行容错。4. 高级技巧与生产环境考量掌握了以上四种模式你已经能应对大部分场景。但要真正用于生产还需要一些“打磨”。4.1 错误处理与重试网络抖动、模型过载、输出格式意外都会导致失败。LCEL 链可以方便地包裹错误处理逻辑。from langchain.schema.runnable import RunnableWithFallbacks from tenacity import retry, stop_after_attempt, wait_exponential import asyncio # 定义一个可能失败的基础链 unstable_chain prompt | model | parser # 方法1使用 tenacity 进行自动重试适用于 invoke retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def invoke_with_retry(chain, input_data): return chain.invoke(input_data) # 方法2使用 RunnableWithFallbacks (更 LCEL 风格) # 首先定义一个降级链 fallback_prompt ChatPromptTemplate.from_template(抱歉系统正忙。简化回答{question}) fallback_chain fallback_prompt | ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) | StrOutputParser() # 用主链和降级链创建带容错的链 robust_chain RunnableWithFallbacks( runnableunstable_chain, fallbacks[fallback_chain], exceptions_to_handle(Exception,) # 捕获所有异常生产环境应更具体 ) # 使用 robust_chain.invoke()如果主链失败会自动尝试降级链4.2 链路监控与调试LangSmithLCEL 链与 LangSmith 深度集成这是生产调试的“神器”。import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] your_langchain_api_key os.environ[LANGCHAIN_PROJECT] My_Production_Project # 现在所有 chain.invoke/ainvoke/batch/astream 调用都会被自动记录到 LangSmith # 你可以在网页上查看详细的调用链、每一步的输入输出、耗时、Token 消耗甚至模型响应的延迟。 result your_lcel_chain.invoke({question: test})在 LangSmith UI 中你可以清晰地看到 LCEL 链被分解成的每一个Runnable步骤像看流程图一样调试极大提升了复杂链路的可观测性。4.3 性能优化批处理与异步LCEL 链原生支持批处理和异步能极大提升吞吐量。# 批处理 (batch) inputs [{question: f问题{i}} for i in range(10)] results your_chain.batch(inputs) # 顺序处理 # 或使用异步批处理以获得更好性能 # results await your_chain.abatch(inputs) # 异步流式 (astream_log) # astream_log 是调试神器它返回一个迭代器不仅包含最终输出还包含链中每一步的日志信息。 async for chunk in your_chain.astream_log({question: ...}, include_names[ChatOpenAI]): # chunk 包含操作名、输入输出等可以实时打印到控制台或发送到监控系统 print(chunk)4.4 自定义 Runnable无限扩展可能当内置组件不够用时你可以轻松创建自定义的Runnable。from langchain.schema.runnable import RunnableConfig from langchain_core.runnables import RunnableSerializable from typing import Any class MyCustomRunnable(RunnableSerializable[str, str]): 一个将输入字符串转换为大写并添加前缀的自定义 Runnable prefix: str OUTPUT: # 可序列化的配置参数 def invoke(self, input: str, config: RunnableConfig | None None) - str: # 同步调用逻辑 return self.prefix input.upper() async def ainvoke(self, input: str, config: RunnableConfig | None None) - str: # 异步调用逻辑 # 这里简单复用同步逻辑实际可能涉及异步IO return self.invoke(input, config) # 像使用内置组件一样使用它 custom_chain prompt | model | StrOutputParser() | MyCustomRunnable(prefixRESULT: )通过继承RunnableSerializable你的自定义类就能无缝融入 LCEL 生态支持invoke、batch、astream等所有操作并且可以被序列化保存。5. 常见问题与避坑指南在实际使用 LCEL 的过程中我总结了一些高频问题和解决方案。Q1: 提示词模板中的变量名和输入字典的键对不上怎么办A: 这是最常见的错误。LCEL 在运行时会严格检查。确保你的PromptTemplate或ChatPromptTemplate中定义的变量如{context}、{question}在链的当前输入字典中都有对应的键。使用RunnablePassthrough或RunnableLambda来调整输入数据的结构。Q2: 如何访问链的中间结果A: 有几种方法。对于调试使用astream_log。如果需要在后续步骤中使用中间结果可以在链中插入一个RunnableLambda来打印或保存它。更结构化的方式是使用 LangSmith 进行追踪。Q3: 链变得很长很复杂难以阅读和维护。A: 这是过度使用|的征兆。良好的实践是分层设计。将功能独立的子链路定义成变量例如retrieval_subchain retriever | format_docs然后将这些子链组合成主链main_chain a | b | retrieval_subchain | c。给每个子链起一个清晰的名字。Q4: 流式输出时如何只输出最终结果不显示中间步骤A: 默认的astream会输出每个Runnable的结果。如果你只想看到模型生成的 Token可以在调用模型后使用StrOutputParser的流式特性或者使用astream_eventsAPI 进行更精细的控制只订阅你关心的事件类型如on_chat_model_stream。Q5: 在异步框架如 FastAPI中使用 LCEL 链有什么需要注意的A: 确保在整个异步上下文async函数中使用ainvoke、abatch、astream等异步方法。避免在异步函数中调用同步的invoke这可能会阻塞事件循环。另外考虑为每个请求创建独立的链实例或妥善管理共享链的状态避免并发问题。Q6: 如何对 LCEL 链进行单元测试A: 由于 LCEL 链是纯函数式的给定输入产生确定输出测试非常方便。你可以 mock 其中的组件例如用FakeListLLM替换真实的ChatOpenAI用固定的列表模拟Retriever。然后直接对链调用invoke并断言输出。LCEL 的模块化设计使得每个Runnable都可以独立测试。从“面条代码”到声明式的数据流管道LCEL 带来的不仅是代码的简洁更是一种工程思维的提升。它迫使你思考应用的模块边界和数据流向其结果就是更健壮、更易维护、也更容易扩展的 AI 应用。刚开始可能需要适应这种“组装”思维但一旦习惯你就会发现构建复杂的 AI 工作流从未如此清晰和高效。