1. 从一次线上抖动说起AI Agent 为什么绕不开 Redis 缓存去年冬天我负责的一个 AI Agent 项目在凌晨两点突然报警接口 P99 从 180ms 飙到 4.2s值班同学第一反应是模型服务挂了结果排查下来发现是 Agent 的会话状态存储层被打爆了——每个用户对话轮次都要读写一次全量上下文QPS 一上来数据库直接跪。那次事故之后我把整个 Agent 的状态管理重构成了 Redis 缓存优先的架构P99 稳定在 200ms 以内成本还降了四成。这件事让我意识到一个很现实的问题大部分 AI Agent 教程都在讲 Prompt 工程、工具调用、多智能体编排却很少有人认真聊缓存这一层而恰恰是这一层决定了你的 Agent 能不能从 Demo 走向生产。这篇内容我想聊的就是 AI Agent 和 Redis 缓存结合这件事。核心关键词是ai agent、redis、缓存但我不想写成一篇 Redis 命令手册也不想写成 Agent 架构综述而是聚焦在一个真实跑起来的 Agent 系统里Redis 到底该放在哪、存什么、怎么存、什么时候失效这几个具体问题上。适合的读者是已经写过基础 Agent 循环、准备上生产或者正在被性能和成本折磨的开发者也适合对分布式缓存感兴趣但没想清楚在 AI 场景怎么用的后端同学。如果你还在纠结 Redis 的 String 和 Hash 有什么区别这篇可能稍微超前一点但我会尽量把每个决策背后的原因讲透让你能直接抄作业。先说结论性的判断AI Agent 对 Redis 的依赖和传统 Web 应用对 Redis 的依赖本质上是两回事。传统应用缓存的是数据库查询结果是读多写少、可重建的数据而 Agent 缓存的是会话状态、工具调用结果、向量检索中间态、Token 预算计数这些东西它们有的可重建、有的不可重建、有的生命周期只有几秒、有的要跨天保留。用同一套缓存策略去套必然出问题。下面我按实际项目里的分层来拆。2. Agent 会话状态该不该进 Redis一次架构取舍的完整推演2.1 会话状态放内存、放数据库还是放 Redis最开始做 Agent 的时候最自然的做法是把对话历史放在进程内存里一个 dict 或者 list 就搞定了。Demo 阶段完全没问题但一旦你要做多副本部署、要做滚动更新、要支持用户跨设备继续对话内存方案立刻崩盘。我见过一个团队为了保住内存里的会话硬是把服务做成了单副本加长连接粘性路由结果一次发布要停服三分钟用户全部掉线。那放数据库行不行行但要看你怎么用。Agent 的会话状态有两个特点读写频率极高每一轮对话至少读写一次工具调用密集时一轮内可能读写十几次数据体量不大但结构复杂消息列表、工具调用记录、中间推理步骤、Token 计数。用关系型数据库存每次都要序列化反序列化加事务开销QPS 上千就开始吃力。我实测过一个中等复杂度的 Agent单轮对话平均触发 6 次状态读写用 PostgreSQL 做存储时 P95 在 400ms 左右换成 Redis 之后降到 60ms 以内。所以结论很清晰Redis 做会话状态的热存储数据库做冷备份和审计。具体做法是每轮对话结束后异步把完整会话快照写一份到数据库Redis 里只保留最近 N 轮的活跃上下文。这样既保证了读取性能又不会因为 Redis 故障丢失全部历史。这里有个细节要注意异步落库要用消息队列解耦不要直接在请求链路里写库否则 Redis 省下来的时间又被数据库吃回去了。2.2 会话数据的 Redis 数据结构选型选数据结构这件事很多人上来就用 String 存 JSON简单是简单但有几个坑。第一每次更新都要读全量、改、写全量网络传输和序列化开销随会话长度线性增长第二并发更新同一会话时容易覆盖需要额外加锁第三想单独取某个字段比如只取 Token 计数也得反序列化整个 JSON。我的做法是分层存储会话元信息用户 ID、创建时间、模型配置、Token 预算用 Hash消息列表用 List 或者 Stream工具调用结果用独立的 String 加 TTL。这样更新 Token 计数只需要HINCRBY追加消息只需要RPUSH都是 O(1) 操作不用读全量。消息列表如果要做只保留最近 20 轮的滑动窗口用LTRIM一行搞定。import redis import json r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def append_message(session_id, role, content): msg json.dumps({role: role, content: content}, ensure_asciiFalse) pipe r.pipeline() pipe.rpush(fsession:{session_id}:messages, msg) pipe.ltrim(fsession:{session_id}:messages, -40, -1) # 保留最近40条 pipe.expire(fsession:{session_id}:messages, 86400) pipe.execute() def get_recent_messages(session_id, limit20): raw r.lrange(fsession:{session_id}:messages, -limit, -1) return [json.loads(m) for m in raw]注意LTRIM的索引是闭区间-40, -1表示保留最后 40 个元素。如果你要保留最近 20 轮对话而每轮包含 user 和 assistant 两条消息那应该是 40 条别搞错了。2.3 序列化格式的选择JSON、MessagePack 还是 Protobuf序列化这件事看起来是细节但在高频读写的 Agent 场景里影响不小。JSON 可读性好、调试方便但体积大、解析慢MessagePack 体积能压到 JSON 的 60% 左右解析速度也快一截Protobuf 体积最小、速度最快但需要预定义 schema改字段麻烦。我的经验是分场景选会话消息这种需要人工排查、字段经常变动的用 JSON工具调用结果这种结构固定、体量可能很大的比如一次网页抓取返回的正文用 MessagePackToken 计数这种纯数字直接用 Redis 的计数器别序列化。有个项目我一开始全用 JSON后来发现单次会话状态平均 12KB一天 50 万轮对话就是 6GB 的读写量换成 MessagePack 后降到 7KB 左右Redis 带宽压力明显缓解。import msgpack def set_tool_result(session_id, tool_name, result, ttl300): key fsession:{session_id}:tool:{tool_name} packed msgpack.packb(result, use_bin_typeTrue) r.setex(key, ttl, packed) def get_tool_result(session_id, tool_name): key fsession:{session_id}:tool:{tool_name} packed r.get(key) if packed is None: return None return msgpack.unpackb(packed, rawFalse)3. 工具调用结果的缓存Agent 降本增效最狠的一刀3.1 为什么工具调用结果是缓存收益最高的部分一个 Agent 跑一轮对话最贵的往往不是模型推理而是工具调用。搜索 API 按次收费、数据库查询占连接池、第三方接口有速率限制而很多工具调用在短时间内是幂等且可复用的。比如用户问今天北京天气怎么样Agent 调用天气 API五分钟后另一个用户问同样的问题如果每次都打 API纯属浪费。我在一个客服 Agent 项目里做过统计工具调用结果缓存命中率稳定在 35% 到 50% 之间也就是说将近一半的外部调用可以直接省掉。按当时搜索 API 每次 0.02 元算一天 20 万次调用光这一项每月省下近千元还不算延迟降低带来的体验提升。缓存命中时响应时间从平均 800ms 降到 30ms 以内用户感知非常明显。3.2 缓存键怎么设计才能既命中又不串味工具结果缓存的键设计是核心难点。设计得太粗不同参数的结果会串设计得太细命中率上不去。我的原则是把工具名和所有影响结果的参数一起做哈希但要注意参数归一化。举个例子搜索工具的参数是{query: 北京天气, limit: 5}另一个请求是{limit: 5, query: 北京天气}字典顺序不同但语义相同如果直接str(params)做键就会产生两个缓存条目。正确做法是先对参数排序再序列化或者用稳定的哈希函数。import hashlib import json def make_tool_cache_key(tool_name, params): normalized json.dumps(params, sort_keysTrue, ensure_asciiFalse) digest hashlib.sha256(normalized.encode()).hexdigest()[:16] return ftool_cache:{tool_name}:{digest}提示哈希截断到 16 位十六进制64 bit在绝大多数场景下碰撞概率可以忽略但如果你有上亿级别的缓存条目建议用完整 64 位十六进制或者加一层布隆过滤器做前置判断。还有一个容易忽略的点有些工具的结果带时间敏感性。天气、股价、新闻这类数据缓存 5 分钟和缓存 1 小时完全是两回事。我的做法是在工具注册时就声明 TTL而不是全局统一。下面这张表是我项目里常用的 TTL 参考工具类型建议 TTL理由天气查询10 分钟数据更新频率约 10 分钟汇率查询5 分钟波动快但短时内可接受网页抓取30 分钟页面内容变化不频繁知识库检索1 小时内部文档更新周期长数据库查询1 分钟数据实时性要求高计算类工具永久纯函数结果确定3.3 缓存穿透和雪崩在 Agent 场景的特殊表现缓存穿透查询不存在的数据每次都打到后端在 Agent 场景里有个特殊变种用户问了一个工具无法回答的问题Agent 反复调用同一个工具。比如用户问帮我查一下我上个月的订单但用户 ID 传错了工具每次都返回空Agent 可能重试三次三次都穿透到后端。这种情况用布隆过滤器挡不住因为键是存在的只是结果为空。我的处理方式是空结果也缓存但 TTL 短。空结果缓存 30 秒到 1 分钟既能挡住短时间内的重复调用又不会因为数据后来出现了而长时间返回错误。同时要在 Agent 的循环控制里加同一工具连续失败 N 次就停止重试的逻辑从源头减少无效调用。缓存雪崩在 Agent 场景里表现为大量工具缓存同时过期。如果你在服务启动时预热了一批缓存TTL 又设成一样的那它们会在同一时刻集体失效。解决办法是给 TTL 加随机抖动比如基础 TTL 加 0 到 20% 的随机值。import random def set_with_jitter(key, value, base_ttl): jitter int(base_ttl * random.uniform(0, 0.2)) r.setex(key, base_ttl jitter, value)4. 分布式锁与并发控制多副本 Agent 的必修课4.1 什么情况下 Agent 需要分布式锁单副本 Agent 用进程内的threading.Lock就够了但多副本部署后同一个用户的两个请求可能落到不同副本上进程锁完全失效。这时候需要分布式锁。Agent 场景里需要锁的典型情况有三种同一会话的并发写入、工具调用的去重、Token 预算的原子扣减。第一种最常见。用户手速快连点两次发送两个请求同时到达不同副本都去读会话状态、追加消息、写回结果就是消息顺序错乱或者互相覆盖。第二种是防止同一个工具调用被重复执行比如用户问帮我下单两个请求同时触发下单工具那就是重复下单的事故。第三种是 Token 预算控制多个并发请求同时扣减预算不加锁就会超支。4.2 Redis 分布式锁的正确实现与常见误区网上流传最广的分布式锁实现是SET key value NX EX seconds这个命令本身没问题问题出在释放锁的环节。很多人写成DEL key这就埋了个大坑如果 A 获取锁后执行超时锁自动过期B 获取了锁这时 A 执行完去DEL删掉的是 B 的锁。正确做法是释放前校验 value 是不是自己的。import uuid def acquire_lock(lock_name, ttl10): token str(uuid.uuid4()) ok r.set(flock:{lock_name}, token, nxTrue, exttl) return token if ok else None def release_lock(lock_name, 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, flock:{lock_name}, token)注意校验和删除必须用 Lua 脚本保证原子性不能先GET再DEL中间有窗口期。还有一个误区是锁的 TTL 设得太短。Agent 的工具调用可能耗时几秒到几十秒如果锁 TTL 只有 5 秒业务还没执行完锁就过期了等于没锁。我的经验是 TTL 设为业务预期最大耗时的 2 到 3 倍同时业务侧要有超时控制不能无限等。如果业务耗时确实不可控那就需要看门狗机制定期续期但这会显著增加复杂度能不用就不用。4.3 用 Redis 做 Token 预算的原子扣减Token 预算是 Agent 成本控制的关键。每个用户或每个会话有月度 Token 配额每次模型调用前要检查余额、扣减、调用后按实际用量修正。这个过程必须是原子的否则并发下会超支。用 Redis 的DECRBY和INCRBY天然原子但要注意先扣后调还是先调后扣。先扣后调的问题是如果调用失败要回滚先调后扣的问题是并发下可能超支。我的做法是预扣加结算调用前按预估上限预扣调用后按实际用量返还差额。def reserve_tokens(user_id, estimated): key ftoken_budget:{user_id} remaining r.decrby(key, estimated) if remaining 0: r.incrby(key, estimated) # 回滚 return False return True def settle_tokens(user_id, estimated, actual): key ftoken_budget:{user_id} if actual estimated: r.incrby(key, estimated - actual)这套机制配合每日重置用EXPIREAT设到次日零点就能实现比较可靠的预算控制。实测下来预扣加结算的方案在并发 500 的场景下没有出现过超支。5. 缓存一致性Agent 状态更新时最容易翻车的地方5.1 先更新数据库还是先更新缓存这是缓存一致性的经典问题在 Agent 场景里同样适用。我的结论是Agent 会话状态以 Redis 为准数据库是异步备份所以不存在先更新谁的问题直接更新 Redis然后发消息让异步任务落库。但工具结果缓存不一样它的数据源是外部 API缓存只是副本更新策略要另说。对于工具结果缓存我采用的是Cache-Aside 模式读的时候先查缓存没有就查数据源并写回缓存写的时候直接删缓存下次读时自然重建。为什么不更新缓存而是删缓存因为更新缓存需要知道新值而很多工具调用是有副作用的比如下单不能为了更新缓存而重复调用。删缓存更安全。5.2 缓存与数据源不一致的排查思路线上遇到用户说数据不对但刷新一下又对了这类问题八成是缓存不一致。排查链路我一般是这样的先确认是读路径还是写路径的问题读路径看缓存键是否命中、命中的值是否过期、TTL 是否合理写路径看删除缓存的时机是否在数据源更新之后、是否有并发写导致删除被覆盖。有个隐蔽的坑是删除缓存失败但没重试。如果数据源更新成功、删缓存失败缓存里就是脏数据直到 TTL 过期才恢复。解决办法是把删除操作放进消息队列失败自动重试或者用 binlog 订阅的方式异步删缓存。Agent 场景里如果工具结果缓存脏了影响相对可控大不了重新调用但会话状态脏了就是用户看到错误的历史消息体验很差。5.3 会话状态的版本号与乐观锁会话状态并发更新的另一个解法是乐观锁。每次读取会话时带一个版本号写回时用 Lua 脚本校验版本号是否变化变了就重试。这样避免了分布式锁的开销适合冲突不激烈的场景。def update_session_with_version(session_id, new_state, expected_version): lua local current redis.call(hget, KEYS[1], version) if current ARGV[1] then redis.call(hset, KEYS[1], state, ARGV[2]) redis.call(hincrby, KEYS[1], version, 1) return 1 else return 0 end return r.eval(lua, 1, fsession:{session_id}, expected_version, new_state)实测下来在冲突率低于 5% 的场景乐观锁比分布式锁吞吐高 3 倍以上。冲突率高的时候重试成本会上去那就该考虑用锁或者把状态拆分减少冲突面。6. 生产环境踩过的坑与监控要点6.1 内存淘汰策略选错导致会话丢失Redis 默认的noeviction策略在内存满时直接报错而很多教程建议改成allkeys-lru。但在 Agent 场景里allkeys-lru有个致命问题它可能淘汰掉活跃会话的状态因为 LRU 只看访问时间不看数据重要性。我踩过一次用户正在对话中会话状态被淘汰了Agent 突然失忆用户直接投诉。正确做法是给不同用途的键设置不同的淘汰策略。Redis 4.0 之后支持volatile-lru只淘汰设置了过期时间的键。所以我的方案是会话状态、Token 预算这些关键数据不设 TTL 或者设很长的 TTL用volatile-lru时不会被淘汰工具结果缓存、临时中间态设短 TTL让它们优先被淘汰。这样既保证了关键数据安全又能让内存自动回收。如果内存实在紧张还可以用 Redis 的maxmemory-policy配合多实例把关键数据和缓存数据分到不同实例上物理隔离。6.2 大 Key 和热 Key 的识别与处理Agent 场景特别容易产生大 Key。一个会话如果聊了几百轮消息列表可能几 MB一次网页抓取的结果可能几百 KB。大 Key 的危害是操作阻塞、网络传输慢、删除时卡顿。我一般用redis-cli --bigkeys定期扫描或者用MEMORY USAGE key查单个键。处理大 Key 的思路是拆分。消息列表按时间分片比如每 50 条一个 List键名带序号工具结果如果太大考虑存到对象存储Redis 里只存引用。热 Key 则是另一个问题某个热门会话被高频访问单个 Redis 节点压力大。解法是本地缓存加随机后缀在应用层做一层短 TTL 的本地缓存或者把热 Key 复制成多个带随机后缀的键分散压力。6.3 必须监控的几个 Redis 指标生产环境跑 Agent这几个指标我每天必看命中率低于 80% 说明缓存策略有问题、内存使用率和碎片率碎片率超过 1.5 要考虑重启或整理、慢查询数量Agent 场景慢查询往往来自大 Key 操作、连接数Agent 并发高时连接池容易打满、主从延迟如果用了主从延迟大会导致读到旧数据。# 查看命中率 redis-cli info stats | grep keyspace_hits redis-cli info stats | grep keyspace_misses # 查看慢查询 redis-cli slowlog get 10 # 查看内存碎片率 redis-cli info memory | grep mem_fragmentation_ratio提示命中率要结合业务看工具结果缓存命中率 40% 就算不错但会话状态缓存命中率应该接近 100%因为它是热数据。如果会话状态命中率低说明 TTL 设太短或者内存不够导致被淘汰了。7. 我在这套架构上的一些个人体会跑了半年多最大的感受是Agent 的缓存设计没有银弹关键是分清数据的可重建性和时效性。可重建且时效性要求低的大胆缓存、TTL 可以长不可重建或者时效性要求高的要么不缓存要么 TTL 极短加严格的一致性保障。我见过太多团队一上来就全量缓存结果脏数据问题比性能问题还多。另一个体会是监控要走在优化前面。我现在的习惯是任何缓存上线前先把命中率、内存、慢查询的监控面板搭好跑一周看数据再调 TTL 和策略。凭感觉设 TTL 十有八九是错的要么太短没效果要么太长出脏数据。最后分享一个我常用的小技巧给缓存键加业务前缀和版本号比如v2:session:xxx。这样当缓存结构变更时直接升版本号旧键自然过期淘汰不用写迁移脚本也不用担心新旧格式混读。这个习惯帮我省过好几次数据迁移的麻烦。