LCEL工程化实战：构建可观测、流式与高可用的AI应用链路

1. 项目概述为什么我们需要关注LCEL的工程化写法如果你最近在折腾LangChain或者任何基于大语言模型的应用开发大概率已经听过LCELLangChain Expression Language这个名字。它被官方大力推崇号称是构建AI链路的“声明式”标准。但说实话刚上手时很多开发者包括我自己都挺懵的这玩意儿看起来不就是把一堆组件用管道符|连起来吗跟我自己写函数调用有什么区别为什么非得用这个这就是问题的关键。LCEL的核心价值远不止于一个“更酷”的语法糖。它解决的是AI应用工程化中的一系列痛点链路可观测性、流式输出、并行化、错误处理与重试、以及链路本身的复用与组合。当你自己用Python函数硬编码一个RAG流程时你很难优雅地给每一步加上日志追踪也很难让LLM生成第一个词时就流式返回给前端更别提方便地做A/B测试或者动态替换某个环节了。所以“LCEL到底该怎么用”这个问题背后真正问的是如何将我们脑中那个模糊的AI业务流程比如“先检索再总结最后润色”转化为一个健壮、可维护、可观测的生产级代码模块本文不会重复官方文档的基础语法而是聚焦于四种最常见、也最棘手的AI链路模式拆解它们的工程化实现写法。我会结合真实的踩坑经验告诉你哪些写法是“花架子”哪些才是能扛住线上流量的“真功夫”。2. 核心设计思路从“脚本思维”到“链路思维”的转变在深入具体写法之前我们必须先统一思想。用LCEL写链路和写传统脚本是两种截然不同的思维方式。2.1 脚本思维的典型困境假设我们要实现一个简单的“查询理解向量检索”流程。用脚本思维你可能会这么写def naive_rag_query(user_query: str) - str: # 1. 查询改写 rewritten_query llm.invoke(f改写这个查询以更好地进行检索{user_query}) # 2. 向量检索 retriever vectorstore.as_retriever() docs retriever.invoke(rewritten_query) # 3. 组合上下文并生成答案 context \n\n.join([doc.page_content for doc in docs]) final_prompt f基于以下上下文\n{context}\n\n请回答{user_query} answer llm.invoke(final_prompt) return answer这段代码能跑通但它非常脆弱没有观测点如果最终答案错了你很难定位是改写出了问题还是检索到了垃圾文档或者是LLM自己胡编乱造。你只能靠打印日志但日志散落各处线上排查宛如大海捞针。阻塞式调用llm.invoke是同步阻塞的用户必须等待所有步骤完成包括LLM生成全部文本才能看到结果体验很差。错误处理缺失如果向量数据库临时超时整个流程就崩溃了没有重试或降级策略。难以复用和测试这个函数是个黑盒你想单独测试“查询改写”这个环节或者把它用到另一个需要改写的链路里都很麻烦。2.2 链路思维与LCEL的赋能LCEL促使我们将这个流程视为一个由可观测节点和标准化边组成的有向图。每个节点如prompt | llm | retriever都是一个独立的、具备标准输入输出接口的组件。LCEL的管道操作符|负责将这些组件连接起来并自动注入了以下能力流式处理Streaming任何实现了stream方法的组件其输出都可以被流式消费。LCEL链路天然支持从第一个可流式组件到最后输出的逐词流式。异步支持Async只需调用ainvoke或astream整个链路自动以异步方式运行轻松集成到FastAPI等异步框架中。日志与追踪Logging Tracing通过with_config或回调函数可以给整个链路或单个节点添加统一的日志和追踪如LangSmith每个步骤的输入输出、耗时一目了然。并行执行Parallel使用RunnableParallel可以轻松实现分支与合并让不依赖的步骤同时跑。错误处理与重试Fallbacks使用RunnableWithFallbacks可以为链路或节点设置备用方案。因此工程化写法的目标就是最大化地利用LCEL提供的这些基础设施而不是仅仅用它来串联函数。下面我们就进入四种典型链路的实战。3. 第一类典型链路基础检索增强生成RAG链路的稳健写法这是LCEL最常见的应用场景。一个基础的RAG包含检索器Retriever和生成链。但工程上我们需要考虑更多。3.1 基础版本与它的缺陷我们先看一个直观但脆弱的写法from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough template 基于以下上下文回答問題 {context} 問題{question} prompt ChatPromptTemplate.from_template(template) basic_rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )这个链很简洁但它假设检索总是成功的且检索到的文档就是相关的。在生产中这远远不够。3.2 工程化增强引入路由、重排与上下文过滤一个健壮的RAG链路应该能处理“检索不到相关内容”的情况并对检索结果进行精炼。1. 检索结果重排Re-ranking直接使用向量相似度检索到的前k个文档质量可能参差不齐。加入一个轻量级的重排模型如Cohere Rerank或使用LLM进行重排可以显著提升效果。from langchain_core.runnables import RunnableLambda def rerank_docs(query: str, docs: list): # 这里简化实现实际可使用专门的rerank模型 # 模拟一个根据与query相关性打分的逻辑 scored_docs [] for doc in docs: # 此处应有更复杂的评分逻辑例如调用一个评分模型 score some_scoring_function(query, doc.page_content) scored_docs.append((score, doc)) scored_docs.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in scored_docs[:3]] # 返回Top 3 # 将重排逻辑封装为Runnable reranker RunnableLambda(lambda x: rerank_docs(x[question], x[context])) enhanced_rag_chain ( { context: retriever | reranker, # 检索后立即重排 question: RunnablePassthrough() } | prompt | llm | StrOutputParser() )2. 上下文压缩与过滤即使文档相关也可能包含大量无关信息导致LLM注意力分散或超出上下文窗口。可以使用ContextualCompressionRetriever或自定义过滤器。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 使用LLM提取文档中与问题相关的片段 compressor LLMChainExtractor.from_llm(llm) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverretriever ) # 现在retriever返回的已经是压缩后的相关片段了 compressed_rag_chain ( {context: compression_retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )3. 空结果处理与路由当检索器返回空列表或相关性极低时不应该强行让LLM基于“空上下文”胡编乱造。更好的做法是让链路“路由”到一个不同的处理分支比如直接让LLM回答或者告知用户“未找到相关信息”。from langchain_core.runnables import RunnableBranch def route_based_on_context(info): 根据检索到的上下文决定走哪条分支 context_docs info[context] question info[question] if not context_docs: # 分支1无上下文让LLM基于自身知识回答并提示限制 return no_context_branch elif len(context_docs) 2: # 分支2上下文很少可以走一个简化的QA流程 return weak_context_branch else: # 分支3上下文充足走标准RAG流程 return standard_rag_branch # 定义不同分支 no_context_prompt ChatPromptTemplate.from_template( 你是一个有帮助的助手。请注意我未能找到相关的参考信息。请基于你的通用知识回答以下问题如果无法确定请明确说明。\n问题{question} ) weak_context_prompt ChatPromptTemplate.from_template( 参考信息有限{context}\n请基于以上有限信息回答{question}如果信息不足请说明。 ) branch RunnableBranch( (lambda x: route_based_on_context(x) no_context_branch, no_context_prompt | llm | StrOutputParser()), (lambda x: route_based_on_context(x) weak_context_branch, weak_context_prompt | llm | StrOutputParser()), standard_rag_prompt | llm | StrOutputParser() # 默认分支 ) # 组装最终链路 routed_rag_chain ( {context: retriever, question: RunnablePassthrough()} | branch )实操心得重排和压缩会显著增加链路延迟需要权衡。对于延迟敏感的场景可以先在召回阶段使用更精准的检索器如HyDE生成的查询进行检索或者只对高置信度的模糊查询启用重排。路由逻辑的判断函数要尽量轻量避免引入另一个LLM调用导致延迟翻倍。4. 第二类典型链路具备复杂分支与聚合的决策链很多AI应用不是简单的直线流程而是包含条件判断、并行执行和结果聚合的复杂工作流。例如一个分析用户需求并调用不同工具的Agent或者一个需要多角度验证答案的流程。4.1 使用RunnableBranch实现条件路由上面的RAG路由是一个简单例子。更复杂的路由可能基于LLM的决策。from langchain_core.runnables import RunnableBranch, RunnableLambda classifier_prompt ChatPromptTemplate.from_template( 用户的问题是{question} 请判断这个问题属于以下哪一类 A. 需要查询最新信息如新闻、股价、天气。 B. 需要创作或改写文本如写邮件、写诗。 C. 需要基于特定知识库回答如公司内部文档。 D. 其他通用问题。 请只输出字母编号。 ) def parse_classification(output: str) - str: output output.strip().upper() if output in [A, B, C, D]: return output else: return D # 解析失败归为其他类 # 分类链 classification_chain classifier_prompt | llm | StrOutputParser() | RunnableLambda(parse_classification) # 定义各分支的处理链这里用模拟链代替 news_chain RunnableLambda(lambda x: f[新闻查询链] 处理问题{x}) # 模拟 write_chain RunnableLambda(lambda x: f[文本创作链] 处理问题{x}) rag_chain enhanced_rag_chain # 复用之前的RAG链 general_chain RunnableLambda(lambda x: f[通用问答链] 处理问题{x}) # 构建分支注意分支的输入是原始问题但路由判断需要分类结果 # 我们需要一个组合链先分类再根据分类结果路由到不同的处理链同时把原始问题传递过去。 def route_input(data): # data 包含原始问题 ‘question’ 和分类结果 ‘category’ category data[category] question data[question] return {question: question, category: category} main_branch RunnableBranch( (lambda x: x[category] A, news_chain), (lambda x: x[category] B, write_chain), (lambda x: x[category] C, rag_chain), general_chain # 默认分支 ) # 最终链并行获取分类和传递问题 - 路由 - 执行对应链 complex_agent_chain ( RunnableParallel({ category: classification_chain, question: RunnablePassthrough() }) | RunnableLambda(route_input) | main_branch ) # 调用 result complex_agent_chain.invoke(今天北京的天气怎么样) print(result) # 可能输出[新闻查询链] 处理问题今天北京的天气怎么样4.2 使用RunnableParallel实现并行与聚合当链路中有多个独立的任务时并行执行可以大幅降低总延迟。RunnableParallel是关键。from langchain_core.runnables import RunnableParallel def search_web(query: str): # 模拟网络搜索 return f网络搜索结果关于{query}的信息... def search_internal_db(query: str): # 模拟内部数据库检索 return f内部知识库结果相关记录... def ask_llm_directly(query: str): # 模拟直接问LLM return fLLM直接回答对于{query}我认为... # 构建并行执行链路 parallel_search_chain RunnableParallel({ web_result: RunnableLambda(lambda x: search_web(x[question])), internal_result: RunnableLambda(lambda x: search_internal_db(x[question])), llm_opinion: RunnableLambda(lambda x: ask_llm_directly(x[question])) }) # 聚合并行结果并交给一个总结LLM synthesis_prompt ChatPromptTemplate.from_template( 你是一个信息分析专家。用户的问题是{question} 你已经从不同渠道获得了以下信息 - 网络搜索{web_result} - 内部资料{internal_result} - 大模型直接生成{llm_opinion} 请综合以上所有信息整理成一份全面、准确、无矛盾的答案。 ) synthesis_chain synthesis_prompt | llm | StrOutputParser() # 组装完整链路并行搜索 - 综合生成 parallel_rag_chain ( {question: RunnablePassthrough()} | parallel_search_chain | synthesis_chain ) # 调用 result parallel_rag_chain.invoke(什么是量子计算)注意事项并行执行虽然快但要注意资源消耗。如果并行的每个分支都调用LLM或重型API可能会迅速耗尽预算或触发速率限制。务必为这些外部调用设置超时和重试机制。RunnableParallel接收一个字典字典的值必须是Runnable。它会把相同的输入这里是包含question的字典广播给所有分支。5. 第三类典型链路实现流式输出与中间结果访问流式输出对于用户体验至关重要。LCEL让流式变得非常简单但如何流式传输复杂链路的中间结果例如先流式返回检索到的文档标题再流式返回LLM生成的答案则需要一些技巧。5.1 基础流式与astream_events最简单的流式就是直接在链上调用.stream()或.astream()。# 对于之前的 basic_rag_chain async for chunk in basic_rag_chain.astream(LangChain是什么): print(chunk, end, flushTrue) # 会逐词或逐token输出LLM生成的答案但这样你只能看到最终的输出。如果你想在流式过程中也拿到检索到的文档中间步骤的结果就需要使用更强大的astream_eventsAPI。这是LCEL工程化中最具价值的特性之一。from langchain_core.runnables import RunnableConfig async def stream_with_retrieved_docs(chain, question): config RunnableConfig() async for event in chain.astream_events(question, versionv1, configconfig): kind event[event] name event.get(name) if kind on_chain_start and name Retriever: print(f\n[开始检索]...) elif kind on_chain_end and name Retriever: # 检索结束可以拿到输出文档列表 docs event[data][output] print(f[检索完成共找到 {len(docs)} 篇文档]) for i, doc in enumerate(docs): print(f Doc {i1}: {doc.metadata.get(title, No Title)[:50]}...) elif kind on_llm_stream and name ChatModel: # LLM正在流式生成 chunk event[data][chunk] if hasattr(chunk, content): print(chunk.content, end, flushTrue)astream_events提供了链路执行的完整生命周期事件让你可以像在链路中插入探针一样实时获取每一步的状态、输入和输出。这对于构建实时进度提示、调试复杂链路、或在UI中展示分阶段结果如“正在检索...”“正在生成...”至关重要。5.2 自定义组件的流式支持如果你自己写了一个RunnableLambda来处理数据默认它是不支持流式的因为它只是一个函数调用。为了让你的自定义步骤也能融入流式链路你需要让它返回一个可迭代对象或者实现stream方法。from langchain_core.runnables import RunnableLambda from typing import AsyncIterator def slow_processor(input_text: str) - str: # 模拟一个耗时的处理 import time time.sleep(1) return f处理结果: {input_text.upper()} # 普通写法会阻塞流式 blocking_step RunnableLambda(slow_processor) # 支持流式的写法通过生成器模拟“逐步输出” def stream_slow_processor(input_text: str) - AsyncIterator[str]: # 模拟分阶段处理 stages [分析..., 计算..., 整合...] for stage in stages: yield f[处理器] {stage}\n # 流式输出中间状态 import asyncio asyncio.sleep(0.3) # 模拟耗时 yield f[处理器] 最终结果: {input_text.upper()} streaming_step RunnableLambda(lambda x: stream_slow_processor(x[question])) # 需要将这个步骤集成到链中并确保其输出能被后续步骤正确处理。实操心得在生产环境中使用astream_events时要特别注意事件数据的体积和序列化成本。对于高频链路记录所有事件可能会对性能产生影响。通常我们只订阅关键步骤的事件如on_chain_endofRetriever,on_llm_stream。另外前端需要与后端的事件流通常是Server-Sent Events正确对接以平滑地展示流式内容和中间状态更新。6. 第四类典型链路集成外部工具与API的Agent式调用AI应用经常需要与外部世界交互比如查询数据库、调用API、执行代码。LCEL可以很好地组织这种“规划-执行”的Agent工作流。6.1 使用Tool节点构建可执行链路LangChain提供了tool装饰器来轻松创建工具。LCEL链路可以动态地决定是否以及如何调用这些工具。from langchain_core.tools import tool from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate # 1. 定义工具 tool def get_weather(city: str) - str: 获取指定城市的当前天气。 # 这里应调用真实的天气API例如OpenWeatherMap return f{city}的天气是晴朗25摄氏度。 tool def search_company_info(company_name: str) - str: 搜索公司的公开信息。 # 模拟搜索 return f{company_name}是一家专注于人工智能的科技公司。 # 2. 创建Agent其核心是一个LCEL链 tools [get_weather, search_company_info] prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以调用工具来回答问题。), (human, {input}), (placeholder, {agent_scratchpad}), # 用于记录Agent的思考过程 ]) # 注意create_tool_calling_agent 返回的是一个Runnable agent_runnable create_tool_calling_agent(llm, tools, prompt) # 3. 创建执行器它也是一个Runnable负责循环执行Agent直到结束 agent_executor AgentExecutor(agentagent_runnable, toolstools, verboseTrue) # 调用 result agent_executor.invoke({input: 北京和上海的天气怎么样然后介绍一下OpenAI这家公司。}) print(result[output])AgentExecutor本身就是一个复杂的LCEL链它内部包含了循环调用LLM决定是否使用工具- 执行工具 - 将结果返回给LLM - 继续循环直到LLM决定给出最终答案。6.2 将固定工具调用模式封装为链并非所有工具调用都需要Agent的动态规划。对于模式固定的场景我们可以用LCEL构建一个更高效、更可控的“工具使用链”。from langchain_core.runnables import RunnableBranch, RunnableParallel # 假设我们有一个工具选择器可以是规则也可以是小模型 def tool_selector(query: str) - str: if 天气 in query: return weather elif 公司 in query or 介绍 in query: return company_info else: return general # 定义处理链 weather_chain RunnableLambda(lambda x: get_weather.invoke(x[query].replace(天气, ).strip())) company_chain RunnableLambda(lambda x: search_company_info.invoke(x[query])) general_chain RunnableLambda(lambda x: llm.invoke(f回答{x[query]})) # 构建路由链 tool_router_chain ( RunnableParallel({ tool_choice: RunnableLambda(lambda x: tool_selector(x[query])), query: RunnablePassthrough() }) | RunnableBranch( (lambda x: x[tool_choice] weather, weather_chain), (lambda x: x[tool_choice] company_info, company_chain), general_chain ) ) # 调用 result tool_router_chain.invoke({query: 今天杭州天气如何}) print(result)这种方式比通用Agent更轻量、更快且确定性更高适合工具种类少、调用模式清晰的场景。注意事项使用工具时最关键的是错误处理。网络调用可能超时API可能返回错误格式。务必在每个工具调用外包裹健壮的错误处理逻辑并考虑设置重试和回退方案。LCEL的RunnableWithFallbacks可以用于工具链级别但更精细的控制往往需要在工具函数内部实现。7. 工程化实践为LCEL链路注入可观测性与稳定性链路写好了怎么确保它在生产环境稳定运行这就需要注入运维能力。7.1 配置管理与环境隔离不要将API密钥、模型名称等配置硬编码在链路定义里。使用with_config或RunnableConfig进行注入。from langchain_core.runnables import ConfigurableField, RunnableConfig # 定义一个可配置的模型 configurable_llm llm.configurable_alternatives( ConfigurableField(idmodel_name, name模型名称), default_keygpt-4, gpt3another_llm, # 可以指向另一个配置好的LLM实例 claudeclaude_llm, ) # 在链中使用可配置模型 configurable_chain prompt | configurable_llm | StrOutputParser() # 调用时传入配置 config RunnableConfig(configurable{model_name: claude}) result configurable_chain.invoke({question: 你好}, configconfig)这样你可以通过外部配置如环境变量、数据库动态切换链路的组件便于进行A/B测试或故障转移。7.2 集成LangSmith进行追踪与评估LangSmith是LangChain官方的可观测性平台。集成后每一次链路调用都会被自动记录你可以看到每个步骤的输入输出、耗时和成本。import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_ENDPOINT] https://api.smith.langchain.com os.environ[LANGCHAIN_API_KEY] your-api-key os.environ[LANGCHAIN_PROJECT] my-production-project # 设置项目名 # 现在所有LCEL链的调用都会被自动记录到LangSmith result enhanced_rag_chain.invoke(什么是机器学习)在LangSmith界面上你可以清晰地看到Retriever、Prompt、LLM等节点的详细信息快速定位是检索不准还是LLM生成了错误答案。你还可以基于这些追踪数据创建测试数据集和自动化评估。7.3 实现重试与回退机制网络和外部服务是不可靠的。使用RunnableWithFallbacks为关键节点尤其是LLM调用添加回退链。from langchain_core.runnables import RunnableWithFallbacks primary_llm ChatOpenAI(modelgpt-4, temperature0) fallback_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 为主LLM配置回退 robust_llm RunnableWithFallbacks( runnableprimary_llm, fallbacks[fallback_llm], exceptions_to_handle(Exception,), # 捕获所有异常可根据需要细化 ) # 在链中使用 robust_chain prompt | robust_llm | StrOutputParser()你还可以为整个链设置重试from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def invoke_chain_with_retry(chain, input_data): return chain.invoke(input_data)7.4 性能优化与超时控制对于包含多个步骤的链尤其是并行步骤需要设置全局和局部的超时。import asyncio from langchain_core.runnables import RunnableConfig # 为整个链的invoke设置超时异步场景 try: result await asyncio.wait_for( complex_agent_chain.ainvoke({question: 复杂问题}), timeout30.0 # 整体超时30秒 ) except asyncio.TimeoutError: result 请求超时请稍后再试。 # 更细粒度地可以为某个Runnable绑定自定义的with_timeout方法需要封装对于检索器等可能返回大量文档的组件要合理设置search_kwargs中的k值返回文档数避免上下文过长。8. 常见问题排查与调试技巧实录即使按照最佳实践构建了链路在实际运行中还是会遇到各种问题。这里记录几个我踩过的坑和解决方法。问题1链能跑通但输出结果不符合预期如何调试排查步骤隔离测试将链拆开单独测试每个组件。比如单独用retriever.invoke(“你的问题”)看返回的文档是否相关。单独用prompt.format(context..., question...)看拼装好的提示词是否符合预期。启用详细日志设置os.environ[LANGCHAIN_VERBOSE] true会在控制台打印链的详细执行步骤和中间结果。这是最直接的调试方式。使用LangSmith这是终极武器。在LangSmith的Trace详情里你可以点击进入任何一个节点的输入输出精确看到传递给LLM的提示词到底是什么样子以及LLM的原始响应。检查输出解析器如果输出格式混乱问题可能出在OutputParser上。确保你的提示词要求了明确的格式如JSONMarkdown并且输出解析器能处理LLM可能出现的各种“变体”。问题2流式输出不工作或者只流式了最后一部分可能原因与解决组件不支持流式确保链中的每个组件都支持流式。自定义的RunnableLambda默认不支持。如果某个中间步骤是同步函数且耗时很长它会阻塞整个流。使用了错误的调用方法必须使用.stream()或.astream()方法而不是.invoke()。聚合操作吞掉了流如果你在链的最后使用了RunnableLambda对结果进行复杂的字符串处理这个处理过程本身可能不是流式的。尝试将处理逻辑移到链外或者确保你的处理函数也是一个生成器。前端处理问题后端在流式输出但前端没有正确解析text/event-stream或application/x-ndjson格式。确保后端使用了像StreamingResponse(FastAPI) 这样的响应类。问题3链在并行步骤中报错错误信息难以定位解决策略简化并行先将RunnableParallel替换为串行执行定位是哪个分支出的问题。包装异常在每个并行分支的RunnableLambda内部添加详细的异常捕获和日志。def safe_search(query): try: return some_api_call(query) except Exception as e: logger.error(f搜索API调用失败: {e}, query: {query}) return f搜索暂时不可用: {str(e)[:50]}设置超时为每个可能长时间挂起的并行分支设置独立的超时控制例如使用asyncio.wait_for包装。问题4部署后链路性能下降延迟变高优化方向缓存对检索器使用缓存如LangChain的CacheRetriever对LLM的相同提示词使用缓存很多LLM SDK支持。减少不必要的步骤审视链路是否有步骤可以合并或移除例如如果重排模型对简单查询提升不大可以设计一个轻量级分类器来跳过重排。并行化检查是否有可以改为并行的串行步骤。模型降级在非关键路径或对质量要求不高的环节使用更小、更快的模型如GPT-3.5-Turbo vs GPT-4。监控与 profiling使用LangSmith监控每个节点的耗时找出瓶颈。可能是某个外部API慢也可能是向量检索的k值设得太大。问题5如何对LCEL链进行单元测试和集成测试测试方法模拟Mock外部依赖使用unittest.mock来模拟LLM、Retriever、Tool的返回确保你的链路逻辑正确。可以测试不同分支如检索到结果 vs 未检索到结果是否都能正确路由。测试配置使用with_config在测试时注入模拟组件。快照测试对于复杂的提示词生成可以将其格式化后的字符串与一个“黄金标准”快照进行比较确保提示词模板的修改不会引入意外变化。集成测试在一个小型测试数据库中运行完整的RAG链验证端到端的功能。构建LCEL链路是一个迭代过程。从最简单的管道开始逐步添加错误处理、可观测性、性能优化。始终记住LCEL提供的是一种声明式的、可组合的抽象它的威力在于将复杂的AI应用逻辑变得像搭积木一样清晰和可维护。当你熟悉了这些模式后你会发现构建一个健壮的AI应用不再是一件令人头疼的工程难题。