3.6 提速改造：性能优化

3.6 提速改造：性能优化

本节摘要：链式应用的成本大头是模型调用，延迟大头是串行等待。本节给四张提速牌——缓存、批处理与异步、模型分级、输入瘦身——每张都讲清见效点与代价，并强调先测量后优化。

先测量，再动手

优化前先问两个数：单次请求平均延迟多少、每月账单多少、各工序占多少。没有这三个数的优化是盲改。用 2.9 节的回调探杆就能量出来：

from langchain_core.callbacks import BaseCallbackHandler
from time import time

class StageTimer(BaseCallbackHandler):
    """分段计时：给每道工序挂秒表"""
    def __init__(self):
        self.marks = []
        self.t0 = None

    def on_chain_start(self, serialized, inputs, **kw):
        self.t0 = time()

    def on_chain_end(self, outputs, **kw):
        self.marks.append(round(time() - self.t0, 2))

timer = StageTimer()
# 挂到链上跑几个真实请求后看 marks 分布
# 输出示例：[1.8, 2.1, 1.7]  ← 模型段稳定吃掉大头
​

链式应用的延迟结构通常是这样：模型调用占六到八成，检索占一到两成，其余是拼装与解析。而成本结构更集中——几乎全在模型调用的 token 数上。四张牌就冲着这两个结构打。

第一张牌：缓存，让重复问题免费

客服场景里"退货政策"一天被问几百次，每次都真调模型就是白烧钱。框架级缓存一行装配：

from langchain_core.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from time import time

# 全局缓存：命中相同请求直接返回 不再调用模型
set_llm_cache(SQLiteCache(database_path=".cache.db"))

chain = (ChatPromptTemplate.from_template("作为客服回答：{q}")
         | ChatOpenAI(temperature=0)   # 温度为0时缓存最有效
         | StrOutputParser())

t0 = time(); chain.invoke({"q": "退货政策是什么"}); t1 = time()-t0
t0 = time(); chain.invoke({"q": "退货政策是什么"}); t2 = time()-t0
print(f"首次 {t1:.2f} 秒，命中缓存 {t2:.4f} 秒")
# 输出示例：首次 1.62 秒，命中缓存 0.0013 秒
​

两个要点：低温或零温搭配缓存才合理（温度高每次结果都不同，缓存命中就成了陈旧答案）；检索类链要连检索结果一起缓存才有意义，只缓存模型段会漏掉检索延迟。

第二张牌：批处理与异步

一批问题一起过线，吞吐立刻翻数倍——2.1 节的 batch 天然把并发交给运行时：

questions = [{"q": f"问题{i}：退货政策"} for i in range(5)]
results = chain.batch(questions)
print(len(results))
# 输出：5（并发执行 总耗时接近单次而非五倍）
​

服务端场景用异步接口，等模型的时间让给别的请求：

import asyncio

async def ask(q: str) -> str:
    return await chain.ainvoke({"q": q})

async def main():
    # 五个请求并发过线
    outs = await asyncio.gather(*[ask(f"问题{i}") for i in range(5)])
    print(len(outs))
    # 输出：5

asyncio.run(main())
​

第三张牌：模型分级，便宜的干粗活

全链一个旗舰模型是常见的浪费。分级思路：分类、路由、格式校验这类轻工序交给小型号，生成、推理这类主工序才上旗舰：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

# 轻型机：做分类 判断 简单改写
lite = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 旗舰机：做最终生成
flagship = ChatOpenAI(model="gpt-4o", temperature=0.3)

classifier = (ChatPromptTemplate.from_template(
        "把问题归类为 售后 或 其他 只输出类别：{q}")
    | lite | StrOutputParser())

def route(q: str) -> str:
    label = classifier.invoke({"q": q})
    if label == "售后":
        return flagship.invoke(
            f"作为资深客服详细回答：{q}").content
    return lite.invoke(f"简洁回答：{q}").content

print(route("我的定制袜子洗缩水了怎么办"))
# 售后类走旗舰 输出更详细
print(route("你们几点上班"))
# 普通类走轻机 输出简洁
​

分级还有个隐藏收益：轻工序换成本地小模型，延迟与隐私同时改善。

第四张牌：输入瘦身

token 是原料，超发输入就是浪费原料。三个高频瘦身点：检索的 k 值从 5 降到 3（多数场景答案在前两块）；历史记忆用窗口或摘要（2.5 节的裁剪策略）；系统提示去掉自我感动式的长篇人设，留可执行的行为规则。

⚠️ 每张牌打完都要回 3.3 节质检台跑分：缓存看命中率与陈旧率、分级看分类准确率、瘦身看答案质量分数。优化把质量优化没了，是最贵的返工。

💡 优先级建议：缓存与输入瘦身几乎零风险先做；批处理异步在有并发需求时做；模型分级动静最大，等前三张的收益吃干后再上。

本章收官

四类改装全部完成：零件可自制（3.1）、外协可接入（3.2）、质量可度量（3.3）、故障可定位（3.4）、风险可防护（3.5）、性能可运营（3.6）。产线具备了上线资格。下一章进入总装车间：先看四类典型产线的完整图纸，再把其中一条知识库问答产线从零装配到部署成服务。