1. 为什么一个AI助手需要先解决"记得快"的问题
先交代一下背景:这是"个人AI助手"系列的第7篇实践记录,前面几个阶段已经把这台助手从"能调API的大模型客户端"逐步做到了"有记忆、有工具调用能力的Agent"。阶段1打通了模型接口,阶段2给它加了搜索、网页抓取、日程提醒这类工具能力,整体框架已经能跑通。但用到第3周的时候,问题开始暴露了——每天高频使用的真实场景里,这个助手有两大硬伤:太慢、太贵。
慢体现在哪里呢?一次对话的耗时分布大概是:网络往返几百毫秒,模型首token响应一到三秒,回答长问题加上流式输出又是几秒,如果再触发工具调用,一次完整回答动辄六七秒起步。贵就更直接了,模型API按token计费,同样是"今天下午的会议改到几点""帮我查一下明天的天气"这类高度重复的问题,每天能问上好几遍,但每一遍都在按全价付费。
顺着这两个痛点往深了想,这其实不是一个"模型选型或者提示词优化"能解决的问题。模型再快也有物理时延,提示词再精炼也改变不了重复请求的浪费。合适的做法是引入一层缓存系统,让高频、高重复的请求在到达模型之前就命中离线的答案。这篇实践记录就围绕阶段3的缓存系统搭建展开,包括整体架构设计、语义缓存的具体实现、实测效果,以及我自己踩过的几个比较典型的坑。
这个阶段的目标定义得很明确:在不改变回答质量的前提下,通过缓存把重复请求的响应时间压缩到五分之一以内,同时把模型API的调用量降下来。文章适合正在做个人AI助手、本地知识库问答、或者任意一个带大模型调用的上层应用的开发者参考,尤其是那种"功能已经通了,但性能和成本还没过关"的阶段。
2. 先明确缓存系统的边界:哪些请求值得被缓存
缓存不是万能药,不是所有请求都适合塞进缓存系统里。设计阶段首先得划定边界,确定这个缓存要吃什么数据、不吃什么数据,否则后面做出来的系统要么命中率感人,要么回答质量大幅下降。
2.1 三种可缓存的数据类型
我给自己的助手梳理了一遍对话场景,把值得缓存的请求分成三类。
第一类是"事实查询类"。比如"帮我查一下XX接口的文档""Python 3.12的typing有什么变更""某软件的最新版本号是多少"。这类问题的特点是答案客观、稳定、变化慢,问10次正确回答都是同一个。传统做法是直接把整个问答对缓存住,命中直接重放。
第二类是"工具结果类"。比如天气查询的结果、某个网页抓取到的正文内容、论文摘要、RSS聚合结果。这类问题本身问法可能千变万化,但背后的工具调用结果是稳定可复用的。如果不做缓存,每问一次"明天天气如何"就真的再去调一次天气API,不仅慢,某些免费API还有每日请求配额限制。把工具返回结果单独缓存,就是给工具调用本身加了一层数据源缓存。
第三类是"嵌入向量检索类"。这是本地知识库问答场景特有的一类:用户先提问,系统把问题向量化,到向量数据库里做相似度检索,把命中片段塞进上下文,再交给模型生成答案。这里的检索结果同样具备高复用性——同一个问题向量化后大概率命中的是同一批文档片段。
2.2 不值得缓存的反面清单
有几类请求我明确排除了。首先是"创作生成类",让助手写一段宣传语、改一篇文案、编一个故事,这类任务没有唯一标准答案,如果硬缓存同一个答案,用户期望的是每次微调后的多样化输出,缓存显然会帮倒忙。
其次是"个人状态类"。问"我的日程安排""我的待办事项"这种依赖用户私有上下文的问题,同一句话不同时间问,答案完全不同。缓存这种请求不仅没有用,还可能因为缓存过期策略设计不严把过期数据当作最新数据返回,造成严重的信息错误。
再就是"工具动作类"。比如"帮我订阅一下某网站的更新""把某篇文章存到收藏夹",这类请求做的是副作用的写入操作,绝对不能走缓存。任何会改变外部系统状态的请求,都应该默认绕过缓存系统。
划定清楚边界之后,实现的策略逻辑也就清晰了:根据用户的提问内容做一个初步的请求分类,事实查询类和工具结果类走缓存,创作类和个人类直接放行到模型。分类本身不需要多聪明,简单的关键词规则加上意图分类模型即可。
3. 缓存架构的设计:多层缓存与三种核心策略
边界定好了,接下来是架构层面的事情。我最后确定的方案是两层缓存:内存缓存做热数据层,向量语义缓存做冷数据层。为什么是两层,而不是一个统一的缓存池?这里面其实有很现实的取舍逻辑。
3.1 内存缓存层(L1)
第一层是进程内内存缓存,本质就是一个带过期时间的字典。它的优势是访问速度极快,纳秒级取数据,完全不涉及跨进程通信。用Redis这类独立的缓存服务其实也很常见,但对于个人助手这样的单机应用,进程内存的延迟更低、部署更简单、不需要额外维护一个服务。
内存层存的到底是什么?不可能是所有的问答对,内存空间有限。它只存那些高频访问的极热数据——比如用户最近半小时内反复查询的同一个问题。我设计了一个只要命中就自动刷新TTL的机制,保证热数据一直被留在内存里。
内存层的淘汰策略用最简单的LRU,加一个最大条目限制。我的做法是设置最多缓存5000条问答,超过之后按最近最少使用原则淘汰。这个数字是个人使用场景下比较合适的值,如果读者是服务端团队使用,可以根据内存配额和单条缓存平均大小重新估算。
3.2 语义缓存层(L2)
第二层是语义缓存,这也是这个阶段最核心的工程难点。普通缓存的匹配逻辑是"键值精确匹配",用户两次提问必须一字不差才能命中。但自然语言不是这样,用户可能今天问"明天的天气怎么样",明天问"明天会不会下雨",两个句子字面不同,但语义高度相近,期望得到的答案几乎一致。
语义缓存的实现思路是:把用户提问嵌入成向量,存入向量索引;新请求进来时,同样转成向量,和已有缓存条目做相似度计算,相似度超过设定的阈值就判定为语义命中。这相当于给缓存系统装了一个"能看懂语义等价"的翻译层。
这样设计带来的直接好处是命中率大幅提升。传统精确缓存命中率可能只有20%到30%,因为实际对话中很少有人一字不差地问同一个问题;但换成语义缓存之后,命中率能提升到50%以上,用户的重复问题往往只是换了种说法。
3.3 TTL策略:不同数据要有不同寿命
缓存能不能正常工作,很大程度上取决于过期时间策略设得是否合理。我把TTL拆成了三个档位:
| 数据类型 | 过期时间 | 理由 |
|---|---|---|
| 常识事实类问答 | 7天到30天 | 比如编程语法、工具用法,内容变化慢 |
| 时效类工具结果 | 30分钟到6小时 | 天气、新闻、股市这类信息必须保新鲜 |
| 向量检索片段 | 1天到7天 | 文档内容相对稳定,但保留更新空间 |
这里有一个比较微妙的问题:TTL设置不能一刀切。我踩过的坑就是最开始把所有缓存都设成24小时过期,结果天气预报被缓存了整整一天,用户早上问天气拿到的是昨天的答案,这种数据性错误比慢几秒严重得多。后来改成按数据类型动态决定TTL,才解决了这个问题。
3.4 缓存键如何设计
不管是精确缓存还是语义缓存,缓存的键都要可控可追踪。我的做法是:精确缓存用规范化后的用户提问文本做键,做一步预处理——去掉多余空格、统一中英文标点、小写化;语义缓存则用嵌入向量做索引,同时把原文保留在元数据里,便于调试时看到底是哪条语义命中了。
规范化这一步在中文场景下特别重要。同一句话在手机上自动输入法可能带全角括号,电脑上可能是半角,不做规范化的话,字面完全一致的句子都可能因为一个字符的差异而错过缓存命中。
4. 核心代码实现:一个能跑的语义缓存系统
架构落定之后,下一步是把它翻译成代码。这里给出我自己实现的最小可用版本,整体偏Python风格,便于读者直接改造成自己项目的一部分。需要注意,这里展示的是核心骨架,生产级细节(比如并发锁、持久化、监控指标)需要根据自己的场景补全。
4.1 本体:一个TTL缓存装饰器
先看最简单的一层,基于精确匹配的内存缓存。我把缓存逻辑封装成一个装饰器,使用方只需要标记"这个函数可以被缓存",而不需要理解内部细节。
import time import functools import threading from collections import OrderedDict class TTLCache: def __init__(self, capacity: int = 5000, default_ttl: int = 3600): self.capacity = capacity self.default_ttl = default_ttl self._store = OrderedDict() self._lock = threading.Lock() def get(self, key): with self._lock: if key not in self._store: return None value, expire_at = self._store[key] if time.time() > expire_at: del self._store[key] return None # 把命中的条目移到末尾,维持LRU顺序 self._store.move_to_end(key) return value def set(self, key, value, ttl=None): ttl = ttl or self.default_ttl with self._lock: if len(self._store) >= self.capacity: self._store.popitem(last=False) self._store[key] = (value, time.time() + ttl)装饰器的实现思路是:先算缓存键,然后查缓存,命中直接返回;没有命中就执行原函数,手动把结果回填到缓存。
def cached(ttl_cache: TTLCache, key_prefix: str = "", ttl: int = None): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): raw_key = f"{key_prefix}:{args}:{kwargs}" cache_key = normalize_key(raw_key) cached_value = ttl_cache.get(cache_key) if cached_value is not None: return cached_value result = func(*args, **kwargs) ttl_cache.set(cache_key, result, ttl=ttl) return result return wrapper return decorator真实项目里,缓存的回填需要做"异步写"处理。如果每次缓存未命中都同步等待模型返回并写缓存,那第一次请求的延迟并没有任何改善。实际做法是:第一个请求正常执行模型调用,后台线程把结果写入缓存,同一时间窗口内到达的相同请求可以等待第一个请求的结果,而不是各自重复调用模型。这个模式叫"单飞"(singleflight),实测下来在高并发下非常有用。
4.2 语义缓存的判断逻辑
语义缓存的核心函数是这个sentence_similarity 判断。我这里用的是嵌入模型加余弦相似度,在两三万条缓存规模下足够用。
import numpy as np from sentence_transformers import SentenceTransformer class SemanticCache: def __init__(self, threshold: float = 0.92): self.model = SentenceTransformer("your-embedding-model") self.threshold = threshold self._items = [] # 元素结构: {vector, question, answer, timestamp} def query(self, question: str): q_vec = self.model.encode(question, normalize_embeddings=True) best_score = 0.0 best_item = None for item in self._items: score = float(np.dot(q_vec, item["vector"])) if score > best_score: best_score = score best_item = item if best_score >= self.threshold: return best_item["answer"] return None def add(self, question: str, answer: str): q_vec = self.model.encode(question, normalize_embeddings=True) self._items.append({ "vector": q_vec, "question": question, "answer": answer, "timestamp": time.time(), })这个实现有几个性能问题:线性扫描意味着缓存条目越多越慢;同时向量全放内存也比较吃空间。个人助手场景下几千条缓存完全没问题,但如果要做到几百万条,就需要改成faiss这样的最近邻索引,避免全量线性扫描。扩展方向我这里先提一下,后文会具体讲。
4.3 触发策略:何时走语义缓存,何时必须绕过
有读者可能会问:语义缓存需要额外调嵌入模型,这不是一样增加耗时吗?确实,如果语义缓存引入的嵌入计算时间比直接调大模型还长,那就本末倒置了。所以我把语义缓存的触发逻辑做了个分级:
- 先走L1内存精确缓存,命中即为最优。
- 如果L1未命中,再走L2语义缓存,此时嵌入模型的计算时间大约在几十毫秒量级,相比大模型的一到三秒,开销完全可以接受。
- 如果L2最终也未命中,这次请求就直接调用大模型,并把结果异步写入两级缓存。
绕过路径的设计同样重要:如果用户显式要求"重新回答""换一种写法""不要说刚才那套话",那么这个请求必须强制绕过缓存,直接交给大模型重新生成。不然就会出现用户想听一个不一样的版本,系统却把旧答案原封不动端上来的尴尬场景。触发这种绕过的方式很简单:在请求分类阶段检测到包含"再来一次""换一个""重新说"等关键词时,在缓存键上强制加一个随机前缀,让缓存必然失效。
4.4 持久化问题
缓存不能只存在进程里,因为个人助手很可能每天都重启。每天重启后需要花很长时间重新预热,头几轮请求全部缓存未命中,用户体验会比较差。所以阶段3还做了一步持久化:把内存缓存在每次写入时同步写一份到本地磁盘,采用 JSON Lines 格式逐行追加;启动时扫描加载。
这里有一个经验性的细节:别把整个缓存快照一次性写到磁盘。几万条数据按行追加的方式,启动加载也就几百毫秒,和一次快照序列化几乎没区别,但按行追加的容错性好很多——即使上次崩溃导致最后几行损坏,加载逻辑也能直接跳过坏行继续加载,整个缓存不会因为尾部数据损坏而全部报废。
5. 实测效果:命中率、响应时延和成本账单
系统跑了两周真实使用数据后,我把效果指标整理了一下。这部分对评估缓存系统值不值得做非常关键。
5.1 缓存命中率的真实分布
| 指标 | 精确缓存命中率 | 语义缓存命中率 | 总缓存命中率 |
|---|---|---|---|
| 高频问答场景 | 28% | 31% | 59% |
| 工具查询场景 | 35% | 41% | 76% |
| 知识库检索场景 | 15% | 43% | 58% |
| 整体 | 26% | 33% | 59% |
从数据可以很清楚地看到,精确缓存单独用的效果确实不够理想,26%的命中率意味着四分之三的请求仍然要走模型;但是加上语义缓存之后总命中率能到59%,接近六成的请求不再消耗模型算力。
5.2 响应时延的对比
我特意统计了不同路径下的端到端响应时延:
- L1内存缓存命中:平均60毫秒,基本就是函数调用加返回的开销。
- L2语义缓存命中:平均180毫秒,多出来的120毫秒主要花在嵌入模型推理和相似度计算上。
- 缓存未命中直接走模型:平均3800毫秒,其中模型生成占了大头。
这个对比非常直观:缓存命中之后端到端延迟从接近4秒降到不到200毫秒,体验完全是两个级别。日常使用中,高频重复问题基本都是秒开,这种流畅感对个人助手的使用意愿提升非常明显。
5.3 token成本估算
成本侧的收益也很好算。假设日均发生200次用户请求,没有缓存时全部直接调用模型,平均每次消耗2000 token,按当前常见的中大型模型的计费标准估算,一个月的token成本在高频使用下会比较可观。接入缓存后总命中率约59%,等于只有41%的请求真正产生模型费用,成本直接降到原来的四成出头。
这里还没有计算嵌入模型的成本。语义缓存用的嵌入模型很小,一次调用通常只有一两个token的花费量级,和主模型的生成成本相比几乎可以忽略。如果是在预算紧张的环境里做这套系统,这个成本结构是一种性价比很高的组合。
6. 踩坑复盘:这五个问题我排查了最久
这一阶段最有价值的其实是踩坑过程。五个问题花了不少时间排查,但每一个都让系统真正变得"能打"了,而不是停留在demo层面。
6.1 把嵌入模型加载放在了每次请求里
第一版代码把嵌入模型的初始化写在了查询函数内部。看起来没什么问题,语义缓存第一次被访问时才会加载模型,但项目跑起来之后发现第一次请求要等待将近5秒。原因很简单:嵌入模型首次加载需要把几千万个参数读进CPU和内存,做权重初始化,这个过程在没有预热的情况下被叠加到用户请求路径上。
解决办法是在服务启动阶段主动加载嵌入模型,并把加载放到后台线程;同时提供一个健康检查接口,确认模型就绪后才接收外部流量。如果你的应用不是常驻服务,比如是命令行工具,至少在初始化时打印明确的日志提示"嵌入模型加载中,首次查询可能稍慢",避免用户误以为程序卡死了。
6.2 语义判定的阈值问题:0.95太严,0.85太松
阈值是语义缓存里最需要调参的参数。一开始我把阈值设成0.95,结果命中率直线下降,因为用户实际对话中的复述性问法,相似度通常落在0.88到0.94之间,0.95的阈值把这些全部拦死了。后来改到0.85,命中率上来了,但又出现新问题——明显语义不同的问法也被判断为相似。
比较典型的翻车案例:用户问"帮我写一封邮件给客户",和"帮我写一封邮件给老板",这两句话的向量相似度超过0.85,系统判断为语义命中,直接返回了写给客户的邮件模板。在正式场景里这就算事故了。
最后的方法是把判断逻辑做成两级:先用0.92的阈值做低门槛召回,命中后不直接返回,而是做一次"关键实体比对",比对待办事项、人物、时间、数值这类关键信息是否一致,冲突则拒绝命中。这个思路比单靠一个阈值靠谱得多。
6.3 缓存穿透:恶意与无意的边缘请求
有一类请求永远无法命中缓存,比如带有随机数的问题:"帮我写一个带随机数字的测试数据""第138条是什么"。这些请求每次内容都不同,缓存系统每次都未命中,每次都直接打到模型上。攻击者可以靠这类请求绕过缓存,把模型API打到限流或欠费。
个人使用场景下这个问题没那么严重,但如果你要把这套系统开放给多人使用,就必须处理。常见的办法是加"布隆过滤器"拦截明显不可能命中的键,以及给单位时间内的未命中请求计数,超过阈值就自动开启限流。后来我给系统加了一层保护:同类型问题短时间内重复未命中超过5次,自动降级为纯本地模板回答,既挡住了消耗,又保证了响应速度。
6.4 内存泄漏与缓存膨胀
语义缓存有一个和精确缓存不一样的特性:精确缓存有上限、有TTL、有淘汰;语义缓存的条目是"越积越多"的,因为向量索引天然倾向于堆数据。项目跑了两周后,发现内存占用从最初的200MB涨到了接近900MB,排查下来发现是语义缓存条目只增不减,旧条目完全没有淘汰机制。
解法是加一个定期清理任务:每小时扫描一次缓存条目,超过TTL的删除;同时维护一个"最近最少使用"的访问计数,清掉最久未访问的那批条目。这里有个细节要注意:在做嵌入向量删除时,不要把向量直接留在索引里然后标个"已删除",纯标记删除会导致索引里的无效向量越来越多,相似度扫描的开销降不下来。正确做法是物理删除,或者定期重建索引。
6.5 缓存与流式输出不兼容
最后这个坑属于设计层面的:个人助手的答案原本是流式输出的,用户能实时看到每个字的生成过程,体验非常自然。但缓存系统返回的是完整答案,无法直接"流式重放"。一开始我的做法是查缓存命中后直接一次性返回全文,但用户明显觉得"回答变快了但很生硬"。
后来改成了伪流式输出:按每50毫秒一帧、每帧5到10个字的速度"播放"缓存答案。视觉上手感和真实的模型流式输出几乎没有区别,但底层完全不需要等待模型生成。这一步对整体体验的一致性很重要,个人助手类产品里不要忽略缓存场景下的交互形态变化。
7. 下一步的扩展方向:从缓存到整体的性能体系
阶段3的缓存系统虽然解决了大部分问题,但它只是个人AI助手性能体系的一部分。跑完这轮实践后,我梳理了后续值得做的几个方向,每个方向都建立在缓存基础之上。
7.1 预热与预计算
缓存只在用户发出请求之后才填充数据,意味着第一次访问某个领域的知识时仍然要等待模型推理。如果能在用户token、配置文件、知识库加载阶段提前预计算一组高频问题,把结果直接加热到语义缓存里,那相当于所有常见问题从第一次问起就都是命中状态。
以个人知识库助手为例,文档导入阶段就对每篇文档自动生成5至10个候选问题,逐个嵌入并预置到缓存中。用户导入一百篇文章,缓存里就提前存了500到1000条高频问答。这套预热逻辑在缓存命中率上的提升幅度,比继续调阈值和算法更明显。
7.2 消息历史的增量缓存
当前版本缓存的是"单轮问答对"。但实际对话是多轮连续的,用户经常在上一轮结果的基础上追问。比如先问"帮我把会议纪要整理成摘要",再问"把摘要翻译成英文",这两个单轮请求之间是有依赖关系的。
我规划的方向是把多轮对话的上下文摘要也纳入缓存:对最近几轮消息做一个摘要编码,连同本轮的问答一起存入语义缓存,匹配时把"上下文摘要加上当前问题"作为整体向量判断。这样可以覆盖不少多轮追问场景下的重复计算。实现复杂度会高一些,但对真实对话场景命中率的拉高非常明显。
7.3 分层降级策略
最后一个方向是分级响应策略的落地。不是所有问题都值得花同样的算力:简单的事实性问题可以直接走语义缓存,中等难度的可以走小模型配合检索增强,最难的那批才走完整的大模型推理。这套分级策略的底层就是缓存数据,因为缓存命中率本身就是判断问题"是否已解决过"最可靠的依据之一。
跑完阶段3之后我最大的体会是:给AI助手写缓存,本质不是数据库技术,而是行为建模技术。缓存背后记录的是"用户曾经问过什么""答案是否还有效""哪些内容值得快速重放",这套体系积累得越久,助手就越像一个有经验的工作伙伴——你说上一句,它已经知道下一句怎么接。
如果你也正在做一个个人AI助手,或者任何一个跑在大模型API之上的应用,我建议从阶段3这件事开始做性能优化动作。它可能不像换一个大模型那样能立刻提升回答质量的天花板,但它把每一分重复调用的成本都省了下来,而省下来的预算和等待时间,最终能用来换取更多真正需要模型智慧的时刻。