1. 为什么 AI Agent 一上缓存就翻车做过 AI Agent 项目的人大概都有过这种体验本地跑得好好的一上生产环境用户量稍微起来一点响应就开始飘Redis 的command timed out报错一条接一条往外冒日志里全是io.lettuce.core.RedisCommandTimeoutException。这不是个例而是 AI Agent 和传统 CRUD 应用在缓存这件事上最本质的差异导致的。传统 Web 应用的缓存模式很单纯一个请求进来查一次数据库把结果塞进 Redis下次同样的请求直接命中缓存返回。整个链路的耗时是可控的缓存 key 的粒度也是清晰的。但 AI Agent 完全不是这个逻辑。一个 Agent 处理一次用户输入内部可能触发多轮 LLM 调用、工具调用、记忆检索、状态更新每一轮都可能产生需要缓存的数据而且这些数据的生命周期、更新频率、一致性要求各不相同。你如果还用传统那套一个 key 存一个结果的思路去做缓存不但帮不上忙反而会成为系统里最大的不稳定因素。我接手过一个基于 Spring AI Agent 的客服项目上线第一周 Redis 内存直接打满排查下来发现是 Agent 的对话记忆被无差别地全量缓存每个会话的完整上下文都往 Redis 里塞一个活跃用户一天能产生几十 MB 的缓存数据。更麻烦的是这些缓存没有合理的过期策略也没有做序列化优化JDK 默认序列化出来的字节数组比原始数据大了将近三倍。这就是典型的把缓存当数据库用的翻车现场。所以这篇内容我想聊的不是怎么在 AI Agent 里加 Redis这种入门话题而是从架构设计、数据类型选型、并发控制、缓存治理这几个维度把 AI Agent 场景下 Redis 缓存的那些坑和对应的解法讲透。适合正在做 AI Agent 开发、或者准备把 Agent 项目往生产环境推的工程师参考不管你是用 Python 的 LangChain 系还是 Java 的 Spring AI 系底层的缓存逻辑是相通的。2. AI Agent 缓存架构的整体设计思路2.1 Agent 场景下缓存的三个层次在动手写代码之前得先把 AI Agent 里什么该缓存、什么不该缓存这件事想清楚。我的经验是把缓存需求拆成三个层次来看每个层次的策略完全不同。第一个层次是模型调用结果缓存。同一个 prompt 在短时间内被多次请求结果大概率是一样的尤其是那些系统提示词固定、用户输入重复率高的场景。这类缓存的价值最高因为一次 LLM 调用的成本时间和费用远高于一次 Redis 读写。但要注意这类缓存的 key 必须包含完整的 prompt 内容加上模型参数任何一项不同都不能命中。第二个层次是工具调用结果缓存。Agent 调用外部 API 或者查询数据库的结果如果数据本身变化不频繁完全可以缓存。比如查天气、查汇率、查商品库存这类操作缓存几分钟到几十分钟都是合理的。这类缓存的关键是过期时间要跟数据的实际变化频率匹配不能拍脑袋定。第三个层次是会话状态和记忆缓存。这是最容易出问题的部分。Agent 的多轮对话需要维护上下文但上下文不能无限制地往 Redis 里堆。我的做法是只缓存最近 N 轮对话的摘要而不是完整原文同时给每个会话设置硬性的 TTL比如 30 分钟不活跃就自动清理。注意不要试图用 Redis 做 Agent 的长期记忆存储。Redis 是缓存不是数据库。需要持久化的记忆应该落到向量数据库或者关系型数据库里Redis 只负责热数据的快速访问。2.2 为什么选 Redis 而不是本地缓存有人会问Agent 服务如果是单机部署用 Caffeine 或者 Guava 这种本地缓存不是更快吗确实快但问题在于 AI Agent 服务几乎不可能是单机的。原因有两个一是 Agent 的推理过程耗时长单机扛不住并发二是 Agent 的状态需要在多个实例之间共享否则用户第一次请求打到 A 实例、第二次打到 B 实例上下文就断了。Redis 作为分布式缓存天然解决了状态共享的问题。而且 Redis 的数据类型丰富除了最简单的 String还有 Hash、List、Set、Sorted Set 等结构这些在 Agent 场景下都有对应的用法。比如用 Hash 存会话的各个字段用 List 存对话历史用 Sorted Set 做带时间戳的记忆检索。这些是本地缓存做不到的。当然Redis 也不是没有代价。网络往返的延迟、序列化的开销、连接池的管理这些都是要额外处理的。但在 Agent 场景下一次 LLM 调用动辄几秒Redis 那几毫秒的延迟完全可以忽略。真正需要担心的是并发量上来之后 Redis 本身的压力这个后面会专门讲。2.3 缓存粒度怎么定缓存粒度是设计阶段最容易做错的决定。粒度太粗一个 key 存一大坨数据更新的时候要整体重写并发冲突严重粒度太细key 的数量爆炸Redis 的内存碎片和管理成本都上去了。我的经验是按最小可复用单元来定粒度。举个例子Agent 的系统提示词、工具定义、模型配置这些在一次部署周期内是不变的可以整体缓存成一个 key。而用户的会话上下文每个会话一个 key用 Hash 结构存各个字段更新哪个字段就改哪个字段不用整体重写。至于单次的模型调用结果每个 prompt 的哈希值一个 key这是最细的粒度。这里有个实操技巧key 的命名一定要有清晰的层级结构比如agent:session:{sessionId}:context、agent:llm:{promptHash}、agent:tool:{toolName}:{paramHash}。这样在排查问题和做批量清理的时候可以用SCAN命令按前缀匹配比KEYS安全得多。3. Redis 数据类型在 Agent 场景下的选型实战3.1 String 类型模型调用结果缓存的主力String 是 Redis 最基础也最常用的类型在 Agent 场景下主要用来缓存 LLM 的调用结果。用法很直接把 prompt 和模型参数拼起来算一个哈希值作为 key把模型返回的内容作为 value 存进去设置一个合理的 TTL。但这里有几个细节值得展开说。首先是序列化的选择。Java 项目里很多人习惯用 JDK 默认序列化但那个东西又慢又占空间。我一般用 Jackson 或者 Protobuf序列化后的体积能小一半以上。Python 项目里用pickle也行但要注意版本兼容性跨语言场景下还是 JSON 最稳妥。其次是 TTL 的设置。模型调用结果的缓存时间不能太长因为模型本身可能会更新prompt 模板也可能调整。我的做法是设一个相对短的 TTL比如 10 到 30 分钟同时配合一个版本号在 key 里模型或者 prompt 一变版本号一升旧缓存自然就失效了。import hashlib import json import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def cache_llm_result(prompt: str, model: str, params: dict, result: str, ttl: int 1800): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} r.setex(key, ttl, result) return key def get_cached_llm_result(prompt: str, model: str, params: dict): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} return r.get(key)这段代码里有个细节json.dumps的时候加了sort_keysTrue这是为了保证同样的参数字典不管键的顺序如何算出来的哈希值都一样。不加这个参数的话{a:1,b:2}和{b:2,a:1}会算出两个不同的 key缓存命中率直接腰斩。3.2 Hash 类型会话状态的结构化存储会话状态用 Hash 存是最合适的。一个会话一个 key会话里的各个字段用户 ID、当前意图、已收集的槽位、最近一轮的回复等作为 Hash 的 field。这样更新某个字段的时候只需要HSET一个 field不用把整个会话读出来改完再写回去并发安全性好很多。def update_session_field(session_id: str, field: str, value: str): key fagent:session:{session_id} r.hset(key, field, value) r.expire(key, 1800) # 每次更新都续期保证活跃会话不过期 def get_session(session_id: str) - dict: key fagent:session:{session_id} return r.hgetall(key)这里有个坑要注意HSET之后一定要记得EXPIRE。Redis 的 Hash 结构本身不支持在创建时直接设 TTL必须单独调EXPIRE。我见过有人只HSET不EXPIRE结果会话数据越积越多最后内存爆掉。每次更新都续期这个策略也要想清楚如果你的业务希望会话在固定时间后强制过期那就不能每次更新都续期而是只在创建时设一次。3.3 List 类型对话历史的滑动窗口对话历史用 List 存配合LPUSH和LTRIM实现滑动窗口。每来一轮新对话就LPUSH进去然后LTRIM只保留最近 N 条。这样不管对话多长Redis 里存的永远只有最近 N 轮内存占用是可控的。def append_dialogue(session_id: str, role: str, content: str, max_turns: int 20): key fagent:dialogue:{session_id} r.lpush(key, json.dumps({role: role, content: content})) r.ltrim(key, 0, max_turns - 1) r.expire(key, 3600) def get_recent_dialogue(session_id: str, count: int 10) - list: key fagent:dialogue:{session_id} items r.lrange(key, 0, count - 1) return [json.loads(i) for i in items]max_turns这个值怎么定取决于你的模型上下文窗口大小和每轮对话的平均 token 数。假设模型支持 8K token 的上下文系统提示词占了 1K每轮对话平均 200 token那理论上能放 35 轮。但实际用的时候要留余量我一般设 15 到 20 轮超出的部分做摘要压缩后再存。3.4 Sorted Set 类型带权重的记忆检索Sorted Set 在 Agent 场景下的一个高级用法是做记忆的优先级排序。每条记忆的 score 可以是时间戳、重要程度、或者两者的加权组合。检索的时候用ZREVRANGEBYSCORE按 score 范围取比全量扫描高效得多。import time def add_memory(session_id: str, memory_id: str, content: str, importance: float): key fagent:memory:{session_id} score time.time() * importance # 时间戳乘以重要度作为权重 r.zadd(key, {memory_id: score}) r.hset(fagent:memory:content:{session_id}, memory_id, content) r.expire(key, 86400) r.expire(fagent:memory:content:{session_id}, 86400) def get_top_memories(session_id: str, count: int 5) - list: key fagent:memory:{session_id} memory_ids r.zrevrange(key, 0, count - 1) contents r.hmget(fagent:memory:content:{session_id}, memory_ids) return list(zip(memory_ids, contents))这个方案的好处是把哪些记忆更重要这个判断交给了 score检索的时候直接取 top N不用把所有记忆都拉出来再排序。重要度的计算可以结合访问频率、用户反馈、时间衰减等因素这块就是业务逻辑了。3.5 数据类型选型对照表数据类型Agent 场景用途关键优势注意事项String模型调用结果缓存简单直接读写最快序列化方式影响体积和速度Hash会话状态存储支持字段级更新并发友好必须单独设 TTLList对话历史天然有序滑动窗口方便注意 LTRIM 控制长度Sorted Set记忆优先级排序按 score 检索高效score 设计要合理Set去重场景如已用工具集自动去重无序不适合需要顺序的场景4. 并发场景下的缓存一致性保障4.1 AI Agent 为什么更容易出现缓存击穿缓存击穿指的是某个热点 key 过期的一瞬间大量请求同时打到后端。传统应用里这个问题也存在但 AI Agent 场景下它更严重原因在于 Agent 的一次请求链路很长从缓存失效到后端重新计算完成中间可能隔了好几秒。这几秒里如果又来了几十个同样的请求它们全都会去调 LLM费用和延迟都爆炸。更麻烦的是 Agent 的缓存 key 往往跟用户输入强相关热门问题的 key 集中度很高。比如一个客服 Agent用户问得最多的就那么几个问题这些 key 一旦过期瞬间就会被大量请求同时命中。4.2 用分布式锁做缓存重建解决缓存击穿的经典方案是加锁只让一个请求去重建缓存其他请求等待。Redis 的SET NX EX命令可以实现一个简单的分布式锁。import uuid def get_or_rebuild_llm_cache(prompt: str, model: str, params: dict, rebuild_func, ttl: int 1800): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} cache_key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} lock_key fagent:lock:{hashlib.sha256(raw.encode()).hexdigest()} cached r.get(cache_key) if cached: return cached lock_value str(uuid.uuid4()) acquired r.set(lock_key, lock_value, nxTrue, ex10) if acquired: try: # 双重检查防止在等锁期间已经有其他请求重建好了 cached r.get(cache_key) if cached: return cached result rebuild_func(prompt, model, params) r.setex(cache_key, ttl, result) return result finally: # 用 Lua 脚本保证只释放自己的锁 unlock_script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end r.eval(unlock_script, 1, lock_key, lock_value) else: # 没抢到锁短暂等待后重试读缓存 time.sleep(0.1) cached r.get(cache_key) if cached: return cached # 兜底直接走重建避免无限等待 return rebuild_func(prompt, model, params)这段代码里有几个关键点。第一是双重检查抢到锁之后要再查一次缓存因为可能在等锁的过程中已经有别的请求把缓存建好了。第二是释放锁必须用 Lua 脚本保证判断是不是自己的锁和删除锁这两个操作是原子的否则可能出现误删别人锁的情况。第三是没抢到锁的请求不能无限等待要有兜底逻辑否则一旦持锁的请求挂了其他请求全卡死。注意锁的过期时间要设得比缓存重建的最长时间略长一点。LLM 调用如果超时是 30 秒锁的过期时间至少设 35 秒。但也不能太长否则持锁进程崩溃后其他请求要等很久才能拿到锁。4.3 缓存更新时的并发问题缓存更新比缓存重建更复杂因为涉及到先更新数据库还是先更新缓存这个经典问题。在 Agent 场景下需要更新的通常是会话状态。我的做法是直接更新 Redis不搞延迟双删那一套因为 Agent 的会话状态本身就是以 Redis 为准的不存在数据库和缓存不一致的问题。但如果你的 Agent 有持久化的会话存储比如 MySQL那就得考虑一致性了。我一般用先更新数据库再删除缓存的策略而不是更新缓存。删除缓存比更新缓存安全因为删除操作是幂等的而且下次读取时会自然地从数据库加载最新数据。def update_session_with_persistence(session_id: str, field: str, value: str): # 先更新数据库 db.update_session(session_id, field, value) # 再删除缓存而不是更新缓存 r.delete(fagent:session:{session_id})这个策略在绝大多数场景下够用了。极端情况下读请求在删缓存之前拿到旧数据然后写请求删缓存读请求再把旧数据写回缓存确实会有一致性问题但概率极低而且 Agent 场景对会话状态的实时一致性要求没那么高可以接受。4.4 热点 key 的分散策略如果某个 key 的访问量特别大单个 Redis 节点的压力会很大。这时候可以在 key 后面加随机后缀把同一个逻辑 key 分散到多个物理 key 上。import random def get_sharded_cache(base_key: str, shard_count: int 10): shard random.randint(0, shard_count - 1) return r.get(f{base_key}:shard:{shard}) def set_sharded_cache(base_key: str, value: str, ttl: int, shard_count: int 10): for i in range(shard_count): r.setex(f{base_key}:shard:{i}, ttl, value)这个方案的代价是写入时要写多份读取时随机读一份。适合读多写极少的热点数据比如 Agent 的系统提示词、工具定义这些。写多读少的数据不适合用这个方案。5. 缓存治理从能用到好用5.1 内存监控与淘汰策略Redis 的内存是有限的Agent 缓存如果不加治理迟早会把内存吃满。首先要做的是设置合理的内存上限和淘汰策略。在redis.conf里配置maxmemory 4gb maxmemory-policy allkeys-lruallkeys-lru表示当内存达到上限时优先淘汰最近最少使用的 key。这个策略适合缓存场景因为缓存数据丢了可以重建。但要注意如果你的 Redis 里同时存了锁、计数器这类不能丢的数据就不能用allkeys-lru得用volatile-lru只淘汰设置了过期时间的 key。监控方面INFO memory命令能看到当前内存使用情况INFO stats能看到命中率。命中率低于 80% 就说明缓存策略有问题要么是 key 设计不合理要么是 TTL 设得太短。redis-cli INFO memory | grep used_memory_human redis-cli INFO stats | grep keyspace_hits redis-cli INFO stats | grep keyspace_misses5.2 大 key 和热 key 的排查大 key 是 Redis 性能的隐形杀手。一个几 MB 的 Hash 或者 List每次操作都要传输大量数据网络带宽和序列化开销都很可观。Agent 场景下最容易出现大 key 的地方是对话历史如果不做LTRIM一个活跃会话的 List 能长到几万条。排查大 key 用redis-cli --bigkeys它会扫描所有 key 并报告每种类型中最大的几个。热 key 的排查稍微麻烦一点可以用redis-cli --hotkeys需要 Redis 4.0 以上并且开启了 LFU 淘汰策略或者用MONITOR命令采样一段时间看哪些 key 出现频率最高。redis-cli --bigkeys redis-cli --hotkeys发现大 key 之后的处理方式如果是 List 或 Hash用LTRIM或者HDEL分批删除不要用DEL一次性删因为DEL大 key 会阻塞 Redis 主线程。Redis 4.0 之后可以用UNLINK命令异步删除不阻塞主线程。5.3 缓存预热与降级Agent 服务重启后缓存是空的这时候所有请求都会穿透到后端。如果 QPS 高后端可能直接被压垮。解决办法是缓存预热服务启动时主动把热点数据加载到 Redis 里。def warmup_cache(): hot_prompts load_hot_prompts_from_db() for prompt in hot_prompts: cache_llm_result(prompt[text], prompt[model], prompt[params], prompt[result])预热的数据来源可以是历史访问日志里出现频率最高的 prompt也可以是运营配置的固定问题集。预热的时机最好在服务正式接流量之前比如在健康检查通过之后、注册到负载均衡之前。降级策略是另一道保险。当 Redis 不可用时Agent 不能直接挂掉而是应该降级到直接调用后端。实现方式是在缓存访问层加 try-catchRedis 异常时记录日志并走 fallback 逻辑。def safe_get_cache(key: str): try: return r.get(key) except redis.RedisError as e: logger.warning(fRedis unavailable, fallback to backend: {e}) return None5.4 缓存 key 的规范与清理key 的命名规范前面提过了这里补充一下清理策略。Agent 的缓存 key 数量会随着用户量增长而增长必须有一套自动清理机制。除了 TTL 自动过期还需要定期清理那些僵尸 key——已经不再被访问但还没到过期时间的 key。我的做法是给每个 key 在写入时记录一个最后访问时间的 field如果是 Hash 结构然后跑一个定时任务扫描超过一定时间没被访问的 key 并删除。或者更简单一点直接用较短的 TTL让 Redis 自己淘汰。注意不要用KEYS *命令做批量清理这个命令会阻塞 Redis 主线程生产环境绝对不能用。要用SCAN命令游标遍历每次只取一小批。def clean_expired_session_keys(pattern: str agent:session:*, batch_size: int 100): cursor 0 while True: cursor, keys r.scan(cursor, matchpattern, countbatch_size) for key in keys: if r.ttl(key) -1: # 没有设置过期时间的 key r.expire(key, 1800) # 补设 TTL if cursor 0: break6. 常见问题与排查技巧实录6.1 RedisCommandTimeoutException 的排查思路这个报错在 Agent 项目里出现频率极高但原因可能有好几种。我的排查顺序是这样的先看是不是大 key 导致的。用--bigkeys扫一遍如果有超过 1MB 的 key基本就是它了。大 key 的读写会占用大量网络带宽导致后续命令排队超时。再看是不是慢查询。SLOWLOG GET 10能看到最近 10 条慢查询如果某条命令执行时间超过 10ms就要关注了。Agent 场景下常见的慢查询是KEYS、HGETALL大 Hash、SMEMBERS大 Set。最后看连接池配置。Lettuce 默认是单连接多路复用如果并发高单连接会成为瓶颈。可以改成连接池模式配置spring.redis.lettuce.pool.max-active等参数。排查项命令判断标准处理方式大 keyredis-cli --bigkeys单 key 超过 1MB拆分或分批删除慢查询SLOWLOG GET 10单命令超过 10ms优化命令或数据结构连接池INFO clients连接数接近上限调大连接池或加节点内存INFO memory使用率超过 80%扩容或调整淘汰策略网络redis-cli --latency延迟超过 1ms检查网络或部署位置6.2 缓存穿透与空值缓存缓存穿透是指查询一个不存在的 key缓存里没有每次都打到后端。Agent 场景下如果用户输入了一个从没出现过的问题就会触发穿透。解决办法是缓存空值即使后端返回空也往 Redis 里存一个特殊标记TTL 设短一点比如 60 秒。def get_with_null_cache(key: str, loader_func, null_ttl: int 60): cached r.get(key) if cached is not None: return None if cached __NULL__ else cached result loader_func() if result is None: r.setex(key, null_ttl, __NULL__) else: r.setex(key, 1800, result) return result6.3 序列化兼容性问题Java 项目里用 JDK 序列化的时候如果实体类改了字段反序列化旧数据会报InvalidClassException。解决办法是显式声明serialVersionUID或者干脆换成 JSON 序列化。Spring Data Redis 里配置 JSON 序列化Bean public RedisTemplateString, Object redisTemplate(RedisConnectionFactory factory) { RedisTemplateString, Object template new RedisTemplate(); template.setConnectionFactory(factory); Jackson2JsonRedisSerializerObject serializer new Jackson2JsonRedisSerializer(Object.class); ObjectMapper mapper new ObjectMapper(); mapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY); serializer.setObjectMapper(mapper); template.setKeySerializer(new StringRedisSerializer()); template.setValueSerializer(serializer); template.setHashKeySerializer(new StringRedisSerializer()); template.setHashValueSerializer(serializer); template.afterPropertiesSet(); return template; }6.4 分布式锁的常见坑分布式锁在 Agent 场景下用得很多但坑也不少。最常见的是锁过期了业务还没执行完导致两个请求同时持有锁。解决办法是加看门狗机制定期给锁续期。Redisson 已经内置了这个机制直接用RLock就行。RLock lock redissonClient.getLock(agent:lock: key); try { if (lock.tryLock(3, 30, TimeUnit.SECONDS)) { // 业务逻辑 } } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } }另一个坑是锁的粒度。锁的粒度太粗所有请求都串行化性能差粒度太细锁的数量爆炸管理成本高。我的经验是锁的粒度跟缓存 key 的粒度保持一致一个缓存 key 对应一把锁。6.5 缓存与数据库双写一致性速查场景策略一致性性能适用性先更新 DB 再删缓存推荐最终一致好绝大多数场景先删缓存再更新 DB不推荐差好几乎不用延迟双删可用较好中一致性要求高时更新 DB 同时更新缓存不推荐差好并发低时可用订阅 binlog 异步删缓存复杂最终一致好大型系统7. 一些踩坑之后的经验之谈做 AI Agent 的 Redis 缓存我最大的体会是别把缓存当数据库用。Agent 的数据天然是多样化的有临时的、有持久的、有热的有冷的一股脑全塞 Redis 里最后一定是内存爆掉、性能下降。该分层的分层该落库的落库Redis 只负责它最擅长的那部分。另一个体会是 TTL 一定要设而且要根据数据的实际生命周期来设。我见过太多项目因为没设 TTL 导致 Redis 内存无限增长最后不得不半夜起来手动清理。宁可 TTL 设短一点缓存命中率低一点也不要让内存失控。最后说一个实操小技巧在 Agent 的缓存 key 里加一个环境前缀比如prod:agent:llm:...和dev:agent:llm:...。这样多个环境共用一个 Redis 实例的时候不会互相干扰排查问题的时候也能快速区分。这个习惯看起来不起眼但在多环境部署的项目里能省很多事。
1. 为什么 AI Agent 一上缓存就翻车做过 AI Agent 项目的人大概都有过这种体验本地跑得好好的一上生产环境用户量稍微起来一点响应就开始飘Redis 的command timed out报错一条接一条往外冒日志里全是io.lettuce.core.RedisCommandTimeoutException。这不是个例而是 AI Agent 和传统 CRUD 应用在缓存这件事上最本质的差异导致的。传统 Web 应用的缓存模式很单纯一个请求进来查一次数据库把结果塞进 Redis下次同样的请求直接命中缓存返回。整个链路的耗时是可控的缓存 key 的粒度也是清晰的。但 AI Agent 完全不是这个逻辑。一个 Agent 处理一次用户输入内部可能触发多轮 LLM 调用、工具调用、记忆检索、状态更新每一轮都可能产生需要缓存的数据而且这些数据的生命周期、更新频率、一致性要求各不相同。你如果还用传统那套一个 key 存一个结果的思路去做缓存不但帮不上忙反而会成为系统里最大的不稳定因素。我接手过一个基于 Spring AI Agent 的客服项目上线第一周 Redis 内存直接打满排查下来发现是 Agent 的对话记忆被无差别地全量缓存每个会话的完整上下文都往 Redis 里塞一个活跃用户一天能产生几十 MB 的缓存数据。更麻烦的是这些缓存没有合理的过期策略也没有做序列化优化JDK 默认序列化出来的字节数组比原始数据大了将近三倍。这就是典型的把缓存当数据库用的翻车现场。所以这篇内容我想聊的不是怎么在 AI Agent 里加 Redis这种入门话题而是从架构设计、数据类型选型、并发控制、缓存治理这几个维度把 AI Agent 场景下 Redis 缓存的那些坑和对应的解法讲透。适合正在做 AI Agent 开发、或者准备把 Agent 项目往生产环境推的工程师参考不管你是用 Python 的 LangChain 系还是 Java 的 Spring AI 系底层的缓存逻辑是相通的。2. AI Agent 缓存架构的整体设计思路2.1 Agent 场景下缓存的三个层次在动手写代码之前得先把 AI Agent 里什么该缓存、什么不该缓存这件事想清楚。我的经验是把缓存需求拆成三个层次来看每个层次的策略完全不同。第一个层次是模型调用结果缓存。同一个 prompt 在短时间内被多次请求结果大概率是一样的尤其是那些系统提示词固定、用户输入重复率高的场景。这类缓存的价值最高因为一次 LLM 调用的成本时间和费用远高于一次 Redis 读写。但要注意这类缓存的 key 必须包含完整的 prompt 内容加上模型参数任何一项不同都不能命中。第二个层次是工具调用结果缓存。Agent 调用外部 API 或者查询数据库的结果如果数据本身变化不频繁完全可以缓存。比如查天气、查汇率、查商品库存这类操作缓存几分钟到几十分钟都是合理的。这类缓存的关键是过期时间要跟数据的实际变化频率匹配不能拍脑袋定。第三个层次是会话状态和记忆缓存。这是最容易出问题的部分。Agent 的多轮对话需要维护上下文但上下文不能无限制地往 Redis 里堆。我的做法是只缓存最近 N 轮对话的摘要而不是完整原文同时给每个会话设置硬性的 TTL比如 30 分钟不活跃就自动清理。注意不要试图用 Redis 做 Agent 的长期记忆存储。Redis 是缓存不是数据库。需要持久化的记忆应该落到向量数据库或者关系型数据库里Redis 只负责热数据的快速访问。2.2 为什么选 Redis 而不是本地缓存有人会问Agent 服务如果是单机部署用 Caffeine 或者 Guava 这种本地缓存不是更快吗确实快但问题在于 AI Agent 服务几乎不可能是单机的。原因有两个一是 Agent 的推理过程耗时长单机扛不住并发二是 Agent 的状态需要在多个实例之间共享否则用户第一次请求打到 A 实例、第二次打到 B 实例上下文就断了。Redis 作为分布式缓存天然解决了状态共享的问题。而且 Redis 的数据类型丰富除了最简单的 String还有 Hash、List、Set、Sorted Set 等结构这些在 Agent 场景下都有对应的用法。比如用 Hash 存会话的各个字段用 List 存对话历史用 Sorted Set 做带时间戳的记忆检索。这些是本地缓存做不到的。当然Redis 也不是没有代价。网络往返的延迟、序列化的开销、连接池的管理这些都是要额外处理的。但在 Agent 场景下一次 LLM 调用动辄几秒Redis 那几毫秒的延迟完全可以忽略。真正需要担心的是并发量上来之后 Redis 本身的压力这个后面会专门讲。2.3 缓存粒度怎么定缓存粒度是设计阶段最容易做错的决定。粒度太粗一个 key 存一大坨数据更新的时候要整体重写并发冲突严重粒度太细key 的数量爆炸Redis 的内存碎片和管理成本都上去了。我的经验是按最小可复用单元来定粒度。举个例子Agent 的系统提示词、工具定义、模型配置这些在一次部署周期内是不变的可以整体缓存成一个 key。而用户的会话上下文每个会话一个 key用 Hash 结构存各个字段更新哪个字段就改哪个字段不用整体重写。至于单次的模型调用结果每个 prompt 的哈希值一个 key这是最细的粒度。这里有个实操技巧key 的命名一定要有清晰的层级结构比如agent:session:{sessionId}:context、agent:llm:{promptHash}、agent:tool:{toolName}:{paramHash}。这样在排查问题和做批量清理的时候可以用SCAN命令按前缀匹配比KEYS安全得多。3. Redis 数据类型在 Agent 场景下的选型实战3.1 String 类型模型调用结果缓存的主力String 是 Redis 最基础也最常用的类型在 Agent 场景下主要用来缓存 LLM 的调用结果。用法很直接把 prompt 和模型参数拼起来算一个哈希值作为 key把模型返回的内容作为 value 存进去设置一个合理的 TTL。但这里有几个细节值得展开说。首先是序列化的选择。Java 项目里很多人习惯用 JDK 默认序列化但那个东西又慢又占空间。我一般用 Jackson 或者 Protobuf序列化后的体积能小一半以上。Python 项目里用pickle也行但要注意版本兼容性跨语言场景下还是 JSON 最稳妥。其次是 TTL 的设置。模型调用结果的缓存时间不能太长因为模型本身可能会更新prompt 模板也可能调整。我的做法是设一个相对短的 TTL比如 10 到 30 分钟同时配合一个版本号在 key 里模型或者 prompt 一变版本号一升旧缓存自然就失效了。import hashlib import json import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def cache_llm_result(prompt: str, model: str, params: dict, result: str, ttl: int 1800): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} r.setex(key, ttl, result) return key def get_cached_llm_result(prompt: str, model: str, params: dict): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} return r.get(key)这段代码里有个细节json.dumps的时候加了sort_keysTrue这是为了保证同样的参数字典不管键的顺序如何算出来的哈希值都一样。不加这个参数的话{a:1,b:2}和{b:2,a:1}会算出两个不同的 key缓存命中率直接腰斩。3.2 Hash 类型会话状态的结构化存储会话状态用 Hash 存是最合适的。一个会话一个 key会话里的各个字段用户 ID、当前意图、已收集的槽位、最近一轮的回复等作为 Hash 的 field。这样更新某个字段的时候只需要HSET一个 field不用把整个会话读出来改完再写回去并发安全性好很多。def update_session_field(session_id: str, field: str, value: str): key fagent:session:{session_id} r.hset(key, field, value) r.expire(key, 1800) # 每次更新都续期保证活跃会话不过期 def get_session(session_id: str) - dict: key fagent:session:{session_id} return r.hgetall(key)这里有个坑要注意HSET之后一定要记得EXPIRE。Redis 的 Hash 结构本身不支持在创建时直接设 TTL必须单独调EXPIRE。我见过有人只HSET不EXPIRE结果会话数据越积越多最后内存爆掉。每次更新都续期这个策略也要想清楚如果你的业务希望会话在固定时间后强制过期那就不能每次更新都续期而是只在创建时设一次。3.3 List 类型对话历史的滑动窗口对话历史用 List 存配合LPUSH和LTRIM实现滑动窗口。每来一轮新对话就LPUSH进去然后LTRIM只保留最近 N 条。这样不管对话多长Redis 里存的永远只有最近 N 轮内存占用是可控的。def append_dialogue(session_id: str, role: str, content: str, max_turns: int 20): key fagent:dialogue:{session_id} r.lpush(key, json.dumps({role: role, content: content})) r.ltrim(key, 0, max_turns - 1) r.expire(key, 3600) def get_recent_dialogue(session_id: str, count: int 10) - list: key fagent:dialogue:{session_id} items r.lrange(key, 0, count - 1) return [json.loads(i) for i in items]max_turns这个值怎么定取决于你的模型上下文窗口大小和每轮对话的平均 token 数。假设模型支持 8K token 的上下文系统提示词占了 1K每轮对话平均 200 token那理论上能放 35 轮。但实际用的时候要留余量我一般设 15 到 20 轮超出的部分做摘要压缩后再存。3.4 Sorted Set 类型带权重的记忆检索Sorted Set 在 Agent 场景下的一个高级用法是做记忆的优先级排序。每条记忆的 score 可以是时间戳、重要程度、或者两者的加权组合。检索的时候用ZREVRANGEBYSCORE按 score 范围取比全量扫描高效得多。import time def add_memory(session_id: str, memory_id: str, content: str, importance: float): key fagent:memory:{session_id} score time.time() * importance # 时间戳乘以重要度作为权重 r.zadd(key, {memory_id: score}) r.hset(fagent:memory:content:{session_id}, memory_id, content) r.expire(key, 86400) r.expire(fagent:memory:content:{session_id}, 86400) def get_top_memories(session_id: str, count: int 5) - list: key fagent:memory:{session_id} memory_ids r.zrevrange(key, 0, count - 1) contents r.hmget(fagent:memory:content:{session_id}, memory_ids) return list(zip(memory_ids, contents))这个方案的好处是把哪些记忆更重要这个判断交给了 score检索的时候直接取 top N不用把所有记忆都拉出来再排序。重要度的计算可以结合访问频率、用户反馈、时间衰减等因素这块就是业务逻辑了。3.5 数据类型选型对照表数据类型Agent 场景用途关键优势注意事项String模型调用结果缓存简单直接读写最快序列化方式影响体积和速度Hash会话状态存储支持字段级更新并发友好必须单独设 TTLList对话历史天然有序滑动窗口方便注意 LTRIM 控制长度Sorted Set记忆优先级排序按 score 检索高效score 设计要合理Set去重场景如已用工具集自动去重无序不适合需要顺序的场景4. 并发场景下的缓存一致性保障4.1 AI Agent 为什么更容易出现缓存击穿缓存击穿指的是某个热点 key 过期的一瞬间大量请求同时打到后端。传统应用里这个问题也存在但 AI Agent 场景下它更严重原因在于 Agent 的一次请求链路很长从缓存失效到后端重新计算完成中间可能隔了好几秒。这几秒里如果又来了几十个同样的请求它们全都会去调 LLM费用和延迟都爆炸。更麻烦的是 Agent 的缓存 key 往往跟用户输入强相关热门问题的 key 集中度很高。比如一个客服 Agent用户问得最多的就那么几个问题这些 key 一旦过期瞬间就会被大量请求同时命中。4.2 用分布式锁做缓存重建解决缓存击穿的经典方案是加锁只让一个请求去重建缓存其他请求等待。Redis 的SET NX EX命令可以实现一个简单的分布式锁。import uuid def get_or_rebuild_llm_cache(prompt: str, model: str, params: dict, rebuild_func, ttl: int 1800): raw f{prompt}|{model}|{json.dumps(params, sort_keysTrue)} cache_key fagent:llm:v1:{hashlib.sha256(raw.encode()).hexdigest()} lock_key fagent:lock:{hashlib.sha256(raw.encode()).hexdigest()} cached r.get(cache_key) if cached: return cached lock_value str(uuid.uuid4()) acquired r.set(lock_key, lock_value, nxTrue, ex10) if acquired: try: # 双重检查防止在等锁期间已经有其他请求重建好了 cached r.get(cache_key) if cached: return cached result rebuild_func(prompt, model, params) r.setex(cache_key, ttl, result) return result finally: # 用 Lua 脚本保证只释放自己的锁 unlock_script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end r.eval(unlock_script, 1, lock_key, lock_value) else: # 没抢到锁短暂等待后重试读缓存 time.sleep(0.1) cached r.get(cache_key) if cached: return cached # 兜底直接走重建避免无限等待 return rebuild_func(prompt, model, params)这段代码里有几个关键点。第一是双重检查抢到锁之后要再查一次缓存因为可能在等锁的过程中已经有别的请求把缓存建好了。第二是释放锁必须用 Lua 脚本保证判断是不是自己的锁和删除锁这两个操作是原子的否则可能出现误删别人锁的情况。第三是没抢到锁的请求不能无限等待要有兜底逻辑否则一旦持锁的请求挂了其他请求全卡死。注意锁的过期时间要设得比缓存重建的最长时间略长一点。LLM 调用如果超时是 30 秒锁的过期时间至少设 35 秒。但也不能太长否则持锁进程崩溃后其他请求要等很久才能拿到锁。4.3 缓存更新时的并发问题缓存更新比缓存重建更复杂因为涉及到先更新数据库还是先更新缓存这个经典问题。在 Agent 场景下需要更新的通常是会话状态。我的做法是直接更新 Redis不搞延迟双删那一套因为 Agent 的会话状态本身就是以 Redis 为准的不存在数据库和缓存不一致的问题。但如果你的 Agent 有持久化的会话存储比如 MySQL那就得考虑一致性了。我一般用先更新数据库再删除缓存的策略而不是更新缓存。删除缓存比更新缓存安全因为删除操作是幂等的而且下次读取时会自然地从数据库加载最新数据。def update_session_with_persistence(session_id: str, field: str, value: str): # 先更新数据库 db.update_session(session_id, field, value) # 再删除缓存而不是更新缓存 r.delete(fagent:session:{session_id})这个策略在绝大多数场景下够用了。极端情况下读请求在删缓存之前拿到旧数据然后写请求删缓存读请求再把旧数据写回缓存确实会有一致性问题但概率极低而且 Agent 场景对会话状态的实时一致性要求没那么高可以接受。4.4 热点 key 的分散策略如果某个 key 的访问量特别大单个 Redis 节点的压力会很大。这时候可以在 key 后面加随机后缀把同一个逻辑 key 分散到多个物理 key 上。import random def get_sharded_cache(base_key: str, shard_count: int 10): shard random.randint(0, shard_count - 1) return r.get(f{base_key}:shard:{shard}) def set_sharded_cache(base_key: str, value: str, ttl: int, shard_count: int 10): for i in range(shard_count): r.setex(f{base_key}:shard:{i}, ttl, value)这个方案的代价是写入时要写多份读取时随机读一份。适合读多写极少的热点数据比如 Agent 的系统提示词、工具定义这些。写多读少的数据不适合用这个方案。5. 缓存治理从能用到好用5.1 内存监控与淘汰策略Redis 的内存是有限的Agent 缓存如果不加治理迟早会把内存吃满。首先要做的是设置合理的内存上限和淘汰策略。在redis.conf里配置maxmemory 4gb maxmemory-policy allkeys-lruallkeys-lru表示当内存达到上限时优先淘汰最近最少使用的 key。这个策略适合缓存场景因为缓存数据丢了可以重建。但要注意如果你的 Redis 里同时存了锁、计数器这类不能丢的数据就不能用allkeys-lru得用volatile-lru只淘汰设置了过期时间的 key。监控方面INFO memory命令能看到当前内存使用情况INFO stats能看到命中率。命中率低于 80% 就说明缓存策略有问题要么是 key 设计不合理要么是 TTL 设得太短。redis-cli INFO memory | grep used_memory_human redis-cli INFO stats | grep keyspace_hits redis-cli INFO stats | grep keyspace_misses5.2 大 key 和热 key 的排查大 key 是 Redis 性能的隐形杀手。一个几 MB 的 Hash 或者 List每次操作都要传输大量数据网络带宽和序列化开销都很可观。Agent 场景下最容易出现大 key 的地方是对话历史如果不做LTRIM一个活跃会话的 List 能长到几万条。排查大 key 用redis-cli --bigkeys它会扫描所有 key 并报告每种类型中最大的几个。热 key 的排查稍微麻烦一点可以用redis-cli --hotkeys需要 Redis 4.0 以上并且开启了 LFU 淘汰策略或者用MONITOR命令采样一段时间看哪些 key 出现频率最高。redis-cli --bigkeys redis-cli --hotkeys发现大 key 之后的处理方式如果是 List 或 Hash用LTRIM或者HDEL分批删除不要用DEL一次性删因为DEL大 key 会阻塞 Redis 主线程。Redis 4.0 之后可以用UNLINK命令异步删除不阻塞主线程。5.3 缓存预热与降级Agent 服务重启后缓存是空的这时候所有请求都会穿透到后端。如果 QPS 高后端可能直接被压垮。解决办法是缓存预热服务启动时主动把热点数据加载到 Redis 里。def warmup_cache(): hot_prompts load_hot_prompts_from_db() for prompt in hot_prompts: cache_llm_result(prompt[text], prompt[model], prompt[params], prompt[result])预热的数据来源可以是历史访问日志里出现频率最高的 prompt也可以是运营配置的固定问题集。预热的时机最好在服务正式接流量之前比如在健康检查通过之后、注册到负载均衡之前。降级策略是另一道保险。当 Redis 不可用时Agent 不能直接挂掉而是应该降级到直接调用后端。实现方式是在缓存访问层加 try-catchRedis 异常时记录日志并走 fallback 逻辑。def safe_get_cache(key: str): try: return r.get(key) except redis.RedisError as e: logger.warning(fRedis unavailable, fallback to backend: {e}) return None5.4 缓存 key 的规范与清理key 的命名规范前面提过了这里补充一下清理策略。Agent 的缓存 key 数量会随着用户量增长而增长必须有一套自动清理机制。除了 TTL 自动过期还需要定期清理那些僵尸 key——已经不再被访问但还没到过期时间的 key。我的做法是给每个 key 在写入时记录一个最后访问时间的 field如果是 Hash 结构然后跑一个定时任务扫描超过一定时间没被访问的 key 并删除。或者更简单一点直接用较短的 TTL让 Redis 自己淘汰。注意不要用KEYS *命令做批量清理这个命令会阻塞 Redis 主线程生产环境绝对不能用。要用SCAN命令游标遍历每次只取一小批。def clean_expired_session_keys(pattern: str agent:session:*, batch_size: int 100): cursor 0 while True: cursor, keys r.scan(cursor, matchpattern, countbatch_size) for key in keys: if r.ttl(key) -1: # 没有设置过期时间的 key r.expire(key, 1800) # 补设 TTL if cursor 0: break6. 常见问题与排查技巧实录6.1 RedisCommandTimeoutException 的排查思路这个报错在 Agent 项目里出现频率极高但原因可能有好几种。我的排查顺序是这样的先看是不是大 key 导致的。用--bigkeys扫一遍如果有超过 1MB 的 key基本就是它了。大 key 的读写会占用大量网络带宽导致后续命令排队超时。再看是不是慢查询。SLOWLOG GET 10能看到最近 10 条慢查询如果某条命令执行时间超过 10ms就要关注了。Agent 场景下常见的慢查询是KEYS、HGETALL大 Hash、SMEMBERS大 Set。最后看连接池配置。Lettuce 默认是单连接多路复用如果并发高单连接会成为瓶颈。可以改成连接池模式配置spring.redis.lettuce.pool.max-active等参数。排查项命令判断标准处理方式大 keyredis-cli --bigkeys单 key 超过 1MB拆分或分批删除慢查询SLOWLOG GET 10单命令超过 10ms优化命令或数据结构连接池INFO clients连接数接近上限调大连接池或加节点内存INFO memory使用率超过 80%扩容或调整淘汰策略网络redis-cli --latency延迟超过 1ms检查网络或部署位置6.2 缓存穿透与空值缓存缓存穿透是指查询一个不存在的 key缓存里没有每次都打到后端。Agent 场景下如果用户输入了一个从没出现过的问题就会触发穿透。解决办法是缓存空值即使后端返回空也往 Redis 里存一个特殊标记TTL 设短一点比如 60 秒。def get_with_null_cache(key: str, loader_func, null_ttl: int 60): cached r.get(key) if cached is not None: return None if cached __NULL__ else cached result loader_func() if result is None: r.setex(key, null_ttl, __NULL__) else: r.setex(key, 1800, result) return result6.3 序列化兼容性问题Java 项目里用 JDK 序列化的时候如果实体类改了字段反序列化旧数据会报InvalidClassException。解决办法是显式声明serialVersionUID或者干脆换成 JSON 序列化。Spring Data Redis 里配置 JSON 序列化Bean public RedisTemplateString, Object redisTemplate(RedisConnectionFactory factory) { RedisTemplateString, Object template new RedisTemplate(); template.setConnectionFactory(factory); Jackson2JsonRedisSerializerObject serializer new Jackson2JsonRedisSerializer(Object.class); ObjectMapper mapper new ObjectMapper(); mapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY); serializer.setObjectMapper(mapper); template.setKeySerializer(new StringRedisSerializer()); template.setValueSerializer(serializer); template.setHashKeySerializer(new StringRedisSerializer()); template.setHashValueSerializer(serializer); template.afterPropertiesSet(); return template; }6.4 分布式锁的常见坑分布式锁在 Agent 场景下用得很多但坑也不少。最常见的是锁过期了业务还没执行完导致两个请求同时持有锁。解决办法是加看门狗机制定期给锁续期。Redisson 已经内置了这个机制直接用RLock就行。RLock lock redissonClient.getLock(agent:lock: key); try { if (lock.tryLock(3, 30, TimeUnit.SECONDS)) { // 业务逻辑 } } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } }另一个坑是锁的粒度。锁的粒度太粗所有请求都串行化性能差粒度太细锁的数量爆炸管理成本高。我的经验是锁的粒度跟缓存 key 的粒度保持一致一个缓存 key 对应一把锁。6.5 缓存与数据库双写一致性速查场景策略一致性性能适用性先更新 DB 再删缓存推荐最终一致好绝大多数场景先删缓存再更新 DB不推荐差好几乎不用延迟双删可用较好中一致性要求高时更新 DB 同时更新缓存不推荐差好并发低时可用订阅 binlog 异步删缓存复杂最终一致好大型系统7. 一些踩坑之后的经验之谈做 AI Agent 的 Redis 缓存我最大的体会是别把缓存当数据库用。Agent 的数据天然是多样化的有临时的、有持久的、有热的有冷的一股脑全塞 Redis 里最后一定是内存爆掉、性能下降。该分层的分层该落库的落库Redis 只负责它最擅长的那部分。另一个体会是 TTL 一定要设而且要根据数据的实际生命周期来设。我见过太多项目因为没设 TTL 导致 Redis 内存无限增长最后不得不半夜起来手动清理。宁可 TTL 设短一点缓存命中率低一点也不要让内存失控。最后说一个实操小技巧在 Agent 的缓存 key 里加一个环境前缀比如prod:agent:llm:...和dev:agent:llm:...。这样多个环境共用一个 Redis 实例的时候不会互相干扰排查问题的时候也能快速区分。这个习惯看起来不起眼但在多环境部署的项目里能省很多事。
