ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AI Agent 缓存实战:Redis 四层缓存架构与语义缓存落地

AI Agent 缓存实战:Redis 四层缓存架构与语义缓存落地 1. 为什么 AI Agent 的缓存层不能照搬传统 Web 架构做过传统 Web 后端的人第一次给 AI Agent 加缓存时脑子里浮现的往往是同一套东西请求进来先查 Redis命中就返回没命中就查数据库然后回写。这套逻辑在 CRUD 场景里跑了十几年稳得很。但把它原封不动搬到 AI Agent 上大概率会在上线后一周内出问题——不是缓存穿透就是上下文错乱要么就是账单悄悄涨了一截。原因在于 AI Agent 的请求和传统 Web 的请求本质上是两种东西。传统请求是幂等的、无状态的、结果确定的同一个用户 ID 查同一张订单今天查和明天查结果一样除非订单被改。而 AI Agent 的一次调用是有状态的、结果概率性的、成本高昂的同一个问题模型这次回答和下次回答可能不同一次调用可能消耗几千 token背后是实打实的钱更关键的是Agent 的输入往往是一整段对话历史加工具调用结果这段上下文每次都在变。所以给 AI Agent 做 Redis 缓存核心矛盾不是怎么存而是**什么能存、什么不能存、存多久、怎么判断两个请求算不算同一个**。这几个问题想不清楚缓存不但不省钱反而会制造 bug。我自己的经验是AI Agent 的缓存要分四层来看每层的键设计、过期策略、失效逻辑都不一样缓存层级缓存对象典型 TTL命中收益主要风险L1 精确响应缓存完整问答对分钟到小时省 token、降延迟上下文漂移L2 语义缓存相似问题的答案小时到天大幅提升命中率误命中、答非所问L3 工具结果缓存工具/API 返回秒到分钟减少外部调用数据过期L4 会话状态缓存对话历史、中间态会话周期降低存储压力状态不一致这张表是我踩了不少坑之后总结出来的。下面逐个拆开讲重点讲每一层在 Redis 里到底怎么落地以及那些文档里不会写的细节。2. 精确响应缓存键怎么设计决定了命中率的天花板2.1 把 prompt 直接当 key 是最常见的错误新手最容易犯的错是拿用户输入的那句话直接当 Redis key。比如用户问帮我总结这篇文章就把这句话 hash 一下存进去。问题在于Agent 的实际输入远不止这一句——它包含 system prompt、历史对话、检索到的文档片段、工具定义等等。你只拿用户那句话当 key等于把一大堆影响输出的变量全丢了命中之后返回的答案很可能和当前上下文对不上。正确的做法是对完整的、真正送进模型的输入做规范化后哈希。这里的完整输入指的是最终拼装好、准备发给 LLM 的那个 messages 数组或者 prompt 字符串。规范化包括几步去掉时间戳、去掉随机 ID、把工具调用结果的顺序固定、统一空白字符。做完这些再算 SHA-256作为 key 的主体。import hashlib import json def build_cache_key(messages, model, temperature, toolsNone): # 只保留影响输出的字段剔除时间戳、trace_id 等噪声 normalized { model: model, temperature: temperature, messages: [ {role: m[role], content: m[content]} for m in messages ], tools: sorted([t[name] for t in tools]) if tools else [], } payload json.dumps(normalized, sort_keysTrue, ensure_asciiFalse) digest hashlib.sha256(payload.encode(utf-8)).hexdigest() return fagent:resp:{model}:{digest}注意temperature必须进 key。temperature 为 0 时输出相对确定缓存价值高temperature 大于 0 时每次输出本就不同缓存的意义就变成了省一次调用而不是保证一致。我一般只对 temperature0 或极低的场景开精确缓存其他场景交给语义缓存。2.2 存什么别只存答案文本很多人缓存只存模型返回的那段文字这是浪费。一次 LLM 调用的完整返回里有价值的信息包括回答文本、finish_reason、token 用量、工具调用请求、甚至原始 response 对象。把这些一起序列化存进去命中时能直接还原成和真实调用一模一样的结构上层代码完全无感知。序列化方式我推荐 JSON 而不是 pickle。pickle 虽然方便但跨语言、跨版本、安全性都是隐患而且 Redis 里存二进制调试起来很痛苦。JSON 可读、可迁移配合zlib压缩一下体积能压掉六七成。import zlib, json def save_response(redis_client, key, response_obj, ttl3600): raw json.dumps(response_obj, ensure_asciiFalse).encode(utf-8) compressed zlib.compress(raw, level6) redis_client.setex(key, ttl, compressed) def load_response(redis_client, key): data redis_client.get(key) if data is None: return None return json.loads(zlib.decompress(data).decode(utf-8))2.3 TTL 设多久按内容时效性分档别一刀切TTL 一刀切是另一个常见问题。所有缓存都设 1 小时结果该快的没快该新的不新。我的做法是按内容类型分档事实性问答如Redis 有哪些数据类型TTL 可以到 24 小时甚至更长这类知识几乎不变。带时效的查询如今天天气当前股价TTL 压到 1 到 5 分钟甚至干脆不缓存。用户个性化内容如我的订单状态TTL 极短且 key 里必须带用户标识绝不能跨用户命中。这里有个容易忽略的点TTL 要加随机抖动。如果一批缓存同时写入、同时过期会在过期瞬间形成缓存雪崩大量请求同时打到模型延迟飙升、费用暴涨。给 TTL 加个 ±10% 的随机量就能把过期时间打散。import random def ttl_with_jitter(base_ttl): jitter int(base_ttl * 0.1) return base_ttl random.randint(-jitter, jitter)3. 语义缓存命中率翻倍但误命中是悬在头上的刀3.1 精确缓存的天花板很低精确缓存的问题在于太死。用户问Redis 怎么做分布式锁和用 Redis 实现分布式锁的方法语义完全一样但哈希完全不同精确缓存两次都命中不了。在真实 Agent 场景里用户表达同一意图的方式千变万化精确缓存的命中率往往只有个位数百分比投入产出比很低。语义缓存就是来解决这个问题的把 query 转成向量在向量库里找相似的历史 query如果相似度超过阈值就返回那条历史 query 对应的答案。命中率能从个位数拉到 30% 到 50%效果立竿见影。3.2 Redis 做语义缓存的两种落地方式Redis 本身不是向量数据库但通过 RediSearch 模块Redis Stack 自带可以做向量相似度检索。核心是建一个带向量字段的索引然后用 KNN 查询。# 建索引向量维度按你用的 embedding 模型来这里假设 1536 维 FT.CREATE idx:semantic ON HASH PREFIX 1 sem: SCHEMA query TEXT answer TEXT vec VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE写入和查询import numpy as np import redis r redis.Redis(decode_responsesFalse) def add_semantic(r, key, query, answer, embedding): vec np.array(embedding, dtypenp.float32).tobytes() r.hset(key, mapping{ query: query, answer: answer, vec: vec, }) def search_semantic(r, embedding, top_k1, threshold0.12): vec np.array(embedding, dtypenp.float32).tobytes() q ( f*[KNN {top_k} vec $vec AS score] ) res r.ft(idx:semantic).search( q, query_params{vec: vec} ) docs res.docs if not docs: return None top docs[0] score float(top.score) # COSINE 距离越小越相似 if score threshold: return None return top.answer如果你不想引入 RediSearch另一种更轻量的做法是用外部 embedding 服务算向量自己在 Redis 里存query - vector的映射然后应用层做暴力检索。数据量小几千条以内时完全够用实现简单不依赖额外模块。3.3 阈值怎么定这是语义缓存最要命的参数阈值定高了命中率上不去等于白做定低了误命中频发用户问 A 你答 B体验直接崩。我踩过的坑是一开始阈值设得很松结果如何重置密码和如何修改密码被判为相似返回了错误答案用户投诉。我的经验是分场景定阈值并且上线前一定要做人工抽检。具体做法拿一批真实 query两两算相似度人工标注哪些应该算同一个问题然后看相似度分布找一个能分开两类的临界值。COSINE 距离下我实测比较稳的区间是 0.08 到 0.15具体取决于 embedding 模型。注意语义缓存必须带否定词和数字的敏感处理。不要删除数据和删除数据向量上可能很接近但语义完全相反。我的做法是query 里出现否定词、数字、专有名词时降级到精确缓存或直接不走缓存。3.4 语义缓存的失效比精确缓存难十倍精确缓存的失效很简单删 key 就行。语义缓存的麻烦在于你很难知道哪条缓存该删。比如知识库更新了所有基于旧知识的语义缓存都该失效但你没法一条条对应。我的处理方式是给语义缓存打标签 版本号。每条缓存写入时带上知识库版本、模型版本、prompt 版本。查询时先比对版本版本不匹配直接跳过。知识库更新时只需要把版本号加一旧缓存自然全部失效不用逐条删。def add_semantic_with_version(r, key, query, answer, embedding, kb_version): vec np.array(embedding, dtypenp.float32).tobytes() r.hset(key, mapping{ query: query, answer: answer, vec: vec, kb_version: kb_version, })查询时把kb_version作为过滤条件加进 KNN 查询版本不对的直接不参与相似度计算。4. 工具结果缓存与会话状态缓存两个容易被忽视的省钱点4.1 工具调用结果缓存Agent 的隐藏成本大头一个成熟的 AI Agent 往往要调用一堆工具查数据库、调外部 API、读文件、跑代码。这些工具调用里很多是幂等且结果短期不变的。比如查汇率、查天气、查某个静态配置一分钟内调十次和调一次结果一样。如果不缓存Agent 每轮对话都重新调一遍外部 API 的配额和延迟都扛不住。工具结果缓存的 key 设计要包含工具名 规范化后的参数。参数规范化同样要处理顺序、空白、大小写。def tool_cache_key(tool_name, params): normalized json.dumps(params, sort_keysTrue, ensure_asciiFalse) digest hashlib.md5(normalized.encode()).hexdigest() return fagent:tool:{tool_name}:{digest}TTL 按工具性质定查汇率 30 秒查天气 5 分钟查静态配置 1 小时。这里有个细节工具结果缓存要能区分成功和失败。失败的调用超时、报错不要缓存否则会把一次偶发失败固化下来后续一直返回错误。4.2 会话状态缓存别把 Redis 当唯一存储Agent 的多轮对话需要保存历史。很多人图省事直接把整个对话历史塞进 Redis设个长 TTL然后就不管了。这在单机、小流量下没问题但一旦要扩容、要做持久化、要支持会话恢复就会暴露问题Redis 重启数据没了会话就断了。我的做法是Redis 只做热数据的缓存层持久化交给数据库。Redis 里存最近 N 轮的对话比如最近 20 轮TTL 设成会话空闲超时时间比如 30 分钟。超过 N 轮或超过 TTL 的历史从数据库按需加载。这样既保证了热路径快又不会因为 Redis 故障丢数据。def append_message(r, session_id, message, max_turns20, ttl1800): key fagent:session:{session_id} r.rpush(key, json.dumps(message, ensure_asciiFalse)) # 只保留最近 max_turns 轮 r.ltrim(key, -max_turns * 2, -1) r.expire(key, ttl)用 List 存对话ltrim控制长度expire控制空闲超时简单可靠。注意max_turns * 2是因为一轮对话通常包含 user 和 assistant 两条消息。4.3 会话状态的一致性并发写入是坑同一个会话如果同时有两个请求进来比如用户快速连发两条消息两个请求都去rpush顺序可能乱甚至出现消息交错。我的处理是给会话加一把 Redis 分布式锁写入时串行化。import uuid def acquire_lock(r, lock_key, ttl5000): token str(uuid.uuid4()) ok r.set(lock_key, token, nxTrue, pxttl) return token if ok else None def release_lock(r, lock_key, token): # Lua 脚本保证原子性只有 token 匹配才删 script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end r.eval(script, 1, lock_key, token)锁的 TTL 要设得比正常写入耗时长一点但也不能太长否则锁泄漏会阻塞会话。释放锁必须用 Lua 脚本比对 token防止误删别人的锁——这是分布式锁的经典坑网上教程经常漏掉。5. 缓存治理上线之后才是真正的考验5.1 监控指标不看这几个数等于裸奔缓存上线不是终点是起点。没有监控的缓存就是定时炸弹。我必看的几个指标指标含义健康区间异常时的动作命中率命中数/总请求精确5%语义30%检查 key 设计、阈值平均延迟缓存读写耗时5ms查大 key、慢查询内存使用率used_memory/maxmemory80%调 TTL、清理冷数据驱逐数evicted_keys接近 0扩容或缩短 TTL大 key 数量单 key 体积无 1MB拆分或压缩命中率是最核心的。如果精确缓存命中率长期低于 5%说明 key 设计有问题或者场景本身不适合精确缓存。语义缓存命中率低于 30%多半是阈值太严或 embedding 质量不行。5.2 大 key 和热 key两个性能杀手Agent 场景特别容易产生大 key。一次 LLM 调用的完整响应序列化后可能几百 KB一段长对话历史可能上 MB。大 key 的危害是读写阻塞、网络传输慢、删除时卡顿。我的处理原则是单 key 不超过 100KB。超了就拆响应缓存按字段拆成多个 key或者干脆压缩后存。对话历史用 List 分片每片存固定轮数。热 key 是另一个问题。某个高频问题被反复查询所有请求都打到同一个 key 上单节点压力过大。解决办法是给热 key 加随机后缀做分片读的时候随机选一个分片读。代价是写入要写多份但读性能能线性提升。import random def hot_key_shard(base_key, shards8): return f{base_key}:{random.randint(0, shards - 1)}5.3 缓存穿透和击穿Agent 场景的特殊性缓存穿透查不存在的数据在 Agent 场景里表现为用户问了一个缓存里没有、模型也答不上来的问题每次都穿透到模型。处理方式是缓存空结果用一个特殊标记存进去TTL 短一点比如 1 分钟避免反复穿透。缓存击穿热点 key 过期瞬间大量请求在 Agent 场景里更危险因为每次穿透都是一次真金白银的模型调用。处理方式是加互斥锁第一个请求发现缓存失效后先拿锁去调模型其他请求等待或返回旧值。def get_with_lock(r, key, loader, ttl3600, lock_ttl10000): cached r.get(key) if cached is not None: return json.loads(cached) lock_key f{key}:lock token acquire_lock(r, lock_key, lock_ttl) if token: try: value loader() # 真正调模型 r.setex(key, ttl, json.dumps(value, ensure_asciiFalse)) return value finally: release_lock(r, lock_key, token) else: # 没拿到锁短暂等待后重试读缓存 import time time.sleep(0.1) cached r.get(key) return json.loads(cached) if cached else loader()这段代码是缓存击穿的标准解法但有个细节要注意等待的请求如果重试还是没读到会退化成直接调模型等于锁白加了。所以等待时间要设得合理且要有兜底。6. 几个我踩过的坑和对应的处理方式6.1 序列化不一致导致的缓存有但读不出早期我用 pickle 存后来某次升级 Python 版本pickle 协议变了读旧数据直接报错。更隐蔽的是不同服务用不同的序列化方式写同一个 key读的时候互相读不懂。教训是序列化方式必须统一且写进 key 的命名规范里。我现在所有缓存 key 都带一个前缀标识序列化格式比如agent:resp:json:和agent:resp:msgpack:读的时候按前缀选解析器。6.2 上下文漂移缓存命中了但答案不对这是 AI Agent 缓存最隐蔽的坑。用户第一轮问帮我写个函数Agent 回答了第二轮用户说改成异步的如果第二轮请求的完整上下文哈希恰好和某个历史请求撞了概率低但存在就会返回完全无关的答案。更常见的是语义缓存误命中把改成异步的匹配到了改成同步的。我的防御措施有三层一是 key 里必须包含完整的对话历史哈希不能只看当前这句二是语义缓存对短 query少于 10 个字直接跳过因为短 query 歧义太大三是命中后做一次轻量的答案相关性校验用一个小模型或规则判断答案和当前 query 是否匹配不匹配就回退到真实调用。6.3 内存打满被驱逐缓存反而拖慢了系统有次线上 Redis 内存打满开始大量驱逐 key结果缓存命中率暴跌同时驱逐操作本身消耗 CPU整体延迟反而比不加缓存还高。根因是 TTL 设太长、数据只进不出。后来我加了内存水位告警80% 预警并且给不同层级的缓存设了不同的maxmemory-policy。响应缓存用allkeys-lru会话状态用volatile-lru只驱逐带 TTL 的保证关键数据不被误删。6.4 分布式锁的锁泄漏前面提过锁要用 Lua 释放但还有个坑如果业务逻辑执行时间超过了锁 TTL锁会自动过期另一个请求拿到锁此时第一个请求执行完去释放锁会把第二个请求的锁删掉。解法是给锁续期看门狗机制或者把 TTL 设得足够长并监控执行时间。我一般用前者起一个后台线程定期给还没执行完的锁续期。7. 关于选型和落地节奏的一点个人建议如果你刚开始给 Agent 加缓存我的建议是从精确响应缓存做起别一上来就搞语义缓存。精确缓存实现简单、风险低、调试容易先把这套跑通把 key 设计、序列化、TTL、监控这套基础设施搭起来。等精确缓存的命中率稳定了再叠语义缓存这时候你已经有监控和治理能力能及时发现误命中问题。Redis 版本上如果要用语义缓存直接上 Redis Stack省得自己折腾 RediSearch 模块。如果只是精确缓存和会话状态普通 Redis 7.x 完全够用。客户端我倾向用官方的 redis-py 或者 redis-rs如果你用 Rust 写 Agent连接池一定要配别每次请求新建连接。最后说个心态问题缓存不是越多越好。有些场景比如 temperature 很高的创意生成、强个性化的推荐本来就不适合缓存硬加只会增加复杂度和出错概率。判断标准很简单——如果缓存命中返回的答案和真实调用返回的答案用户能感知到差异那这个缓存就不该加。想清楚这一点比学任何技巧都重要。
返回列表