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(host='localhost', port=6379, decode_responses=True) def append_message(session_id, role, content): msg = json.dumps({"role": role, "content": content}, ensure_ascii=False) pipe = r.pipeline() pipe.rpush(f"session:{session_id}:messages", msg) pipe.ltrim(f"session:{session_id}:messages", -40, -1) # 保留最近40条 pipe.expire(f"session:{session_id}:messages", 86400) pipe.execute() def get_recent_messages(session_id, limit=20): raw = r.lrange(f"session:{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；工具调用结果这种结构固定、体量可能很大的（比如一次网页抓取返回的正文），用 MessagePack；Token 计数这种纯数字，直接用 Redis 的计数器，别序列化。有个项目我一开始全用 JSON，后来发现单次会话状态平均 12KB，一天 50 万轮对话就是 6GB 的读写量，换成 MessagePack 后降到 7KB 左右，Redis 带宽压力明显缓解。

import msgpack def set_tool_result(session_id, tool_name, result, ttl=300): key = f"session:{session_id}:tool:{tool_name}" packed = msgpack.packb(result, use_bin_type=True) r.setex(key, ttl, packed) def get_tool_result(session_id, tool_name): key = f"session:{session_id}:tool:{tool_name}" packed = r.get(key) if packed is None: return None return msgpack.unpackb(packed, raw=False)

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_keys=True, ensure_ascii=False) digest = hashlib.sha256(normalized.encode()).hexdigest()[:16] return f"tool_cache:{tool_name}:{digest}"

提示：哈希截断到 16 位十六进制（64 bit）在绝大多数场景下碰撞概率可以忽略，但如果你有上亿级别的缓存条目，建议用完整 64 位十六进制或者加一层布隆过滤器做前置判断。

还有一个容易忽略的点：有些工具的结果带时间敏感性。天气、股价、新闻这类数据，缓存 5 分钟和缓存 1 小时完全是两回事。我的做法是在工具注册时就声明 TTL，而不是全局统一。下面这张表是我项目里常用的 TTL 参考：

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, ttl=10): token = str(uuid.uuid4()) ok = r.set(f"lock:{lock_name}", token, nx=True, ex=ttl) 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, f"lock:{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 = f"token_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 = f"token_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, f"session:{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。这样当缓存结构变更时，直接升版本号，旧键自然过期淘汰，不用写迁移脚本，也不用担心新旧格式混读。这个习惯帮我省过好几次数据迁移的麻烦。