通过调优一个示例 LangChain 链,实践优化概念。这包括找出性能问题,应用特定方法,并衡量其效果。场景:一个多步骤文档分析链设想一个旨在根据技术报告集合回答问题的链。此过程包含:检索: 使用向量存储查找相关文档片段。初步回答生成: 使用大型语言模型(LLM)仅基于检索到的片段生成回答。回答优化: 使用第二个、可能更强大的大型语言模型(LLM)调用来优化初步回答,提高连贯性,并根据原始问题和初步回答添加上下文。这种多步骤过程很常见,但由于多次大型语言模型(LLM)交互和数据检索操作,可能会引入延迟并增加成本。假设我们最初的链实现如下所示:# 假设 retriever、llm_initial、llm_refine 已预先配置 # retriever: 一个向量存储检索器 # llm_initial: 一个中等大小的LLM,用于快速生成回答 # llm_refine: 一个更大、能力更强的LLM,用于优化 from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.output_parsers import StrOutputParser # 简化的RAG设置 retrieve_docs = RunnablePassthrough.assign( context=(lambda x: x["question"]) | retriever ) # 初步回答提示和链 initial_prompt_template = ChatPromptTemplate.from_template( "Based on this context:\n{context}\n\nAnswer the question: {question}" ) initial_answer_chain = initial_prompt_template | llm_initial | StrOutputParser() # 优化提示和链 refine_prompt_template = ChatPromptTemplate.from_template( "Refine this initial answer: '{initial_answer}' based on the original question: '{question}'. Ensure coherence and accuracy." ) refine_chain = refine_prompt_template | llm_refine | StrOutputParser() # 结合所有步骤的完整链 full_chain = retrieve_docs | RunnablePassthrough.assign( initial_answer=initial_answer_chain ) | RunnablePassthrough.assign( final_answer = (lambda x: {"initial_answer": x["initial_answer"], "question": x["question"]}) | refine_chain ) # 调用示例 # result = full_chain.invoke({"question": "系统X的扩展限制是什么?"}) # print(result["final_answer"])步骤1:基线性能测量在优化之前,我们需要一个基线。我们可以使用简单计时或与LangSmith等追踪工具集成(第五章有介绍)。为简单起见,我们使用基本计时。我们用一个示例问题多次运行该链,并计算结果的平均值。import time import statistics question = "Summarize the main findings regarding performance degradation under load." num_runs = 5 latencies = [] # 假设令牌跟踪是单独实现或通过LangSmith实现的 for _ in range(num_runs): start_time = time.time() # result = full_chain.invoke({"question": question}) # 执行链 # 模拟演示执行时间 time.sleep(12 + (random.random() * 6 - 3)) # 模拟 9-15 秒延迟 end_time = time.time() latencies.append(end_time - start_time) average_latency = statistics.mean(latencies) print(f"Average latency (baseline): {average_latency:.2f} seconds") # 假设通过日志/LangSmith观察到的基线令牌数量:每次查询约2100个令牌假设我们的基线测量结果如下:平均延迟: 13.5 秒平均令牌使用量: 2100 个令牌(估计)使用LangSmith或详细日志,我们可能找到如下细分:检索:1.5 秒初步回答LLM(llm_initial):4.0 秒(800 个令牌)优化LLM(llm_refine):8.0 秒(1300 个令牌)优化步骤(llm_refine)在延迟和令牌消耗方面都是最主要的瓶颈。步骤2:应用优化方法我们来应用前面提到的一些方法。方法1:缓存LLM响应相同的问题或中间处理步骤可能会频繁出现。缓存LLM响应可以显著减少重复请求的延迟和成本。我们添加一个内存缓存。对于生产环境,你通常会使用更持久的缓存,例如Redis、SQL或专门的向量缓存。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache # 设置一个简单的内存缓存 set_llm_cache(InMemoryCache()) # 如果LLM是全局配置的,则链定义本身不需要更改 # 或者,在初始化LLM时直接应用缓存: # llm_initial = ChatOpenAI(..., cache=InMemoryCache()) # llm_refine = ChatOpenAI(..., cache=InMemoryCache()) # 重新运行计时测试,确保多次运行*相同*的问题 # 第一次运行会很慢,后续相同的运行应该快得多。添加缓存并再次运行相同查询后:首次运行延迟: 约13.5 秒(缓存未命中)后续运行延迟: 约1.6 秒(两个LLM都缓存命中,主要时间用于检索)令牌使用量(后续运行): 0 个令牌(从缓存提供)缓存对于重复输入非常有效,但对新颖查询没有帮助。方法2:优化优化步骤优化LLM调用是新颖查询的主要瓶颈。提示工程: 我们能否使优化提示更简洁?也许最初的提示可以要求一个更结构化的输出,从而减少优化需求。假设我们对refine_prompt_template进行优化,使其略短,平均每次调用可节省约50个令牌。模型选择: 功能强大的llm_refine是绝对必需的吗?一个稍小、速度更快的模型能否达到可接受的质量?我们假设将llm_refine切换到一个已知在类似任务中平均快约30%、令牌使用量少约30%的模型,也许会接受轻微的质量权衡。条件执行: 也许优化并非总是必需。我们可以在优化之前添加一个步骤,使用更简单的模型或基于规则的检查来判断初步回答是否足够好。如果是,则完全跳过优化调用。我们来模拟将llm_refine切换到更快模型并稍微优化提示的效果。# 假设 llm_refine_faster 已配置(一个更快、能力稍弱的模型) # 假设 refine_prompt_template_optimized 略短 # 更新 refine_chain 部分 refine_chain_optimized = refine_prompt_template_optimized | llm_refine_faster | StrOutputParser() # 更新完整链定义以使用优化后的优化链 full_chain_optimized = retrieve_docs | RunnablePassthrough.assign( initial_answer=initial_answer_chain ) | RunnablePassthrough.assign( final_answer = (lambda x: {"initial_answer": x["initial_answer"], "question": x["question"]}) | refine_chain_optimized ) # 为新颖查询重新运行计时测试(缓存在此处初期不会有帮助) # ... 计时代码 ...步骤3:重新评估性能我们测量full_chain_optimized对于新颖查询(缓存未命中情况)的性能:# 模拟优化后的演示执行时间 # 检索:1.5秒(无变化) # 初步LLM:4.0秒(无变化,800个令牌) # 优化LLM(更快模型+提示):8.0秒 * 0.7 ≈ 5.6秒(1300个令牌 * 0.7 - 50 ≈ 860个令牌) # 总延迟 ≈ 1.5 + 4.0 + 5.6 = 11.1秒 # 总令牌数 ≈ 800 + 860 = 1660个令牌 # --- 模拟和测量Python代码 --- latencies_optimized = [] for _ in range(num_runs): start_time = time.time() # result = full_chain_optimized.invoke({"question": question}) # 执行优化后的链 # 模拟优化后的执行时间 time.sleep(10 + (random.random() * 3 - 1.5)) # 模拟 8.5 - 11.5 秒延迟 end_time = time.time() latencies_optimized.append(end_time - start_time) average_latency_optimized = statistics.mean(latencies_optimized) print(f"Average latency (optimized): {average_latency_optimized:.2f} seconds") # 估计的优化令牌数量:约1660个令牌我们对新颖查询的新测量结果可能如下所示:平均延迟: 10.5 秒(原13.5秒)平均令牌使用量: 1660 个令牌(原2100个令牌)结果比较我们来可视化改进情况:{"layout": {"title": "链性能优化结果", "xaxis": {"title": "指标"}, "yaxis": {"title": "数值"}, "barmode": "group", "legend": {"traceorder": "normal"}}, "data": [{"type": "bar", "name": "基线", "x": ["平均延迟(秒)", "平均令牌数"], "y": [13.5, 2100], "marker": {"color": "#ff6b6b"}}, {"type": "bar", "name": "优化后(新颖查询)", "x": ["平均延迟(秒)", "平均令牌数"], "y": [10.5, 1660], "marker": {"color": "#4dabf7"}}, {"type": "bar", "name": "优化后(缓存查询)", "x": ["平均延迟(秒)", "平均令牌数"], "y": [1.6, 0], "marker": {"color": "#69db7c"}}]}应用缓存和模型优化方法前后平均延迟和令牌使用量的比较。请注意,缓存查询的效果显著提高。成本影响减少令牌数量直接影响成本。如果llm_initial和llm_refine的总成本为每1K令牌0.002美元:基线成本: (2100 / 1000) * 0.002 美元 = 每次查询 0.0042 美元优化后成本(新颖): (1660 / 1000) * 0.0018 美元(假设更快模型更便宜)≈ 每次查询 0.0030 美元(约节省28%)优化后成本(缓存): 每次查询 0.00 美元(节省100%)权衡与后续步骤我们取得了显著的改进:缓存: 对重复查询非常出色,代码改动极小,缓存本身可能有内存/存储成本。模型切换/提示调优: 减少了所有查询的延迟和成本,但可能在优化回答的质量上存在轻微权衡。评估这种质量差异很重要(第五章有介绍)。这项实践练习展示了一个典型的调优流程:测量: 建立基线。找出: 明确最耗费资源的步骤(时间、令牌、成本)。优化: 应用有针对性的方法,如缓存、模型选择、提示工程或结构性改动(例如,条件执行)。重新评估: 衡量你的改动效果。迭代: 性能调优通常是一个迭代过程。进一步的改进可能包括优化检索步骤,尝试并行执行独立任务,或实现更复杂的缓存。请记住,可以使用LangSmith等工具进行详细追踪和分析,这在复杂应用中能极大简化找出和测量阶段。