1. 为什么 AI Agent 需要 Redis 缓存1.1 从一次线上事故说起去年冬天我负责的一个 AI Agent 项目在凌晨两点突然告警。这个 Agent 主要做智能客服底层接的是大语言模型用户问一个问题Agent 要调用三到五个工具再经过两轮推理才能给出答案。平时响应时间在 1.5 秒左右那天晚上直接飙到 12 秒超时率从 0.3% 涨到 17%。排查下来原因很朴素同一个用户连续问了五遍“我的订单到哪了”Agent 每次都重新走一遍完整的推理链路每次都去查一遍订单库每次都调用一次大模型。大模型调用本身就有延迟再加上工具调用的网络开销五次重复请求把后端压得喘不过气。这件事让我彻底意识到AI Agent 和传统 Web 服务在缓存这件事上的逻辑完全不一样。传统接口缓存的是数据Agent 缓存的是“思考过程”和“中间结果”。如果不在架构层面把缓存做进去Agent 的并发能力基本上就是纸糊的。1.2 AI Agent 的缓存到底在缓存什么很多人一听到“AI Agent 缓存”第一反应是缓存大模型的输出。这个理解对但不全对。一个典型的 AI Agent 请求链路大概长这样用户输入 → 意图识别 → 工具选择 → 工具调用 → 结果整合 → 大模型生成 → 返回用户这条链路上至少有四个地方可以缓存意图识别结果同样的问法意图大概率相同没必要每次都跑一遍分类模型工具调用结果查订单、查天气、查库存这些外部数据在一定时间窗口内是稳定的大模型生成结果相同或相似的 prompt输出可以复用会话上下文多轮对话中历史消息的向量化表示可以缓存Redis 在这里扮演的角色不只是一个 KV 存储而是 Agent 的“短期记忆层”。它要扛住高并发读写要支持丰富的数据结构还要能设置精细的过期策略。这也是为什么 Redis 几乎成了 AI Agent 缓存的事实标准。1.3 适合哪些人参考这套方案如果你正在做下面这些事情这篇内容应该能帮到你用 Python、Java 或 Rust 搭建 AI Agent发现并发一上来就崩已经用了 Redis但只是当普通缓存用没针对 Agent 场景做优化想了解 Agent 缓存和传统缓存到底有什么区别正在选型纠结用 Redis 还是本地缓存还是别的方案我不打算讲太多理论重点放在“我实际怎么做的”和“踩过哪些坑”。2. 缓存方案的整体设计与选型逻辑2.1 为什么是 Redis 而不是本地缓存本地缓存比如 Python 的functools.lru_cache或者 Java 的 Caffeine速度快没有网络开销看起来很美。但在 Agent 场景下它有几个致命问题第一Agent 通常是无状态部署的可能同时跑十几个实例。本地缓存各存各的命中率直接除以实例数。用户第一次请求打到实例 A第二次打到实例 B缓存就失效了。第二Agent 的缓存需要精细的过期控制。不同工具的结果有效期不一样查天气可能 10 分钟查订单可能 30 秒。本地缓存做这种差异化过期很别扭。第三Agent 的会话上下文需要跨实例共享。用户在多轮对话中请求可能被负载均衡打到不同实例上本地缓存根本没法共享上下文。Redis 作为集中式缓存天然解决这些问题。代价是每次读写多一次网络往返但在内网环境下这个开销通常在 1 毫秒以内相比大模型动辄几百毫秒的推理时间完全可以忽略。2.2 Redis 数据类型在 Agent 缓存中的具体用法Redis 有五种基础数据类型在 Agent 缓存里各有各的用处。我用一张表说清楚数据类型Agent 中的用途典型 Key 设计过期策略String缓存大模型输出、工具调用结果agent:llm:{hash}5-30 分钟Hash缓存会话上下文、用户画像agent:session:{session_id}30 分钟滑动List缓存对话历史、工具调用链agent:history:{session_id}1 小时Set缓存去重后的意图标签agent:intent:{user_id}10 分钟ZSet缓存带权重的工具优先级agent:tools:{agent_id}1 小时这里重点说两个。String 类型用来缓存大模型输出时key 的设计很关键。我一般用 prompt 的 SHA256 哈希值做 key这样相同的 prompt 自然命中同一个缓存。但要注意Agent 的 prompt 里往往包含时间戳、用户 ID 这些变量直接哈希会导致命中率极低。我的做法是先把 prompt 里的动态部分抽出来只对“模板部分”做哈希。Hash 类型用来存会话上下文特别合适。一个会话里有用户 ID、历史消息、当前意图、已调用的工具列表用 Hash 存可以单独更新某个字段不用整体覆盖。而且 Hash 支持对单个字段设置过期时间Redis 7.4 之后这对会话管理非常友好。2.3 缓存粒度怎么定缓存粒度太粗命中率低太细管理成本高。我的经验是按“工具调用”为最小粒度按“意图”为聚合粒度。举个例子。用户问“帮我查一下北京明天的天气然后推荐一家附近的火锅店”。这个请求包含两个工具调用查天气、查餐厅。我会分别缓存这两个工具的结果key 分别是agent:tool:weather:beijing:2025-01-15和agent:tool:restaurant:beijing:hotpot。同时整个意图的结果也会缓存一份key 是agent:intent:{hash}。这样设计的好处是如果用户换个问法“北京明天天气怎么样”意图变了但工具调用没变工具级缓存依然能命中。实测下来这种双层缓存能把整体命中率从 40% 提升到 75% 左右。2.4 缓存失效策略的选择Agent 缓存的失效策略比传统缓存复杂因为不同数据的时效性差异很大。我一般用三种策略组合TTL 过期最常用适合工具调用结果。天气 10 分钟订单 30 秒新闻 5 分钟。主动失效当底层数据变更时主动删除相关缓存。比如用户下了新订单就删掉该用户的订单查询缓存。版本号控制在 key 里加一个版本号当 Agent 的 prompt 模板或工具逻辑变更时版本号加一旧缓存自然失效。注意不要用KEYS *来批量删除缓存线上环境会阻塞 Redis。用SCAN配合UNLINK或者用 Hash Tag 把相关 key 归到同一个 slot再用 Lua 脚本批量删除。3. 核心细节解析与实操要点3.1 Key 设计规范别让缓存变成垃圾场我见过太多项目Redis 里的 key 五花八门有的用中文有的用 UUID有的干脆把整个 JSON 当 key。这种缓存用不了多久就会变成垃圾场既查不到也删不干净。我的 key 设计规范是这样的{业务前缀}:{数据类型}:{业务标识}:{版本号}具体到 Agent 场景agent:llm:prompt_sha256:abc123:v1 agent:tool:weather:beijing:20250115:v1 agent:session:user_12345:conv_67890:v1几个要点用冒号分隔这是 Redis 社区的惯例很多可视化工具会自动按冒号分组业务标识要能唯一确定数据但不要用太长的字符串版本号放在最后方便整体失效避免在 key 里放用户输入原文一是太长二是有特殊字符风险3.2 序列化方式的选择JSON 还是 MessagePack缓存大模型输出时序列化方式直接影响存储成本和读写速度。我对比过几种方案序列化方式体积速度可读性适用场景JSON大中好调试阶段、小数据MessagePack小快差生产环境、大数据Protobuf最小最快差跨语言、固定结构Pickle中快差仅 Python有安全风险我的选择是开发环境用 JSON方便调试生产环境用 MessagePack体积能省 30%-40%速度也快不少。Python 里用msgpack库Java 里用jackson-dataformat-msgpack都很成熟。注意绝对不要用 Pickle 序列化不可信的数据反序列化时可以执行任意代码这是严重的安全漏洞。3.3 缓存穿透、击穿、雪崩的应对这三个问题是缓存的老朋友了但在 Agent 场景下有新的表现。缓存穿透用户问了一个 Agent 完全无法回答的问题比如“帮我订一张去火星的机票”。这种请求每次都穿透到后端大模型调用一次浪费一次钱。我的做法是对于明确的“无法处理”结果也缓存一个空值TTL 设短一点比如 1 分钟。这样短时间内重复问同样的问题直接返回缓存。缓存击穿某个热点 key 突然过期大量请求同时打到后端。Agent 场景下这通常发生在热门工具上比如“查快递”。我的做法是用 Redis 分布式锁只让一个请求去回源其他请求等待或返回旧值。缓存雪崩大量 key 同时过期。这个在 Agent 场景下特别容易发生因为很多缓存是在同一时间批量写入的。我的做法是在 TTL 上加一个随机抖动比如原本 10 分钟的过期时间实际设置为 10 分钟 ± 2 分钟。3.4 分布式锁的正确用法Agent 调用工具时有些操作需要加锁比如“扣减库存”“更新订单状态”。Redis 分布式锁是常用方案但坑很多。一个可靠的 Redis 分布式锁需要满足加锁和设置过期时间是原子操作用SET key value NX PX 30000value 要唯一释放锁时校验 value防止误删别人的锁释放锁要用 Lua 脚本保证原子性锁的过期时间要合理太短会提前释放太长会阻塞import redis import uuid r redis.Redis() def acquire_lock(lock_key, ttl30000): token str(uuid.uuid4()) ok r.set(lock_key, token, nxTrue, pxttl) return token if ok else None def release_lock(lock_key, token): lua if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end r.eval(lua, 1, lock_key, token)注意如果你的 Agent 部署在多个机房Redis 主从切换时可能丢锁。这种场景下要考虑 Redlock 或者用更严格的共识方案。不过大多数 Agent 项目单机房部署普通分布式锁就够了。4. 实操过程与核心环节实现4.1 环境准备Redis 安装与基础配置先解决 Redis 安装。不同系统命令不一样我列一下常用的macOS 用 Homebrewbrew install redis brew services start redisLinuxUbuntu/Debiansudo apt update sudo apt install redis-server sudo systemctl enable redis-server sudo systemctl start redis-serverDocker 方式推荐环境隔离好docker run -d --name redis-agent \ -p 6379:6379 \ -v /data/redis:/data \ redis:7.4 redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru这里几个参数解释一下--appendonly yes开启 AOF 持久化防止重启丢数据--maxmemory 2gb限制最大内存Agent 缓存通常不需要太大--maxmemory-policy allkeys-lru内存满时淘汰最近最少使用的 key注意生产环境不要用noeviction策略内存满了会直接报错Agent 请求全部失败。用allkeys-lru或volatile-lru更稳妥。4.2 Python 接入 Redis 的完整代码我用redis-py库这是 Python 生态里最成熟的 Redis 客户端。先装依赖pip install redis msgpack然后封装一个 Agent 缓存类import redis import msgpack import hashlib import time import random class AgentCache: def __init__(self, hostlocalhost, port6379, db0): self.client redis.Redis( hosthost, portport, dbdb, decode_responsesFalse, socket_timeout2, socket_connect_timeout2, retry_on_timeoutTrue ) def _make_key(self, prefix, identifier, versionv1): return fagent:{prefix}:{identifier}:{version} def _hash_prompt(self, prompt): return hashlib.sha256(prompt.encode()).hexdigest()[:16] def get_llm_result(self, prompt): key self._make_key(llm, self._hash_prompt(prompt)) data self.client.get(key) if data: return msgpack.unpackb(data, rawFalse) return None def set_llm_result(self, prompt, result, ttl600): key self._make_key(llm, self._hash_prompt(prompt)) jitter random.randint(-60, 60) self.client.setex(key, ttl jitter, msgpack.packb(result)) def get_tool_result(self, tool_name, params): param_str :.join(f{k}{v} for k, v in sorted(params.items())) key self._make_key(tool, f{tool_name}:{param_str}) data self.client.get(key) if data: return msgpack.unpackb(data, rawFalse) return None def set_tool_result(self, tool_name, params, result, ttl300): param_str :.join(f{k}{v} for k, v in sorted(params.items())) key self._make_key(tool, f{tool_name}:{param_str}) jitter random.randint(-30, 30) self.client.setex(key, ttl jitter, msgpack.packb(result))这段代码有几个设计点值得说第一socket_timeout设成 2 秒。Agent 对延迟敏感Redis 如果 2 秒还没响应说明有问题直接失败比拖着强。第二retry_on_timeoutTrue让客户端在超时后自动重试一次应对偶发的网络抖动。第三TTL 加了随机抖动避免批量过期导致的雪崩。第四用msgpack而不是 JSON体积和速度都更优。4.3 会话上下文的管理多轮对话是 Agent 的核心能力会话上下文的管理直接影响用户体验。我用 Hash 结构存会话class SessionManager: def __init__(self, cache): self.cache cache self.client cache.client def get_session(self, session_id): key fagent:session:{session_id}:v1 data self.client.hgetall(key) if not data: return None return {k.decode(): msgpack.unpackb(v, rawFalse) for k, v in data.items()} def update_session(self, session_id, field, value, ttl1800): key fagent:session:{session_id}:v1 self.client.hset(key, field, msgpack.packb(value)) self.client.expire(key, ttl) def append_history(self, session_id, message, max_len20): key fagent:history:{session_id}:v1 self.client.lpush(key, msgpack.packb(message)) self.client.ltrim(key, 0, max_len - 1) self.client.expire(key, 3600) def get_history(self, session_id, count10): key fagent:history:{session_id}:v1 data self.client.lrange(key, 0, count - 1) return [msgpack.unpackb(item, rawFalse) for item in data]这里用lpushltrim维护一个固定长度的对话历史最新的消息在最前面。ltrim保证 List 不会无限增长最多保留 20 条。每次操作都刷新过期时间实现滑动过期。4.4 缓存命中率的监控缓存做了但不知道效果怎么样等于白做。我一般用 Redis 自带的INFO命令看命中率redis-cli INFO stats | grep keyspace输出里关注两个指标keyspace_hits命中次数keyspace_misses未命中次数命中率 hits / (hits misses)。Agent 场景下工具级缓存命中率能到 60%-80%大模型输出缓存命中率低一些20%-40% 是正常水平。如果命中率太低排查方向key 设计是否包含太多动态变量TTL 是否设得太短缓存写入是否失败看 Redis 日志我还习惯在应用层埋点记录每次请求的缓存命中情况这样能精确到具体是哪个工具、哪个 prompt 命中率低。5. 常见问题与排查技巧实录5.1 连接超时Redis command timed out这是最常见的报错完整信息通常是io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 second(s)原因可能有几种Redis 服务器负载高响应慢网络抖动客户端连接池不够用有大 key 操作阻塞了 Redis排查步骤先看 Redis 慢查询日志redis-cli SLOWLOG GET 10看当前连接数redis-cli INFO clients看内存使用redis-cli INFO memory检查是否有大 keyredis-cli --bigkeys如果是连接池不够调大maxTotal和maxIdle。如果是大 key把大 key 拆成多个小 key。5.2 缓存与数据库不一致Agent 缓存的数据往往来自数据库如果数据库更新了但缓存没更新用户就会看到旧数据。这个问题没有银弹我的策略是对于实时性要求高的数据如订单状态缓存 TTL 设短30 秒以内对于实时性要求低的数据如商品描述TTL 可以设长几小时关键更新操作采用“先更新数据库再删除缓存”的策略极端场景下用消息队列异步刷新缓存注意不要用“先删除缓存再更新数据库”并发场景下会导致脏数据。也不要用“更新数据库再更新缓存”两个并发更新可能写入旧值。5.3 内存溢出与淘汰策略Agent 缓存如果不控制很容易把 Redis 内存吃满。除了设置maxmemory还要注意大模型输出可能很长单个 value 几 KB 到几十 KB要限制单条大小会话历史要设上限不能无限增长定期清理无用 key可以用SCANTTL检查我一般给 Agent 缓存单独分配一个 Redis 实例或 db避免和其他业务互相影响。5.4 常见问题速查表问题现象可能原因排查方法解决方案命中率低key 含动态变量抽样看 key抽取模板部分做 key响应变慢大 key 阻塞--bigkeys拆分大 key内存增长快TTL 设置不当INFO memory调整 TTL加淘汰策略连接超时连接池不足INFO clients调大连接池数据不一致更新顺序错误查日志先更新 DB 再删缓存锁失效主从切换查 Redis 日志用 Redlock 或单点5.5 几个我踩过的坑第一个坑用KEYS *清理缓存。线上 Redis 有几百万 keyKEYS *直接把 Redis 阻塞了 3 秒所有 Agent 请求超时。后来改用SCAN游标遍历每次只取 100 个对 Redis 几乎无影响。第二个坑缓存了大模型的流式输出。Agent 用流式返回时我试图把每个 chunk 都缓存下来结果 Redis 写入量暴增而且流式输出本身就不适合缓存。后来改成只缓存最终完整结果。第三个坑忘了给会话 Hash 设置过期时间。用户聊完天后会话数据一直留在 Redis 里一个月后内存涨到 8GB。后来统一加了 30 分钟滑动过期内存稳定在 500MB 以内。第四个坑分布式锁的 value 用了固定字符串。两个请求同时加锁第一个释放时把第二个的锁也删了。后来改成 UUID释放时校验 value问题解决。5.6 性能优化的几个实用技巧用 pipeline 批量操作减少网络往返。比如一次要写 10 个 key用 pipeline 能把 10 次 RTT 压缩成 1 次。用 Lua 脚本做原子操作比如“检查缓存是否存在不存在则写入”避免竞态。热点 key 做本地缓存 Redis 二级缓存进一步降低 Redis 压力。用 Redis 7 的CLIENT NO-EVICT保护关键连接不被淘汰。监控用INFO命令别用MONITOR后者会严重拖慢 Redis。6. 从单机到集群的扩展思路6.1 什么时候需要 Redis 集群单机 Redis 一般能扛 5 万到 10 万 QPS对于大多数 Agent 项目够用了。但如果你的 Agent 日活用户超过 10 万或者缓存数据超过 16GB就该考虑集群了。Redis 集群有两种模式主从复制和分片集群。主从复制解决高可用分片集群解决容量和吞吐。Agent 缓存通常两者都需要。6.2 主从复制的配置要点Docker 起主从很简单# 主节点 docker run -d --name redis-master -p 6379:6379 redis:7.4 # 从节点 docker run -d --name redis-slave -p 6380:6379 redis:7.4 \ redis-server --replicaof redis-master 6379主从复制下读可以走从节点写走主节点。Agent 缓存读多写少这个架构能显著提升读吞吐。注意主从复制是异步的主节点写入后从节点可能还没同步。如果 Agent 对一致性要求高读也走主节点或者用WAIT命令等待同步。6.3 分片集群的 key 设计调整分片集群把数据分散到多个节点key 会按 CRC16 哈希到 16384 个 slot 上。这带来一个新问题如果相关的 key 落在不同节点就没法用 Lua 脚本或事务原子操作。解决办法是用 Hash Tag。在 key 里用{}包裹的部分会被用来计算 slot保证相同 tag 的 key 落在同一节点agent:session:{user_123}:conv_456 agent:history:{user_123}:conv_456这样user_123相关的 key 都在同一节点可以原子操作。6.4 集群下的监控与运维集群环境下监控要覆盖每个节点。我一般用 Redis Insight 或者 Another Redis Desktop Manager 这类工具做可视化管理用 Prometheus Grafana 做指标监控。关键指标每个节点的 QPS、内存、连接数集群整体命中率慢查询数量主从延迟如果某个节点内存明显高于其他节点说明 key 分布不均需要检查 key 设计。7. 一些个人体会这套 Agent 缓存方案我在三个项目里用过最大的感受是缓存不是加个 Redis 就完事了key 设计、TTL 策略、失效逻辑每一个细节都影响最终效果。我见过太多项目Redis 装了代码也写了但命中率只有 10%等于白做。另一个体会是Agent 缓存和传统缓存最大的区别在于“语义”。传统缓存缓存的是数据Agent 缓存缓存的是“思考”。这意味着你不能简单地用 URL 做 key而要考虑 prompt 的语义相似性。未来如果向量数据库和 Redis 结合得更紧密基于语义的缓存命中率还能再上一个台阶。最后分享一个小技巧在 Agent 的 prompt 里加一句“如果缓存中有相关结果请优先使用”虽然大模型不一定听但配合工具调用的缓存检查逻辑能进一步减少不必要的工具调用。这个技巧在工具调用成本高的场景下特别有用。