1. 从一次线上抖动说起:AI Agent 为什么绕不开 Redis
做 AI Agent 项目的人,迟早会撞上一个很尴尬的场景:本地跑得好好的智能体,一上生产环境就开始抽风。用户发一句“帮我查一下上周的订单”,Agent 要拆解意图、调用工具、检索记忆、拼装上下文,一轮下来少则几百毫秒,多则好几秒。并发一上来,模型接口还没崩,自己的服务先扛不住了——会话状态丢了、工具调用结果重复拉取、限流计数错乱,各种问题像约好了一样同时冒出来。
我最早做 Agent 的时候也天真,觉得“不就是调个大模型接口嘛,能有多复杂”。结果第一次压测就被现实教育了:50 个并发用户,Agent 的平均响应时间从 1.2 秒飙到 8 秒以上,日志里全是重复的工具调用和超时。排查下来发现,问题根本不在模型本身,而在于状态管理和缓存层完全缺失。Agent 是有“记忆”的,多轮对话要记住上下文,工具调用结果要复用,用户会话要隔离,这些如果每次都重新计算或者全塞在进程内存里,单机还能凑合,一旦多实例部署就彻底乱套。
这就是 Redis 在 AI Agent 架构里真正的价值所在。它不是简单地“加个缓存提速”,而是承担了三个关键角色:会话状态的共享存储、工具调用结果的缓存层、以及并发控制的协调中心。你可以把 Agent 想象成一个记忆力很好但很忙的助理,Redis 就是他手边那块随时能查、随时能写的白板——所有实例共享同一块白板,谁写了什么大家都看得见,不用每个人都把信息记在自己脑子里。
这篇文章适合两类人看:一类是正在搭建 AI Agent、被并发和状态问题折磨的开发者;另一类是想搞清楚“Redis 在 Agent 里到底怎么用才不踩坑”的工程师。我会从实际架构出发,把会话缓存、工具结果缓存、分布式锁、缓存失效策略这些核心问题一个个拆开讲,配上能直接抄的代码和参数,也会分享几个我踩过的坑。Redis 的安装、数据类型这些基础我不多废话,重点放在 Agent 场景下的特殊用法和治理经验上。
2. Agent 的会话状态到底该存什么、怎么存
2.1 会话数据的结构设计:别把整个对话历史一股脑塞进去
很多人第一次做 Agent 会话存储,图省事直接把整个 messages 数组序列化成 JSON 丢进一个 key 里。刚开始没问题,但对话轮次一多,这个 value 会膨胀到几百 KB 甚至上 MB,每次读写都要全量序列化和反序列化,延迟肉眼可见地上升。更麻烦的是并发写入——两个请求同时更新同一个会话,后写的会把先写的覆盖掉,上下文直接丢失。
我的做法是分层存储。把会话数据拆成三部分:
- 热数据:最近 N 轮对话(比如最近 10 轮),用 Redis 的 List 或 Stream 存储,读写频繁,需要快速访问。
- 温数据:会话的元信息,比如用户 ID、会话创建时间、当前状态(进行中/已完成)、累计 token 消耗,用 Hash 存储,字段级更新,避免全量覆盖。
- 冷数据:完整的历史对话,定期归档到数据库或对象存储,Redis 里只保留摘要或指针。
具体结构可以这样设计:
# 会话元信息,用 Hash,支持字段级更新 HSET agent:session:{session_id} user_id "u_12345" status "active" created_at "1716000000" token_used "1520" # 最近对话轮次,用 List,LPUSH 新消息,LTRIM 控制长度 LPUSH agent:session:{session_id}:messages "{...}" LTRIM agent:session:{session_id}:messages 0 19 # 设置过期时间,避免僵尸会话堆积 EXPIRE agent:session:{session_id} 7200 EXPIRE agent:session:{session_id}:messages 7200这里有个细节值得说:为什么用 List 而不是 String 存消息。List 的 LPUSH + LTRIM 组合天然适合“只保留最近 N 条”的场景,而且支持范围查询,取最近 10 轮就是LRANGE key 0 9,不用把整个历史都拉出来。String 存 JSON 数组的话,每次追加都要读出来、解析、追加、再序列化写回去,既慢又容易并发冲突。
2.2 会话隔离与多租户:key 的命名不是小事
Agent 服务通常要同时服务多个用户、多个租户,甚至多个 Agent 实例。如果 key 命名不规范,很容易出现数据串号——A 用户的对话被 B 用户读到,这在生产环境是严重事故。
我习惯用冒号分隔的层级命名,把租户、业务、实体类型、ID 都体现在 key 里:
{tenant}:{service}:{entity}:{id}:{sub_entity}比如:
t_001:agent:session:sess_abc123:meta t_001:agent:session:sess_abc123:messages t_001:agent:toolcache:weather:beijing:20240518这样做的好处有三个:一是可读性强,运维排查时一眼能看出这个 key 属于谁;二是便于批量操作,比如要清理某个租户的所有缓存,用SCAN匹配t_001:agent:*就行(注意生产环境别用KEYS,后面会讲);三是为分片预留空间,如果将来要做 Redis Cluster,可以在 key 里加 hash tag,比如{t_001}:agent:session:...,保证同一租户的数据落在同一个槽上。
注意:key 的长度也要控制。虽然 Redis 对 key 长度没有硬限制,但过长的 key 会浪费内存,而且在高并发下网络传输也是开销。一般建议控制在 100 字节以内,把能简化的部分简化,比如
session可以缩写成sess,但别缩到看不懂。
2.3 会话过期与续期:TTL 设多少才合理
TTL 设置是个经验活。设太短,用户聊到一半会话过期了,体验极差;设太长,僵尸会话堆积,内存被白白占用。
我的经验值是根据业务场景分档:
| 场景类型 | 建议 TTL | 续期策略 |
|---|---|---|
| 一次性问答 | 5-10 分钟 | 不续期,用完即弃 |
| 多轮客服对话 | 30-60 分钟 | 每次交互后重置 TTL |
| 长期助理型 Agent | 7-30 天 | 滑动过期 + 定期归档 |
| 工具结果缓存 | 1-24 小时 | 按数据时效性设定 |
续期的实现很简单,每次读写会话时顺手EXPIRE一下:
import redis r = redis.Redis(host='localhost', port=6379, decode_responses=True) def touch_session(session_id, ttl=3600): """每次交互后刷新会话过期时间""" pipe = r.pipeline() pipe.expire(f"agent:session:{session_id}", ttl) pipe.expire(f"agent:session:{session_id}:messages", ttl) pipe.execute()用 pipeline 把多个 EXPIRE 打包发送,减少网络往返。别小看这个优化,在高频交互场景下,每次省几毫秒,累积起来很可观。
还有一个坑:如果会话数据分散在多个 key 里,一定要保证它们的 TTL 一致。我曾经遇到过 meta 过期了但 messages 还在的情况,结果读会话时拿到一个没有元信息的孤儿消息列表,程序直接报错。解决办法要么用 pipeline 统一设置,要么用 Redis 7.0 以上的 hash 字段过期功能(HEXPIRE),把相关数据放在同一个 Hash 里。
3. 工具调用结果缓存:Agent 提速最狠的一刀
3.1 哪些工具结果值得缓存,哪些绝对不能缓存
Agent 调用工具是耗时大户。一次天气查询、一次数据库检索、一次外部 API 调用,动辄几百毫秒到几秒。如果同样的查询在短时间内重复出现,缓存能直接把响应时间打到毫秒级。
但不是所有工具结果都能缓存。我总结了一个判断标准:
可以缓存的:
- 幂等查询类:天气、汇率、股票行情(短时效)、百科知识、静态配置
- 计算密集类:向量检索结果、复杂 SQL 查询结果、文档解析结果
- 外部 API 类:有明确时效性且调用成本高的接口
绝对不能缓存的:
- 涉及用户隐私的操作:转账、下单、修改密码
- 实时性要求极高的:库存扣减、秒杀库存
- 有副作用的操作:发消息、发邮件、写数据库
判断的核心就一条:同样的输入,在缓存有效期内,返回同样的输出,且不产生任何副作用。满足这条,就可以缓存。
3.2 缓存 key 的设计:怎么把工具参数变成唯一标识
工具调用的缓存 key,核心是把工具名 + 参数映射成一个唯一字符串。最直接的做法是拼 JSON 再哈希:
import hashlib import json def make_tool_cache_key(tool_name, params, version="v1"): """生成工具调用缓存 key""" # 参数排序,保证相同参数不同顺序生成相同 key normalized = json.dumps(params, sort_keys=True, ensure_ascii=False) param_hash = hashlib.md5(normalized.encode()).hexdigest()[:16] return f"agent:toolcache:{version}:{tool_name}:{param_hash}"这里有几个细节:
- 参数要排序:
{"a":1,"b":2}和{"b":2,"a":1}应该命中同一个缓存,所以用sort_keys=True。 - 加版本号:工具逻辑变更时,旧缓存可能失效,用版本号隔离,避免脏数据。比如
v1升级到v2,新请求走新 key,旧缓存自然过期。 - 哈希截断:MD5 取前 16 位足够,既保证低碰撞率,又控制 key 长度。
- 中文处理:
ensure_ascii=False让中文参数保持原样,虽然哈希后看不出区别,但调试时更友好。
对于参数特别复杂的工具,比如带嵌套对象的,还可以考虑只取关键参数做 key。比如一个“搜索商品”工具,参数有keyword、page、page_size、sort、filters,其中filters是个大对象,可以只取keyword + page + sort做 key,牺牲一点精确性换取更高的命中率。这个取舍要看业务,没有标准答案。
3.3 缓存穿透、击穿、雪崩在 Agent 场景下的真实表现
这三个经典问题在 Agent 场景下有自己的特点,我一个个说。
缓存穿透:查询一个根本不存在的工具结果,每次都打到后端。在 Agent 里,这通常表现为用户反复问一个冷门问题,或者工具参数组合特别偏。解决办法是缓存空结果,但 TTL 设短一点:
def get_tool_result(tool_name, params): key = make_tool_cache_key(tool_name, params) cached = r.get(key) if cached is not None: if cached == "__NULL__": return None # 命中空结果缓存 return json.loads(cached) # 缓存未命中,调用真实工具 result = call_real_tool(tool_name, params) if result is None: # 空结果缓存 60 秒,防止穿透 r.setex(key, 60, "__NULL__") else: r.setex(key, 3600, json.dumps(result, ensure_ascii=False)) return result缓存击穿:某个热点 key 刚好过期,大量并发请求同时打到后端。Agent 里最典型的就是热门问题——比如某个突发事件,大量用户同时问同一个问题。解决办法是互斥锁重建,只让一个请求去加载,其他请求等待:
def get_tool_result_with_lock(tool_name, params, lock_timeout=10): key = make_tool_cache_key(tool_name, params) cached = r.get(key) if cached is not None: return json.loads(cached) if cached != "__NULL__" else None lock_key = f"{key}:lock" # 尝试获取锁,NX 保证只有一个能拿到 if r.set(lock_key, "1", nx=True, ex=lock_timeout): try: result = call_real_tool(tool_name, params) r.setex(key, 3600, json.dumps(result, ensure_ascii=False) if result else "__NULL__") return result finally: r.delete(lock_key) else: # 没拿到锁,短暂等待后重试读缓存 import time time.sleep(0.1) return get_tool_result_with_lock(tool_name, params, lock_timeout)缓存雪崩:大量 key 在同一时间集中过期,请求全部涌向后端。Agent 场景下,如果批量预热了一批工具缓存,又设了相同的 TTL,就很容易触发。解决办法是TTL 加随机抖动:
import random def set_with_jitter(key, value, base_ttl=3600): """TTL 加 ±10% 随机抖动,避免集中过期""" jitter = random.randint(-base_ttl // 10, base_ttl // 10) ttl = base_ttl + jitter r.setex(key, ttl, value)这个抖动看起来不起眼,但在大规模缓存下能有效削峰。我实测过一个场景,加了抖动之后,后端峰值 QPS 从 3000 降到了 800 左右。
4. 并发控制:AI Agent 扛并发的 Redis 实战手段
4.1 分布式锁:工具调用的互斥与限流
Agent 调用外部工具时,有些工具是有速率限制的,比如某个 API 每秒只允许 10 次调用。多实例部署下,每个实例各自限流是没用的,必须有一个全局协调者,这就是 Redis 分布式锁的用武之地。
最简单的实现是SET key value NX EX:
import uuid def acquire_lock(lock_name, timeout=10): """获取分布式锁,返回锁标识或 None""" token = str(uuid.uuid4()) acquired = r.set(f"agent:lock:{lock_name}", token, nx=True, ex=timeout) return token if acquired else None def release_lock(lock_name, token): """释放锁,用 Lua 脚本保证原子性""" 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"agent:lock:{lock_name}", token)释放锁必须用 Lua 脚本,因为“判断是不是自己的锁”和“删除锁”这两步必须是原子的。如果先 GET 再 DEL,中间锁过期被别人拿走了,你就会误删别人的锁。
但说实话,单实例 Redis 锁在 Redis 主从切换时是有风险的——主节点写入锁后还没同步到从节点就宕机了,从节点升主后锁就丢了。如果业务对锁的可靠性要求极高,可以考虑 Redlock 算法(多个独立 Redis 实例投票),但 Redlock 本身也有争议,实现复杂且性能下降。我的建议是:大多数 Agent 场景用单实例锁 + 合理超时就够了,真正需要强一致的地方,应该用数据库的唯一约束或事务来兜底,而不是把宝全押在 Redis 锁上。
4.2 限流器:滑动窗口与令牌桶的 Redis 实现
Agent 服务需要限流的地方很多:单用户请求频率、单工具调用频率、全局 QPS 上限。Redis 实现限流有两种常用方案。
固定窗口计数最简单,但边界问题明显——比如限制每分钟 100 次,用户在 0:59 发 100 次,1:00 又发 100 次,实际两秒内发了 200 次。Agent 场景下这种突发很容易打垮后端。
滑动窗口更平滑,用 ZSET 实现:
import time def sliding_window_limit(user_id, limit=100, window=60): """滑动窗口限流,返回 True 表示允许""" key = f"agent:ratelimit:{user_id}" now = time.time() pipe = r.pipeline() # 移除窗口外的记录 pipe.zremrangebyscore(key, 0, now - window) # 统计窗口内请求数 pipe.zcard(key) # 添加当前请求 pipe.zadd(key, {str(now): now}) # 设置过期,避免冷用户占用内存 pipe.expire(key, window) results = pipe.execute() count = results[1] return count < limit这个实现每次请求要执行 4 个命令,用 pipeline 打包后一次网络往返,性能可以接受。如果限流粒度更粗(比如全局 QPS),可以用令牌桶,用 Lua 脚本在 Redis 里维护令牌数和上次填充时间,原子地取令牌。
提示:限流的 key 一定要设过期时间。我见过有人忘了设,结果每个用户都留一个 ZSET,几个月后内存爆了。滑动窗口的 key 过期时间设成窗口大小就行,窗口外的数据反正也没用。
4.3 并发下的会话一致性:读改写竞态怎么破
Agent 处理一轮对话,通常要经历“读会话 → 调模型 → 调工具 → 写会话”这个过程。如果同一个会话有两个请求并发进来(比如用户手快连发两条消息),就会出现读改写竞态:两个请求都读到旧状态,各自处理后写回,后写的覆盖先写的,一条消息就丢了。
解决办法有两个思路。一是串行化,用分布式锁把同一会话的请求串起来:
def handle_message(session_id, message): lock_token = acquire_lock(f"session:{session_id}", timeout=30) if not lock_token: raise Exception("会话正忙,请稍后重试") try: session = load_session(session_id) # ... 处理逻辑 save_session(session_id, session) finally: release_lock(f"session:{session_id}", lock_token)二是乐观锁,用版本号或 WATCH/MULTI 实现 CAS:
def save_session_optimistic(session_id, session, expected_version): """乐观锁保存,版本不匹配则失败""" key = f"agent:session:{session_id}:meta" with r.pipeline() as pipe: try: pipe.watch(key) current = pipe.hget(key, "version") if current and int(current) != expected_version: pipe.unwatch() return False pipe.multi() pipe.hset(key, mapping={"version": expected_version + 1, "updated_at": int(time.time())}) pipe.execute() return True except redis.WatchError: return False串行化简单粗暴但会降低并发度,乐观锁并发度高但需要重试逻辑。我的经验是:会话级别的操作,用串行化更省心,因为 Agent 处理本身就不快,串行化的性能损失可以接受;而工具缓存写入这种高频操作,用乐观锁或无锁设计更合适。
5. 缓存治理:上线之后才是真正的考验
5.1 内存监控与淘汰策略:别等 OOM 了才想起来看
Redis 内存爆掉是 Agent 服务最常见的线上事故之一。工具缓存、会话数据、限流记录,样样都吃内存,不监控就是定时炸弹。
首先要设置maxmemory和淘汰策略。Agent 场景我推荐allkeys-lru或volatile-lru:
maxmemory 4gb maxmemory-policy allkeys-lruallkeys-lru对所有 key 做 LRU 淘汰,适合缓存和会话混存的场景;volatile-lru只淘汰设了过期时间的 key,如果你把重要数据设了持久化(不设 TTL),用这个更安全。千万别用 noeviction,内存满了之后所有写操作直接报错,Agent 会全面瘫痪。
监控方面,这几个指标必须盯:
| 指标 | 命令 | 关注点 |
|---|---|---|
| 内存使用 | INFO memory | used_memory 接近 maxmemory 就要警惕 |
| 命中率 | INFO stats | hit_rate 低于 80% 说明缓存设计有问题 |
| 淘汰数 | INFO stats | evicted_keys 持续增长说明内存不够 |
| 慢查询 | SLOWLOG GET | 超过 10ms 的命令要优化 |
| 连接数 | INFO clients | connected_clients 突增可能是连接泄漏 |
我一般会把这些指标接到监控面板上,设好告警阈值。特别是evicted_keys,如果它开始持续增长,说明缓存正在被大量淘汰,命中率会下降,后端压力会上升,得赶紧扩容或优化 key 设计。
5.2 大 key 与热 key:Agent 缓存里的隐形杀手
大 key是指 value 特别大的 key,比如一个存了几万条消息的 List,或者一个几 MB 的 JSON String。大 key 的危害在于:读写慢、网络传输慢、删除时会阻塞 Redis(Redis 4.0 之前用DEL删大 key 会卡住,之后可以用UNLINK异步删除)。
排查大 key 用redis-cli --bigkeys,或者用MEMORY USAGE key看单个 key 的内存占用。Agent 场景下最容易出大 key 的地方就是会话消息列表,一定要用LTRIM控制长度,别让它无限增长。
热 key是指访问频率极高的 key,比如某个热门问题的缓存。热 key 会导致单个 Redis 节点 CPU 飙升,成为瓶颈。排查热 key 可以用redis-cli --hotkeys(需要 maxmemory-policy 是 lru 类),或者自己在客户端埋点统计。
热 key 的解决办法有几种:本地缓存(在应用进程内再缓存一份,减少 Redis 访问)、key 拆分(把hotkey拆成hotkey:1、hotkey:2分散到不同节点)、读写分离(热 key 走从节点读)。Agent 场景下,我常用本地缓存 + 短 TTL 的组合,比如热门工具结果在进程内缓存 5 秒,能挡掉大部分重复请求。
5.3 缓存与数据库的一致性:Agent 记忆的持久化策略
Redis 是内存数据库,虽然可以持久化(RDB/AOF),但不能把 Redis 当作唯一的数据存储。Agent 的会话记忆、用户偏好这些重要数据,必须有数据库兜底。
我的做法是Cache-Aside 模式:读的时候先查 Redis,没有再查数据库并回填;写的时候先写数据库,再删除 Redis 缓存(注意是删除不是更新,避免并发写导致脏数据)。
def get_session(session_id): # 先查缓存 cached = r.get(f"agent:session:{session_id}") if cached: return json.loads(cached) # 缓存未命中,查数据库 session = db.query_session(session_id) if session: r.setex(f"agent:session:{session_id}", 3600, json.dumps(session)) return session def update_session(session_id, data): # 先写数据库 db.update_session(session_id, data) # 再删缓存 r.delete(f"agent:session:{session_id}")这里有个经典问题:先删缓存还是先写数据库。两种顺序都有并发问题,业界没有完美方案。我的选择是“先写库再删缓存”,因为这种顺序下最坏情况是缓存里短暂存在旧数据,但不会出现“缓存已删、库还没写”导致读到旧库数据的更糟情况。如果业务对一致性要求极高,可以用延迟双删:写库后删一次缓存,延迟几百毫秒再删一次,覆盖掉可能的并发读回填。
对于 Agent 的长期记忆(比如用户画像、历史偏好),我建议异步持久化:会话过程中数据主要在 Redis 里流转,定期(比如每 5 分钟或会话结束时)批量同步到数据库。这样既保证了性能,又不会丢数据。
6. 几个我踩过的坑和对应的解法
6.1 序列化选型:JSON 不是万能的
一开始我所有缓存都用 JSON 序列化,简单直观。但后来发现两个问题:一是性能,JSON 序列化/反序列化在数据量大时很吃 CPU;二是类型丢失,JSON 不支持二进制、日期等类型,存进去再取出来就变样了。
后来我根据数据类型分开处理:
- 会话元信息:用 Hash 存储,字段直接是字符串,不需要序列化。
- 工具结果:如果结构简单,用 JSON;如果包含二进制或复杂类型,用MessagePack或Pickle(Python 场景)。
- 大对象:考虑压缩后再存,比如 gzip + base64,牺牲一点 CPU 换内存。
import msgpack def set_msgpack(key, value, ttl=3600): packed = msgpack.packb(value, use_bin_type=True) r.setex(key, ttl, packed) def get_msgpack(key): data = r.get(key) if data: return msgpack.unpackb(data, raw=False) return NoneMessagePack 比 JSON 快 2-5 倍,体积小 30% 左右,在 Agent 这种高频读写的场景下收益明显。但可读性差,调试时得用工具解码,所以我在开发环境用 JSON,生产环境用 MessagePack,通过配置切换。
6.2 连接池配置:别让连接数成为瓶颈
Redis 客户端连接池配置不当,会导致连接数暴涨或请求排队。Python 的 redis-py 默认连接池没有上限,高并发下会创建大量连接,把 Redis 的maxclients撑爆。
我的配置经验:
pool = redis.ConnectionPool( host='localhost', port=6379, max_connections=50, # 根据并发量调整 socket_timeout=2, # 读写超时,别设太长 socket_connect_timeout=1, # 连接超时 retry_on_timeout=True, # 超时重试 health_check_interval=30 # 定期健康检查 ) r = redis.Redis(connection_pool=pool)max_connections设多少合适?一个经验公式是并发请求数 × 1.5。比如你的 Agent 服务峰值 100 并发,每个请求平均持有连接 10ms,那理论上 100 × 0.01 = 1 个连接就够,但考虑到突发和阻塞,设 50-100 比较稳妥。设太小会排队,设太大浪费资源且可能触发 Redis 的连接数限制。
socket_timeout特别重要。Agent 调用 Redis 如果卡住,没有超时的话线程会一直挂着,最终拖垮整个服务。我一般设 2 秒,超过就报错走降级逻辑。
6.3 缓存失效时的降级:Redis 挂了 Agent 还能不能跑
这个问题很现实:Redis 是单点(或者主从),万一挂了,Agent 服务是直接不可用,还是能降级运行?
我的设计原则是Redis 故障时核心功能可用,非核心功能降级:
- 会话数据:Redis 挂了,降级到数据库读写,性能下降但功能可用。
- 工具缓存:Redis 挂了,直接跳过缓存调用真实工具,慢但能用。
- 限流:Redis 挂了,降级到本地限流(单机限流),虽然不精确但能防止打垮后端。
- 分布式锁:Redis 挂了,降级到数据库锁或直接放行(根据业务容忍度)。
实现上就是包一层 try-except,捕获 Redis 异常后走降级逻辑:
def safe_redis_call(func, fallback, *args, **kwargs): try: return func(*args, **kwargs) except (redis.ConnectionError, redis.TimeoutError) as e: logger.warning(f"Redis 调用失败,走降级: {e}") return fallback(*args, **kwargs)这个降级逻辑一定要在上线前就测试过,别等真出故障了才发现降级代码有 bug。我一般会用工具模拟 Redis 不可用,跑一遍全流程,确保降级路径是通的。
6.4 那些年我误用的 Redis 命令
最后分享几个命令层面的坑:
KEYS *:生产环境绝对禁用。它会阻塞 Redis 直到遍历完所有 key,key 多了直接卡死。要用SCAN游标分批遍历:
def scan_keys(pattern, batch=100): cursor = 0 while True: cursor, keys = r.scan(cursor, match=pattern, count=batch) for key in keys: yield key if cursor == 0: breakDEL大 key:删除大 key 会阻塞,用UNLINK异步删除。
FLUSHALL:手滑执行一次,整个 Redis 清空。生产环境应该用rename-command禁用或重命名这个命令:
rename-command FLUSHALL "" rename-command FLUSHDB "" rename-command KEYS ""HGETALL大 Hash:如果 Hash 字段很多,HGETALL会一次性拉取所有字段,阻塞且占带宽。用HSCAN分批读,或者只读需要的字段HMGET。
这些坑我都真实踩过,尤其是KEYS,当年在测试环境跑了一次,Redis 卡了十几秒,所有请求超时。从那以后我在配置文件里直接把KEYS命令禁了,逼着自己用SCAN。
7. 关于 Agent 缓存治理的一点个人体会
做了几个 Agent 项目之后,我越来越觉得 Redis 在 Agent 架构里的角色被低估了。很多人把它当成一个简单的“加速层”,但实际上它承担的是状态协调的职责——多实例之间怎么共享会话、怎么协调工具调用、怎么控制并发,这些问题的答案都在 Redis 里。
我的核心体会是:缓存设计要从业务的一致性要求出发,而不是从技术便利出发。哪些数据可以容忍短暂不一致,哪些必须强一致,哪些可以异步持久化,想清楚这些,再决定用什么数据结构、设什么 TTL、要不要加锁。技术选型是结果,不是起点。
另外一个教训是监控和治理要前置。别等线上出问题了才去查大 key、热 key、内存使用。上线前就把监控搭好,把淘汰策略配好,把危险命令禁掉,这些准备工作花不了多少时间,但能省掉后面无数个加班的夜晚。
最后说个具体的:Agent 的会话数据,我现在的做法是Redis 存热数据 + 数据库存全量 + 定期归档冷数据,三层配合。Redis 里只保留最近 20 轮对话和会话元信息,TTL 设 2 小时;数据库存完整历史;超过 30 天的数据归档到对象存储。这样 Redis 内存可控,查询性能也好,数据也不会丢。这套方案在几个项目里跑下来,单实例支撑几千并发没什么压力,内存占用也稳定在可控范围内。