从玩具到工程：LCEL如何构建可靠可维护的AI应用链路

1. 从“玩具”到“工程”为什么我们需要LCEL如果你最近在折腾LangChain或者任何基于大语言模型的应用开发大概率已经听过LCELLangChain Expression Language这个名字。刚开始接触时我也有过疑惑不就是把几个LLM调用串起来吗用Python函数写个流程控制或者用LangChain的SequentialChain不也一样能跑通为什么非要学一套新的“语言”这个疑问在我第一次尝试把一个本地跑通的“玩具”Demo部署到线上并面对真实用户流量时得到了残酷的解答。当时我用最直观的if-else加函数调用写了一个包含用户意图分类、信息检索和总结回复的流程。在本地测试时一切完美但一上线就遇到了几个致命问题某个环节的API调用超时导致整个流程卡死无法优雅降级我想给中间某个步骤的输出加个缓存发现要侵入式地修改多处代码更头疼的是当我想把这个流程的某个部分比如信息检索复用到另一个项目时发现它和前后逻辑耦合得太紧几乎要重写。这时我才真正理解了LCEL的价值。它不是一个可有可无的语法糖而是一套用于构建可靠、可维护、可观测的AI应用链路的工程化框架。它把AI应用开发中那些琐碎但至关重要的工程问题——错误处理、流式输出、并行化、调试、部署——通过声明式的表达方式标准化了。简单说LCEL让你能用写“配置”的方式来获得写“系统”的健壮性。举个例子没有LCEL之前你想实现一个链的流式输出每个token实时返回可能需要深入框架内部处理复杂的异步生成器和回调。而在LCEL里这只是一行.stream()的事。这种抽象对于需要快速迭代和稳定交付的AI工程来说是生产力的巨大飞跃。2. LCEL核心思想将一切视为可组合的“Runnable”要用好LCEL首先要跳出“链Chain”是固定流程的思维定式。在LCEL的世界观里万物皆可为“Runnable”可运行对象。一个LLM模型是Runnable一个提示词模板是Runnable一个解析输出的函数是Runnable甚至一个简单的字符串转换也是Runnable。LCEL的核心操作符极其简单主要就两个管道符|和.invoke()或.stream(),.batch()。管道符|代表“然后”。A | B | C表示数据先经过A处理其结果再传给B最后传给C。这构成了链式调用。调用方法.invoke()用于同步调用.stream()用于流式调用.batch()用于批量处理.with_retry()用于添加重试逻辑等。这种设计的美妙之处在于一致性和可组合性。无论你组合的是多么复杂的模块最终得到的依然是一个Runnable对象。你可以像调用一个函数一样调用整个链路也可以把它作为一个子模块嵌入到另一个更复杂的链路中。这种乐高积木式的体验是工程化复用的基础。让我们从一个最简单的例子开始看看LCEL如何将想法落地from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义两个可运行对象提示词模板和LLM模型 prompt ChatPromptTemplate.from_template(请用一句话介绍{product}。) model ChatOpenAI(modelgpt-4) # 使用管道符组合成一个链 chain prompt | model # 调用链 result chain.invoke({product: LCEL}) print(result.content) # 输出可能为“LCEL是LangChain表达式语言用于声明式地组合AI模块以构建复杂应用。”在这个例子中prompt和model都是Runnable。prompt | model创造了一个新的Runnable即一个链。当我们invoke这个链时内部发生了输入字典{product: LCEL}传给promptprompt渲染成完整的消息列表再传给modelmodel生成AI回复。3. 四类典型链路的工程化写法详解理解了核心思想后我们进入实战。下面我将拆解四种在AI应用中最常见、也最能体现LCEL工程价值的链路模式并给出每种模式下的“玩具写法”与“工程化写法”对比。3.1 模式一基础问答链RAG检索增强生成这是目前最流行的应用模式。核心流程是用户提问 - 检索相关文档 - 组合文档和问题生成提示 - LLM生成答案。玩具写法常见问题# 伪代码示意问题 query 什么是机器学习 docs vectorstore.similarity_search(query) # 检索 context \n.join([doc.page_content for doc in docs]) # 粗暴拼接 prompt f请根据以下上下文回答问题\n{context}\n\n问题{query} response llm.invoke(prompt) # 调用问题错误处理缺失如检索为空、提示词模板硬编码、难以调试中间结果context到底长啥样、无法流式输出。LCEL工程化写法from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate # 1. 定义检索函数也包装成Runnable def retriever(query: str): # 这里可以加入业务逻辑如查询改写、多路召回、过滤等 docs vectorstore.similarity_search(query, k4) return \n\n.join([d.page_content for d in docs]) # 2. 定义提示词模板更清晰、可配置 template 你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。 如果你不知道答案就诚实地回答你不知道不要编造信息。 上下文 {context} 问题{question} 请给出详细、准确的回答 prompt ChatPromptTemplate.from_template(template) # 3. 定义LLM和输出解析器 model ChatOpenAI(modelgpt-4, temperature0) output_parser StrOutputParser() # 4. 使用RunnablePassthrough和管道符组合链路 # RunnablePassthrough用于传递输入或对输入进行简单转换 chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | output_parser ) # 使用 answer chain.invoke(什么是机器学习) print(answer) # 流式输出体验 for chunk in chain.stream(什么是机器学习): print(chunk, end, flushTrue)工程化优势解析声明式组合链路(A | B | C)清晰表达了数据流一目了然。结构化输入{context: retriever, question: ...}这种字典形式明确了每个环节需要的数据字段避免了参数传递的混乱。RunnablePassthrough()表示原封不动传递question字段。易于调试你可以轻松地测试链路的任何一个环节。例如想看看检索到的上下文是什么可以单独运行retriever.invoke(“什么是机器学习”)。内置超能力.stream()直接支持流式输出无需额外代码。.with_retry()可以轻松为整个链或单个组件如model添加重试逻辑应对API的不稳定。可观测性可以方便地接入LangSmith等调试平台追踪每个Runnable的输入输出这对排查生产环境问题至关重要。3.2 模式二分支与路由链动态流程控制很多复杂场景下流程不是线性的需要根据AI的中间输出决定下一步做什么。例如先让AI判断用户意图再路由到不同的处理子链。玩具写法痛点大量if-else嵌套逻辑分散子链复用困难路由逻辑与业务逻辑耦合。LCEL工程化写法LCEL提供了RunnableBranch和RunnableLambda来处理分支逻辑。from langchain_core.runnables import RunnableBranch, RunnableLambda # 定义几个不同的处理子链假设它们都已定义好 # 1. 客服问答链 customer_service_chain ( prompt_customer_service | model | StrOutputParser() ) # 2. 订单查询链 order_query_chain ( prompt_order_query | model | StrOutputParser() ) # 3. 闲聊链 chitchat_chain ( prompt_chitchat | model | StrOutputParser() ) # 4. 默认兜底链 default_chain ( prompt_default | model | StrOutputParser() ) # 定义路由判断函数让AI判断意图 def route_classifier(input_data: dict): user_query input_data[question] # 这里可以用一个简单的LLM调用来分类也可以使用更复杂的逻辑 classifier_prompt ChatPromptTemplate.from_template( 请判断用户问题的意图类别只输出类别名称不要输出其他任何内容。 可选类别customer_service客服咨询, order_query订单查询, chitchat闲聊。 用户问题{query} 意图类别) classifier_chain classifier_prompt | ChatOpenAI(modelgpt-3.5-turbo) | StrOutputParser() category classifier_chain.invoke({query: user_query}) return {category: category, question: user_query} # 使用RunnableBranch构建路由 branch RunnableBranch( (lambda x: customer_service in x[category], customer_service_chain), (lambda x: order_query in x[category], order_query_chain), (lambda x: chitchat in x[category], chitchat_chain), default_chain # 默认分支 ) # 组合完整链路先分类再路由 full_chain RunnableLambda(route_classifier) | branch # 使用 result full_chain.invoke({question: 我的订单12345发货了吗}) print(result)工程化优势解析解耦路由与业务路由逻辑route_classifier和RunnableBranch与各个业务处理链完全分离。每个子链可以独立开发、测试和优化。灵活的条件判断RunnableBranch的条件可以是任何返回布尔值的函数让你能实现基于内容、分数、甚至外部API响应的复杂路由。清晰的拓扑结构整个动态流程依然用一个full_chain对象表示保持了LCEL组合性的优点。你可以像调用简单链一样调用这个复杂路由链。易于扩展新增一个处理分支只需要定义新的子链并在RunnableBranch中添加一个条件元组即可符合开闭原则。3.3 模式三并行与聚合链Map-Reduce当需要处理多个输入或者将一个任务拆分成多个并行子任务再合并结果时例如总结多篇长文档就需要并行处理模式。玩具写法痛点手动管理线程池或异步任务错误处理复杂结果聚合代码冗长。LCEL工程化写法LCEL通过.map()和归约操作简化了并行处理。from langchain_core.runnables import RunnableParallel, RunnableLambda # 场景用户输入一个主题我们需要并行地从多个角度技术、商业、伦理进行分析最后汇总。 # 定义三个并行的分析子链 tech_analysis_chain ( ChatPromptTemplate.from_template(从技术原理角度分析{theme}。) | ChatOpenAI() | StrOutputParser() ) business_analysis_chain ( ChatPromptTemplate.from_template(从商业模式和市场角度分析{theme}。) | ChatOpenAI() | StrOutputParser() ) ethics_analysis_chain ( ChatPromptTemplate.from_template(从伦理和社会影响角度分析{theme}。) | ChatOpenAI() | StrOutputParser() ) # 使用RunnableParallel进行并行执行 parallel_analysis RunnableParallel( techtech_analysis_chain, businessbusiness_analysis_chain, ethicsethics_analysis_chain ) # 定义汇总链将并行结果聚合 def synthesize_results(inputs: dict): tech inputs[tech] business inputs[business] ethics inputs[ethics] summary_prompt ChatPromptTemplate.from_template( 你是一位高级分析师。请将以下关于“{theme}”的三个分析视角整合成一份全面的报告。 技术视角{tech} 商业视角{business} 伦理视角{ethics} 请生成一份结构清晰、逻辑连贯的综合性分析报告) summary_chain summary_prompt | ChatOpenAI(modelgpt-4) | StrOutputParser() return summary_chain.invoke({ theme: inputs[theme], tech: tech, business: business, ethics: ethics }) # 组合完整链路先并行分析再汇总 full_chain ( {theme: RunnablePassthrough()} | parallel_analysis | RunnableLambda(synthesize_results) ) # 使用 report full_chain.invoke(人工智能大模型) print(report)工程化优势解析声明式并行RunnableParallel清晰地表达了哪些任务可以同时执行框架会负责底层的并发调度。结构化输出并行任务的输出被自动组织成一个字典如{“tech”: “…”, “business”: “…”}极大方便了后续的聚合处理。错误隔离理论上一个并行分支的失败不应直接影响其他分支取决于具体实现和配置这比手动编写的线程池更健壮。资源优化LCEL底层可以与LangChain的异步调用良好结合更高效地利用I/O等待时间。3.4 模式四循环与迭代链自主Agent工作流在一些高级的AI智能体Agent场景中需要让LLM根据当前状态反复思考、执行工具、观察结果直到达成目标。这构成了一个循环。玩具写法痛点手写while循环管理状态停止条件判断复杂历史上下文管理混乱难以调试单步。LCEL工程化写法LCEL通过将循环本身也抽象为Runnable提供了更优雅的解决方案。虽然LangChain有专门的AgentExecutor但用LCEL核心原语也能构建类似的循环逻辑理解其本质。from typing import Dict, Any, List from langchain_core.runnables import RunnableConfig # 模拟一个简单的ReActReasoning Acting风格Agent循环 # 状态结构 class AgentState(Dict): question: str # 原始问题 thoughts: List[str] # 思考记录 action: str # 下一步动作 action_input: str # 动作输入 observation: str # 执行动作后的观察结果 final_answer: str # 最终答案 # 1. 思考节点分析当前状态决定下一步思考 or 行动 or 结束 def think_node(state: AgentState) - AgentState: prompt ChatPromptTemplate.from_messages([ (system, 你是一个善于思考和分析的助手。请根据当前状态决定下一步。), (human, 当前问题{question} 历史思考{thoughts} 上一步观察{observation} 请分析如果需要更多信息请决定调用哪个工具输出格式Action: 工具名\nAction Input: 输入。如果已有足够信息回答问题则直接输出最终答案输出格式Final Answer: 答案。如果还需要继续推理就输出你的思考输出格式Thought: 思考内容。 ) ]) chain prompt | ChatOpenAI() | StrOutputParser() response chain.invoke(state) # 解析LLM的响应更新状态 if response.startswith(Final Answer:): state[final_answer] response.replace(Final Answer:, ).strip() state[action] FINISH elif response.startswith(Action:): lines response.split(\n) state[action] lines[0].replace(Action:, ).strip() state[action_input] lines[1].replace(Action Input:, ).strip() if len(lines) 1 else state[thoughts].append(f决定执行动作{state[action]}输入{state[action_input]}) elif response.startswith(Thought:): thought response.replace(Thought:, ).strip() state[thoughts].append(thought) state[action] THINK return state # 2. 执行节点根据action调用工具 def act_node(state: AgentState) - AgentState: if state[action] FINISH: return state # 模拟工具调用 tools { search: lambda q: f这是关于{q}的搜索结果摘要。, calculate: lambda expr: f计算结果为{eval(expr)}, # 注意生产环境勿用eval } tool_name state[action] tool_input state[action_input] if tool_name in tools: state[observation] tools[tool_name](tool_input) else: state[observation] f未知工具{tool_name} state[thoughts].append(f执行结果{state[observation]}) return state # 3. 条件判断是否继续循环 def should_continue(state: AgentState) - bool: # 如果有了最终答案或者思考次数太多则停止 if state.get(final_answer): return False if len(state[thoughts]) 10: # 防止无限循环 return False return True # 4. 使用RunnableLambda和循环逻辑组合简化示意 # 注意LCEL本身不直接提供while循环原语但可以通过组合实现或使用LangChain的更高阶抽象如StateGraph。 # 这里展示一种用函数包装的思路说明如何用LCEL思维构建循环单元。 def single_step_chain(state: AgentState): # 一个步骤先思考再执行如果需要 new_state think_node(state) if new_state[action] ! FINISH and new_state[action] ! THINK: new_state act_node(new_state) return new_state # 初始化状态 initial_state AgentState({ question: 请计算圆的面积已知半径为5。然后搜索一下圆在历史上的文化意义。, thoughts: [], observation: , action: , action_input: , final_answer: }) # 手动模拟循环在实际中可用更结构化的方式 current_state initial_state step 0 while should_continue(current_state) and step 5: print(f\n 步骤 {step1} ) current_state single_step_chain(current_state) step 1 if current_state.get(final_answer): print(f\n最终答案{current_state[final_answer]}) else: print(\n未能在限制步数内得到答案。)工程化优势解析状态封装将Agent运行中的所有信息问题、思考、动作、观察封装在一个状态对象中数据流清晰。节点化将“思考”、“执行”等步骤抽象成独立的函数或Runnable每个节点职责单一易于测试和替换。可观测与控制每个循环步骤的输入输出都是明确的可以方便地记录日志、插入监控点或设置断点调试。与高阶框架集成这个模式是理解LangChain中StateGraph、AgentExecutor等更高级循环抽象的基础。LCEL保证了这些高阶框架中的每个节点本身也是Runnable保持了组合性。4. 进阶工程实践让LCEL链路坚如磐石掌握了基本模式接下来是让链路真正具备生产级可靠性的关键技巧。这些往往是“玩具”与“工程”的分水岭。4.1 全面的错误处理与降级策略线上服务必须考虑各种失败场景LLM API超时、返回格式错误、检索结果为空、第三方工具不可用等。LCEL提供了多层级的错误处理机制组件级重试使用.with_retry()为易失败的组件如LLM调用、外部API添加自动重试。from langchain_core.runnables import RunnableConfig robust_model model.with_retry( stop_after_attempt3, wait_exponential_jitterTrue, retry_if_exception_type(Exception,) # 可根据需要指定异常类型 ) chain prompt | robust_model | parserFallback链当主链失败时自动切换到备用链。这可以用RunnableBranch实现。from langchain_core.runnables import RunnableBranch main_chain prompt | gpt4_model | parser fallback_chain prompt | gpt3_model | parser # 使用更稳定的模型 robust_chain RunnableBranch( (lambda x: True, main_chain), # 总是先尝试主链 ).with_fallbacks([fallback_chain]) # 主链失败则回退 # 或者更精细的控制捕获特定异常才回退 # 需要结合try/except和RunnableLambda实现输入/输出验证使用Pydantic模型在链的入口和出口验证数据格式防止脏数据导致下游崩溃。from pydantic import BaseModel, Field from langchain_core.runnables import RunnableLambda class ChainInput(BaseModel): question: str Field(description用户问题) user_id: str Field(description用户ID) class ChainOutput(BaseModel): answer: str confidence: float def validate_input(raw_input: dict): return ChainInput(**raw_input) def validate_output(raw_output: dict): # 这里可以加入业务逻辑如检查answer是否为空 if not raw_output.get(answer): raw_output[answer] 抱歉我暂时无法回答这个问题。 raw_output[confidence] 0.0 return ChainOutput(**raw_output) validated_chain ( RunnableLambda(validate_input) | core_chain # 你的核心业务链 | RunnableLambda(validate_output) )4.2 性能优化并行、批处理与缓存利用.batch()处理批量请求对于离线处理或异步任务批量调用能显著减少API往返开销。questions [问题1, 问题2, 问题3] # 注意chain本身需要支持批量输入。很多内置组件如ChatOpenAI支持。 answers chain.batch([{question: q} for q in questions])缓存中间结果对于耗时的、结果稳定的步骤如文档检索、复杂计算可以引入缓存。LCEL可以方便地与langchain.cache集成。from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache # 使用内存缓存开发用 set_llm_cache(InMemoryCache()) # 或使用SQLite缓存持久化 set_llm_cache(SQLiteCache(database_path.langchain.db)) # 现在相同的LLM调用会自动从缓存读取并行化独立步骤如模式三所示使用RunnableParallel对无依赖的步骤进行并行化。4.3 可观测性与调试LangSmith集成这是LCEL在工程上最大的优势之一。通过集成LangSmith你可以像拥有一个分布式追踪系统一样可视化链路的每一次调用。import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] your-api-key os.environ[LANGCHAIN_PROJECT] your-project-name # 之后所有chain.invoke()/stream()/batch()的调用详情都会记录在LangSmith上。 # 你可以看到每个Runnable节点的输入、输出、耗时、token消耗甚至LLM的完整提示词。在生产环境中这 invaluable。你可以快速定位是哪个环节慢了是检索慢还是LLM生成慢哪个环节出错了LLM返回了奇怪格式以及用户的真实输入和AI的输出到底是什么。4.4 配置与参数管理LCEL链是高度可配置的。你可以通过RunnableConfig在运行时动态传递参数比如为不同的用户请求设置不同的LLM温度或模型。config RunnableConfig( configurable{ llm_model: gpt-4-0125-preview, # 可配置的模型名 llm_temperature: 0.2, } ) # 在定义链时使用ConfigurableField来声明可配置参数 from langchain_core.runnables import ConfigurableField model ChatOpenAI( modelgpt-3.5-turbo ).configurable_alternatives( ConfigurableField(idllm_model), default_keygpt-3.5-turbo, gpt4ChatOpenAI(modelgpt-4), claudeChatAnthropic(modelclaude-3-haiku) ) # 调用时传入config result chain.invoke({question: ...}, configconfig)这使得你可以用同一条链的定义轻松支撑A/B测试不同模型对比、多租户不同客户不同配置等复杂场景。5. 从开发到部署LCEL链路的生命周期管理一条写好的LCEL链路最终要服务于用户。这就涉及到测试、版本管理和部署。单元测试与集成测试测试单个Runnable因为每个组件都是独立的你可以单独测试你的提示词模板、检索函数、输出解析器。def test_prompt_template(): prompt ChatPromptTemplate.from_template(Hello {name}!) messages prompt.invoke({name: World}) assert messages.to_string() Human: Hello World!测试完整链使用固定的输入断言输出符合预期。可以利用langchain.smith中的评估工具进行更复杂的测试。版本控制与序列化LCEL链可以方便地序列化和反序列化这有利于版本管理和部署。# 保存链 chain.save(my_awesome_chain.json) # 加载链 from langchain_core.runnables import load_runnable loaded_chain load_runnable(my_awesome_chain.json)注意序列化可能不会保存所有状态如网络连接对于包含外部资源如已连接的向量数据库客户端的链需要特殊处理。部署为API服务使用FastAPI、LangServe等框架可以轻松将LCEL链包装成HTTP API。# 使用LangServe推荐 from fastapi import FastAPI from langserve import add_routes app FastAPI() add_routes(app, chain, path/my_chain) # 现在你的链就有了一个标准的POST接口LangServe会自动生成OpenAPI文档并支持异步调用、流式响应等。6. 避坑指南与心得分享在大量使用LCEL后我总结了一些容易踩坑的地方和最佳实践坑1过度嵌套与可读性。LCEL的管道符虽然强大但过长的链如A | B | C | D | E | F会降低可读性。好的实践是将功能相关的步骤组合成有意义的子链并为子链起一个描述性的变量名。例如将prompt | model | parser组合成qa_chain然后在主链中引用它。坑2错误处理粒度。不要只在最外层加一个try-except。应该根据组件的可靠性在不同层级设置错误处理。例如为LLM调用设置重试和降级为整个链设置输入验证和兜底回答。坑3流式输出的中断。在使用.stream()时如果链中有一个环节不是“流式友好”的比如一个耗时的同步函数它会阻塞整个流。确保链中所有组件都支持异步或流式处理或者将阻塞操作放在链的最后。心得1拥抱“配置即代码”。LCEL鼓励你将链的结构和配置分开。将模型名称、温度、检索数量等参数提取到配置对象或环境变量中。这样当你需要从测试环境切换到生产环境或者调整参数时无需修改核心业务逻辑代码。心得2为一切命名。给每个重要的Runnable变量起一个好名字比如intent_classifier_chain,document_retrieval_chain。这不仅能提高代码可读性在LangSmith的追踪视图中这些名字也会显示出来极大方便调试。心得3从简单开始逐步复杂化。不要一开始就试图构建一个完美的、包含所有异常处理和分支的超级链。先用LCEL把核心流程跑通然后像搭积木一样逐步添加错误处理、缓存、日志、监控等非功能性需求。LCEL的可组合性完美支持这种渐进式开发。说到底LCEL是一种思维模式。它要求开发者从“如何写流程控制代码”转向“如何声明数据处理流程”。这种转变初期可能需要适应但一旦掌握你会发现构建健壮、可维护、可观测的AI应用变得前所未有的高效和清晰。它真正在AI应用开发的“想法”与“工业化实现”之间架起了一座坚实的桥梁。