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

1. 从“玩具”到“工程”：为什么我们需要LCEL

如果你最近在折腾LangChain，或者任何基于大语言模型的应用开发，大概率已经听过LCEL（LangChain 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(model="gpt-4") # 使用管道符组合成一个链 chain = prompt | model # 调用链 result = chain.invoke({"product": "LCEL"}) print(result.content) # 输出可能为：“LCEL是LangChain表达式语言，用于声明式地组合AI模块以构建复杂应用。”

在这个例子中，prompt和model都是Runnable。prompt | model创造了一个新的Runnable（即一个链）。当我们invoke这个链时，内部发生了：输入字典{"product": "LCEL"}传给prompt，prompt渲染成完整的消息列表，再传给model，model生成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, k=4) return "\n\n".join([d.page_content for d in docs]) # 2. 定义提示词模板（更清晰、可配置） template = """你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。 如果你不知道答案，就诚实地回答你不知道，不要编造信息。 上下文： {context} 问题：{question} 请给出详细、准确的回答：""" prompt = ChatPromptTemplate.from_template(template) # 3. 定义LLM和输出解析器 model = ChatOpenAI(model="gpt-4", temperature=0) 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="", flush=True)

工程化优势解析：

1. 声明式组合：链路(A | B | C)清晰表达了数据流，一目了然。
2. 结构化输入：{"context": retriever, "question": ...}这种字典形式，明确了每个环节需要的数据字段，避免了参数传递的混乱。RunnablePassthrough()表示原封不动传递question字段。
3. 易于调试：你可以轻松地测试链路的任何一个环节。例如，想看看检索到的上下文是什么，可以单独运行retriever.invoke(“什么是机器学习？”)。
4. 内置超能力：.stream()直接支持流式输出，无需额外代码。.with_retry()可以轻松为整个链或单个组件（如model）添加重试逻辑，应对API的不稳定。
5. 可观测性：可以方便地接入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(model="gpt-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)

工程化优势解析：

1. 解耦路由与业务：路由逻辑（route_classifier和RunnableBranch）与各个业务处理链完全分离。每个子链可以独立开发、测试和优化。
2. 灵活的条件判断：RunnableBranch的条件可以是任何返回布尔值的函数，让你能实现基于内容、分数、甚至外部API响应的复杂路由。
3. 清晰的拓扑结构：整个动态流程依然用一个full_chain对象表示，保持了LCEL组合性的优点。你可以像调用简单链一样调用这个复杂路由链。
4. 易于扩展：新增一个处理分支，只需要定义新的子链，并在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( tech=tech_analysis_chain, business=business_analysis_chain, ethics=ethics_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(model="gpt-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)

工程化优势解析：

1. 声明式并行：RunnableParallel清晰地表达了哪些任务可以同时执行，框架会负责底层的并发调度。
2. 结构化输出：并行任务的输出被自动组织成一个字典（如{“tech”: “…”, “business”: “…”}），极大方便了后续的聚合处理。
3. 错误隔离：理论上，一个并行分支的失败不应直接影响其他分支（取决于具体实现和配置），这比手动编写的线程池更健壮。
4. 资源优化：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 # 模拟一个简单的ReAct（Reasoning + 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=== 步骤 {step+1} ===") 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未能在限制步数内得到答案。")

工程化优势解析：

1. 状态封装：将Agent运行中的所有信息（问题、思考、动作、观察）封装在一个状态对象中，数据流清晰。
2. 节点化：将“思考”、“执行”等步骤抽象成独立的函数或Runnable，每个节点职责单一，易于测试和替换。
3. 可观测与控制：每个循环步骤的输入输出都是明确的，可以方便地记录日志、插入监控点或设置断点调试。
4. 与高阶框架集成：这个模式是理解LangChain中StateGraph、AgentExecutor等更高级循环抽象的基础。LCEL保证了这些高阶框架中的每个节点本身也是Runnable，保持了组合性。

4. 进阶工程实践：让LCEL链路坚如磐石

掌握了基本模式，接下来是让链路真正具备生产级可靠性的关键技巧。这些往往是“玩具”与“工程”的分水岭。

4.1 全面的错误处理与降级策略

线上服务必须考虑各种失败场景：LLM API超时、返回格式错误、检索结果为空、第三方工具不可用等。

LCEL提供了多层级的错误处理机制：

1. 组件级重试：使用.with_retry()为易失败的组件（如LLM调用、外部API）添加自动重试。

   from langchain_core.runnables import RunnableConfig robust_model = model.with_retry( stop_after_attempt=3, wait_exponential_jitter=True, retry_if_exception_type=(Exception,) # 可根据需要指定异常类型 ) chain = prompt | robust_model | parser

2. Fallback链：当主链失败时，自动切换到备用链。这可以用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实现

3. 输入/输出验证：使用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 性能优化：并行、批处理与缓存

1. 利用.batch()处理批量请求：对于离线处理或异步任务，批量调用能显著减少API往返开销。

   questions = ["问题1", "问题2", "问题3"] # 注意：chain本身需要支持批量输入。很多内置组件（如ChatOpenAI）支持。 answers = chain.batch([{"question": q} for q in questions])

2. 缓存中间结果：对于耗时的、结果稳定的步骤（如文档检索、复杂计算），可以引入缓存。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调用会自动从缓存读取

3. 并行化独立步骤：如模式三所示，使用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( model="gpt-3.5-turbo" ).configurable_alternatives( ConfigurableField(id="llm_model"), default_key="gpt-3.5-turbo", gpt4=ChatOpenAI(model="gpt-4"), claude=ChatAnthropic(model="claude-3-haiku") ) # 调用时传入config result = chain.invoke({"question": "..."}, config=config)

这使得你可以用同一条链的定义，轻松支撑A/B测试（不同模型对比）、多租户（不同客户不同配置）等复杂场景。

5. 从开发到部署：LCEL链路的生命周期管理

一条写好的LCEL链路，最终要服务于用户。这就涉及到测试、版本管理和部署。

1. 单元测试与集成测试：

   • 测试单个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中的评估工具进行更复杂的测试。

2. 版本控制与序列化：LCEL链可以方便地序列化和反序列化，这有利于版本管理和部署。

   # 保存链 chain.save("my_awesome_chain.json") # 加载链 from langchain_core.runnables import load_runnable loaded_chain = load_runnable("my_awesome_chain.json")

   注意：序列化可能不会保存所有状态（如网络连接），对于包含外部资源（如已连接的向量数据库客户端）的链，需要特殊处理。

3. 部署为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应用开发的“想法”与“工业化实现”之间，架起了一座坚实的桥梁。