大模型响应缓存实战：TTL缓存装饰器设计与落地

大模型应用跑起来之后，最先让人头疼的往往不是模型效果，而是账单。每次对话都要实打实调用一次大模型接口，按token计费，高峰期一秒几十个请求，链路一慢用户就开始抱怨。我最早做某知识库问答系统时，第一版上线不到一周，成本报表就让我清醒了：同样的用户问题，反复问、换个说法问、不同入口问，背后都在重复烧钱。后来我动手给响应层加了一套TTL缓存装饰器，把重复计算直接命中缓存，成本和响应时间肉眼可见地降了下来。这篇就是我落地这套方案的全过程，包括设计思路、核心代码、性能实测和踩坑记录，适合正在做类ChatGPT应用、智能客服、内容总结这类大模型产品的开发者参考。

1. 为什么要给大模型响应加缓存：账不算清，方案没法选

1.1 大模型调用的真实成本组成

先说成本。大模型API的计费一般按输入token加输出token来算，表面看一次调用几厘钱到几分钱不等，但实际业务里远不是这么简单。一个典型问答请求，可能要带上系统提示词、历史对话、检索到的知识片段，再加上用户当前问题，输入token常常是输出token的三到五倍。对话一长，整个上下文反复重发，每一轮都是全额计费。

更要命的是延迟。一次完整的大模型生成，短则两三秒，长则十几秒。如果业务里有查询接口、批量处理任务，响应时间直接拖垮用户体验。我见过一个内容审核系统，每条文本都调大模型判断，单条机器耗时平均四秒，整个流水线卡在模型调用上。

这两部分成本，就是缓存要解决的直接目标。缓存命中时不需要再走网络请求，省掉API费用，也省掉等待生成的这段时间。

1.2 什么场景值得缓存，什么场景不该缓存

不是所有请求都适合缓存，这个判断很重要，不然要么缓存没用，要么缓存出错。我总结下来，适合缓存的大模型请求通常有几个特征：

• 请求可重复：同一问题或近似问题会被不同用户、不同会话反复提出。
• 结果相对稳定：答案不依赖时间、不依赖个人私有数据，或者即使略微变化，用户感知不强烈。
• 时效性要求低：比如知识库条目解释、常见FAQ问答、商品卖点生成，答案不会几分钟就变。
• 对延迟敏感：缓存能让首包时间从几秒降到几十毫秒，体验改善非常明显。

不适合缓存的场景也很典型：实时性强的股票行情解读、依赖用户私有上下文的个性化回复、法律或医疗等要求每次核实的场景。这些如果硬套缓存，要么返回过期信息造成事故，要么因为key变化太碎导致命中率极低，白白维护一套复杂逻辑。

1.3 为什么最终选择TTL缓存装饰器这套方案

当时我对比过几种方案。第一种是做HTTP层缓存，直接缓存接口响应，好是好，但颗粒度太粗，session、用户ID、时间戳混在请求头里，很难统一控制key。第二种是用外部存储如Redis做通用缓存，灵活，但需要单独部署，小团队维护起来麻烦。第三种是写个内部工具函数，每次调用前手动查缓存、写缓存，也能用，但代码会散落到各处，加缓存逻辑的地方多了之后，到处都是重复代码，改个策略要全局搜索。

最后我选了装饰器方案。装饰器可以把缓存逻辑和业务逻辑彻底分离，函数定义上挂一行注解就完成了缓存接入，调用方无感知。TTL即Time To Live，给每条缓存设置存活时间，到点自动失效，既能保证数据不过时，又不需要额外的过期清理线程。这个组合非常契合大模型响应场景：语义明确的文本请求做哈希作为key，响应结果存起来，五分钟内直接复用，五分钟之后重新生成，逻辑清晰，扩展也方便。后续想换成Redis后端，只需要改装饰器内部的数据存取部分，业务代码几乎不用动。

2. 核心设计：接口、数据结构与关键决策

2.1 需求梳理：装饰器要解决的五个问题

动手写代码之前，我把装饰器要满足的需求列成了清单，后面所有实现都围绕这些点展开。

第一，支持同步和异步函数。大模型调用绝大多数是异步IO，但同时也有少量同步调用场景，装饰器必须两套都支持，不然用到哪套写哪套会让代码分裂。

第二，缓存key的可控性。默认情况下，装饰器应该根据函数名和参数自动生成key，但也要允许调用方手动指定某些参数不参与缓存计算，比如调试用的verbose标记、临时埋点参数。

第三，TTL过期时间可配置。有的接口想缓存十分钟，有的缓存两分钟，装饰器应该支持在装饰器参数里指定默认TTL，也支持函数调用时临时覆盖。

第四，并发控制。缓存过期瞬间，如果有十个请求同时到达，不能十个人同时去调用大模型接口，否则缓存不但没省成本，反而制造了峰值。所有相同key的请求必须合并成一次模型调用。

第五，容错。缓存系统是加速层，不能因为缓存挂了导致业务失败。缓存读写出现异常时，应该自动降级为直接调用原始函数。

2.2 数据存储选择：先本地内存，再考虑外部化

第一版我直接用进程内字典存储，原因很直接：大多数大模型应用中，缓存的主要价值是挡住短时间内的重复请求，这些重复请求大概率落在同一个服务实例上。我们当时服务实例数量不多，平均重复率超过六成，单机内存缓存已经能覆盖大部分收益。进程内字典不需要网络IO，读取开销接近零，也没有Redis连不上、序列化出错这类额外故障点。

有人问，多实例部署时缓存不就各存各的了吗？确实会，但这个问题可以通过两层缓存来解决：本地内存TTL缓存做第一层，Redis做第二层。第一层挡住单实例内的重复请求，第二层挡住跨实例的重复请求。第一版连第二层都没做，实测单层命中率已经很理想，后面再加也不迟。设计接口时留好存储后端抽象就行了，不必一开始就引入Redis。

2.3 缓存key的设计：输入归一化比想象中重要

缓存key的设计是整个方案成败的关键。大模型接口的输入一般是一个结构化对象，包含system、messages、temperature、max_tokens等字段。直接把整个对象序列化做哈希当然可以，但往往过于严格：用户问"什么是TTL"和"TTL是什么"，得到的结果可能是同一个意思，但哈希值完全不同，缓存永远命不中。

所以我在生成key之前增加了一步输入归一化。Messages数组里，把对话内容做简单的文本规范化处理，比如统一全半角、压缩多余空白、去除表情符号。注意，这一步只能做表层归一化，复杂语义归一化在本场景里会引入误判风险，比如把意思不同但字面接近的问题错误合并，反而返回了错误答案。所以默认策略是保守归一化，宁可漏命中，不可错命中。

参数级别上，temperature和top_p这类采样参数会直接影响生成结果，必须参与key计算。而request_id、trace_id这类调试字段必须排除在key之外。我用了一个名为skip_params的参数列表来声明哪些参数忽略不计。

3. 完整实现：从基础版到生产级装饰器

3.1 基础实现：定义一个缓存存储类

先写一个最简单的存储类，内部用字典加过期时间戳来管理数据。这一步不引入任何外部依赖，所有逻辑只有几十行，适合理解TTL缓存的最小工作原理。

import time import threading class TTLCache: def __init__(self): self._data = {} self._lock = threading.Lock() def get(self, key): item = self._data.get(key) if item is None: return None value, expire_at = item if expire_at < time.time(): with self._lock: self._data.pop(key, None) return None return value def set(self, key, value, ttl): expire_at = time.time() + ttl with self._lock: self._data[key] = (value, expire_at) def delete(self, key): with self._lock: self._data.pop(key, None)

这段代码的核心在于，每条数据保存的是value + 过期时间戳，读取时比对当前时间，过期就直接视为缺失并顺手清理。锁的唯一作用是保护字典的并发读写，避免多线程环境下出现数据竞争。get里并没有全程持锁，因为GIL保证了单条字典操作的原子性，先读后判的瞬时竞态在业务上可以容忍。

这个类很简单，但已经具备完整TTL语义：数据放进去，到点消失，不需要任何后台任务。对大模型响应这种秒级到分钟级有效期的数据，完全够用。

3.2 装饰器主体：同步函数版本

有了存储类，接下来写装饰器。我先写同步版本，因为逻辑更直观，方便讲清楚每个环节。

import hashlib import functools import inspect cached_funcs = {} def ttl_cache(ttl=300, max_size=1024, skip_params=()): def decorator(func): cache_store = TTLCache() @functools.wraps(func) def wrapper(*args, **kwargs): key = _make_key(func, args, kwargs, skip_params) cached = cache_store.get(key) if cached is not None: return cached result = func(*args, **kwargs) cache_store.set(key, result, ttl) return result cached_funcs[func.__name__] = cache_store return wrapper return decorator

_make_key是核心，它的任务是把函数名、位置参数、关键字参数统一编码成稳定的哈希值。具体做法是把所有参数收集起来，按参数名排序，再和函数名一起做SHA256哈希。用哈希值而不是原始字符串做key，是为了避免超长文本直接成为内存里的键名。一千个token的输入文本直接做key，内存和比较开销都不小，哈希后固定长度则完全可控。

这里还有另一个容易被忽略的点：key的稳定性。Python内置的hash()函数每次进程启动会随机化，所以绝对不能用于缓存key，否则服务重启后所有旧key全部失效，缓存形同虚设。我用的是hashlib.sha256，保证在不同进程、不同时间点对相同输入产生相同的key，这样即使重启后内存缓存清空了，将来接外部存储时，key也能跨实例复用。

3.3 支持异步函数：规避协程中的锁陷阱

大模型SDK基本都是异步接口，所以异步版本的装饰器是生产中真正高频使用的。

import asyncio import functools def ttl_cache_async(ttl=300, max_size=1024, skip_params=()): def decorator(func): cache_store = TTLCache() @functools.wraps(func) async def wrapper(*args, **kwargs): key = _make_key(func, args, kwargs, skip_params) cached = cache_store.get(key) if cached is not None: return cached async with asyncio.Lock(): cached = cache_store.get(key) if cached is not None: return cached result = await func(*args, **kwargs) cache_store.set(key, result, ttl) return result return wrapper return decorator

注意这里我在拿到锁之后又做了一次缓存查询，这就是经典的double-check。第一次查询没命中，说明当前没有有效缓存，但可能已经有另一个协程正在生成结果。如果直接冲进去调用模型，就会造成重复请求。拿到锁后再次检查，如果刚才那个协程已经把结果写进缓存，就直接返回，避免重复计算。

锁的粒度是全局的，理论上不同key的请求也会互相等待，但大模型调用本身耗时是秒级，而缓存读写的耗时是微秒级，锁竞争的开销完全可以忽略。相比为每个key维护独立锁的复杂度，这种"粗锁+double-check"的方案更加工程化，代码不容易出并发bug。

3.4 参数序列化与skip_params实现

_make_key函数把参数处理拆成三步。第一步，过滤掉skip_params声明的参数，这些参数可能每次请求都变，但它们的变化不该影响缓存决策。第二步，把剩余参数转成一个规范化字典，位置参数按函数签名映射到参数名。第三步，把字典序列化成JSON，序列化时用sort_keys=True保证字段顺序一致。

def _make_key(func, args, kwargs, skip_params): sig = inspect.signature(func) bound = sig.bind_partial(*args, **kwargs) bound.apply_defaults() params = bound.arguments processed = {} for name, value in params.items(): if name in skip_params: continue processed[name] = _normalize(value) raw_key = func.__name__ + json.dumps(processed, sort_keys=True, default=str) return hashlib.sha256(raw_key.encode("utf-8")).hexdigest()

这里有个实用细节：inspect.signature可以把位置参数和关键字参数统一映射到参数名上，避免同一个函数用(query, top_k=3)和(query, 3)两种调用方式产生不同的key。如果不用签名绑定，这两种调用会被视为两个缓存条目，命中率白白损失。default=str用于处理字典值里嵌套的自定义对象，比如Message对象，虽然不够精细，但至少不会因为序列化失败抛异常。

在实际业务里，我会为Messages里每个角色单独建字段名，再统一处理文本字段。整体指导思想就是：让相同语义的请求落在同一个key上，让不同语义的请求尽量落在不同key上。

4. 落地实践：缓存命中率与生产化改造

4.1 实测：一个知识库问答系统的缓存效果

装饰器写完后，我第一时间接入了知识库问答系统的查询接口。这个场景非常典型：用户先问一个问题，得到答案后，还会顺着答案追问细节，所以同一个问题会被不同会话反复问。

接入方式是找接口里最耗时的函数，在它上面挂装饰器：

@ttl_cache_async(ttl=300, skip_params=("request_id",)) async def generate_answer(question: str, context: str, temperature: float = 0.3): messages = _build_messages(question, context) resp = await async_client.chat.completions.create(...) return resp.choices[0].message.content

跑了两周，我统计了缓存命中率。整体请求总量约12万次，缓存命中了7.8万次，命中率约65%。按单次请求平均token费用和相关成本估算，这7.8万次请求每次省下的成本大致相当于正常调用的一半以上，因为不仅省了模型推理费，还省了上游检索、日志处理等一系列联动开销。响应时长方面，命中缓存的请求平均从4.2秒降到了280毫秒，约15倍的延迟改善。

最明显的变化是重复问题的响应时间。用户问"什么是引用计数"，第一次等了三秒，第二次刷新页面，秒开。这种体验差异，用户端感知非常强烈，客服侧收到的"系统很慢"类反馈明显变少了。

4.2 缓存容量控制：不能让内存无限涨

内存缓存最大问题是容量。每条大模型响应的文本长度可能从几百字到几千字不等，如果无限缓存，服务进程的内存会被撑爆，尤其是知识库场景，上下文有几百上千Token，响应再附带引用片段，单条缓存轻松超过10KB。十万个key就是1GB内存，相当可观。

我采用了两层保障。第一层是设置max_size，装饰器在写入缓存时检查当前条目数量，超过阈值就开始清理。清理策略用的是近似LRU，做法是每个key记录最后一次访问时间，清除时按访问时间排序，淘汰最旧的20%。这么做比完整LRU实现简单，效果上足够接近。第二层是限制单条缓存的最大字节数，比如超过50KB的响应不缓存。理由很直接，超长内容本身出现频率低，缓存它们的内存收益远低于短内容，而且长文再生成一次的成本虽高，但既然低频，缓存价值的兑现率也低。

def set(self, key, value, ttl): if self._max_size <= 0: return with self._lock: if len(self._data) >= self._max_size: self._evict(ratio=0.2) expire_at = time.time() + ttl self._data[key] = (value, expire_at) def _evict(self, ratio): now = time.time() items = sorted( self._data.items(), key=lambda kv: kv[1][0] # 实际应存最后访问时间 ) remove_count = max(1, int(len(items) * ratio)) for k, _ in items[:remove_count]: self._data.pop(k, None)

这里顺便提醒一句，缓存淘汰要管住的是写入路径，而不是等到内存报警才被动清。主动淘汰配合过期删除，内存水位才能真正稳定。

4.3 缓存穿透、击穿与雪崩：按场景逐个处理

这三种问题在缓存领域是标配，但大模型场景下表现形式不太一样。

缓存穿透是指查询一个根本不可能有缓存结果的key，比如随机生成的乱码文本，每次都直接打到模型接口。在大模型场景里，穿透问题通常不严重，因为正常业务下请求都是真实用户问题，不会是恶意构造的乱序字符。但批量测试、刷接口时可能出现，简单做法是对空结果也做短TTL缓存，比如3秒，这样即使某个异常query大量到来，也能挡住大部分重复调用。

缓存击穿是指某个热点key过期瞬间大量请求同时涌来。我们的业务里，热点问题集中在少数几个高频FAQ上，确实发生过一次：某个运营活动页上线，所有用户都问同一个商品退换货条款，结果恰好缓存过期，一瞬间几十个请求全部打到模型接口。解决办法就是前文提到的double-check加锁，让同一key的并发请求合并为一次。如果再想精细一点，可以实现每个key单独一个锁对象，把锁竞争面缩小。

缓存雪崩在大模型场景往往不是缓存层本身的问题，而是多个key同时过期时，下游模型服务被打爆。我们当时的缓解方案是给TTL加入抖动，也就是基础TTL加上一个随机偏移量，比如300秒到360秒之间随机，避免所有key在同一时刻集体失效。

import random def _jitter_ttl(self, ttl): return int(ttl * random.uniform(1.0, 1.2))

这个思路简单有效，尤其适合批量导入缓存和定时任务场景。

4.4 缓存存储后端的扩展：从内存到Redis

内存缓存上线两周后，我发现有一个路由服务有两个实例，虽然单实例命中率还行，但两个实例之间存在大量重复的模型调用，白白浪费钱。于是给存储层做了一层抽象，引入Redis后端。

做法是定义一个CacheBackend接口，包含get和set两个方法，内存版和Redis版各自实现。装饰器内部只依赖这个接口，切换后端只需要改一处初始化配置。

class CacheBackend: def get(self, key): raise NotImplementedError def set(self, key, value, ttl): raise NotImplementedError class RedisBackend(CacheBackend): def __init__(self, redis_client): self._redis = redis_client def get(self, key): raw = self._redis.get(key) if raw is None: return None return pickle.loads(raw) def set(self, key, value, ttl): self._redis.set(key, pickle.dumps(value), ex=ttl)

要说清楚，这里pickle序列化仅适用于可信任的内部环境，如果缓存数据会经过不可信来源，要换成JSON序列化或者先用msgpack，避免反序列化安全问题。Redis版的好处是多个实例共享同一份缓存，key的命中范围从单实例扩展到了整个集群。

不过也有一点代价，每次缓存读取多了一次网络往返，大约0.2到0.5毫秒，相比大模型接口的秒级延迟，这点开销完全可以忽略。引入Redis后，整体命中率从65%提升到了78%，几个实例的重复调用几乎被消除了。

5. 常见问题与运维实战记录

5.1 命中率低于预期：先用日志定位key碎掉的原因

上线初期，命中率曾经低到只有20%多，当时第一反应是装饰器写错了，后来通过日志定位才发现，问题出在参数上。业务方在调用生成接口时，习惯性地传入了一个timestamp字段，日志里显示每次请求的timestamp都在变化，几乎每一秒都是全新key。这个字段其实只是用来记录请求发起时间，完全不参与生成逻辑，但由于没有在skip_params里声明，导致每次请求都被当成不同请求处理。

解决方式很简单，把timestamp加进skip_params，命中率立刻从20%回升到50%以上。

后来还遇到一种情况：前端页面上用户提问时会自动在句末补一个空格，有的用户输入法还会带上全角标点。这些细微差异都会被哈希算法放大成完全不同的key。我们的对策是在进装饰器之前，对question字段做一次预处理：

def _normalize_text(text): text = text.replace("\u3000", " ").replace(" ", " ") text = re.sub(r"\s+", " ", text) text = text.strip() return text

5.2 缓存数据异常：永远不要直接信任缓存结果

缓存返回的是历史生成结果，一旦业务需求调整了生成逻辑，比如改了提示词模板、换了模型版本，旧缓存会变成"过期正确答案"，用户看到新提示词下应该出现的格式完全对不上。

这类问题最隐蔽，因为程序不会报错，只有业务反馈说"回答风格不对"才会发现。我后来在装饰器里加了一个cache_version参数，业务层调整提示词或模型版本时手动递增版本号，key计算时带上版本号，旧缓存自然全部失效。这是比单纯等TTL过期更主动的数据一致性手段。

def _make_key(func, args, kwargs, skip_params, cache_version): # ... raw_key = f"{cache_version}:{func.__name__}:{json.dumps(processed, sort_keys=True)}" return hashlib.sha256(raw_key.encode("utf-8")).hexdigest()

5.3 装饰器参数传递陷阱：直接挂装饰器导致函数签名变化

给函数挂装饰器后，如果不加functools.wraps，函数签名、__name__、__doc__全都会变成wrapper的信息。对于不依赖反射的业务可能无所谓，但一旦API文档自动生成或者依赖函数元信息做路由，就会出问题。我的习惯是所有装饰器一律加functools.wraps，这行代码成本为零，但省掉很多排查麻烦。

还有一个相关坑，当被装饰的函数定义在类里时，self参数会自动进入args，导致不同实例调用同一个方法时生成不同的key。解决方案是在装饰器初始化时检查第一个参数名，如果是self或cls，就把这个参数排除在key计算之外，或者要求在skip_params里显式声明。

5.4 排查速查表：遇到问题先对号入座

我把平时最容易遇到的问题整理成了一个速查表，团队新成员接入这套装饰器时，拿着对照排查会快很多。

5.5 缓存监控与告警：让问题在影响用户之前暴露

缓存系统本身也要可观测，否则出了问题只能等用户投诉。我在装饰器里埋了三个指标，通过日志或者指标上报到监控平台。

第一个指标是缓存命中率，按分钟聚合，低于某个阈值触发告警。这个指标最能反映key设计和业务波动，命中率突然下降往往意味着参数结构改变、请求分布变化或者代码刚发布了新版本。第二个指标是缓存写入数量，写入骤增说明大量请求未命中，需要检查是否发生了缓存穿透，或者key设计里混入了高基数字段。第三个指标是缓存淘汰数量，淘汰数量长期很高说明容量设置不合理，或者max_size相对于业务体量太小。

在实际运维中，有一次发现淘汰数量点状暴增，排查后发现是某个爬虫在短时间内请求了大量长尾问题，导致新缓存不断写入，把热点缓存挤出去了。后来我把淘汰策略改成了同时保护热点key，具体做法是记录每个key的命中次数，命中次数高的key在淘汰时降低优先级，相当于给LRU加了一层频率保护。

这套方案落地后，我在几个不同业务场景里复用，包括关键词提取、文本分类和摘要生成，基本都是挂装饰器、调整TTL和skip_params就能跑起来。TTL缓存装饰器本身并不复杂，真正考验人的是缓存key的设计、并发控制和业务变更时的版本一致性管理。把这些细节想清楚，大模型的重复调用成本就能被稳稳地挡在缓存层外面。最后再分享一个经验：上线缓存之后一定要把命中率统计接进日常监控，否则你根本无法判断系统的成本优化是否真正生效，也无法提前发现业务变化带来的缓存风暴。