1. 为什么 AI Agent 需要 Redis 缓存1.1 从一次线上事故说起去年冬天我负责的一个 AI Agent 项目在凌晨两点突然告警。用户反馈对话响应时间从平均 1.2 秒飙升到 8 秒以上部分请求直接超时。排查后发现Agent 在处理多轮对话时每次都要重新读取用户的历史会话、工具调用结果和知识库检索片段而这些数据全部来自磁盘上的向量数据库和关系型数据库。并发一上来数据库连接池瞬间被打满整个服务雪崩。那次事故之后我花了三天时间给 Agent 加了一层 Redis 缓存。改造完成后同样的并发压力下P99 响应时间从 8 秒降到了 400 毫秒以内数据库 QPS 下降了 70%。这篇文章就把这套缓存方案完整拆解一遍包括架构设计、数据结构选型、代码实现和踩过的坑。如果你正在搭建 AI Agent或者已经上线但被性能问题困扰这篇文章应该能帮你少走不少弯路。即使你之前没怎么用过 Redis我也会把关键概念用生活化的方式讲清楚。1.2 AI Agent 的缓存需求到底特殊在哪传统 Web 应用的缓存逻辑相对简单查数据库、写缓存、设过期时间。但 AI Agent 不一样它的缓存需求有几个鲜明特点。第一数据形态多样。Agent 需要缓存的不只是用户信息还有对话历史、工具调用结果、向量检索片段、模型推理的中间状态、甚至整个 Agent 的执行计划。这些数据的结构、大小、生命周期完全不同。第二读写模式复杂。对话历史是典型的读多写多每轮对话都要追加新消息工具调用结果往往是写一次读多次而模型推理的中间状态可能是写一次读一次就丢弃。第三一致性要求分层。用户余额、订单状态这类数据绝对不能脏读但对话历史、检索片段稍微旧一点完全可以接受。这就要求缓存策略不能一刀切。第四并发压力集中。一个 Agent 任务可能触发十几次工具调用和模型推理每次都要读写状态。如果不做缓存数据库和向量库会被反复冲击。理解了这些特点才能设计出真正好用的缓存方案。下面我从整体架构开始拆解。2. 整体架构设计与选型思路2.1 缓存分层L1 本地缓存 L2 Redis我的方案是两层缓存。L1 用进程内的本地缓存比如 Python 的cachetools或 Java 的 Caffeine存那些变化极少、访问极频繁的数据比如 Agent 的配置信息、工具描述、系统提示词。L2 用 Redis存对话历史、工具结果、检索片段这些需要跨实例共享的数据。为什么不全用 Redis因为本地缓存没有网络开销纳秒级访问对于配置类数据性价比极高。但本地缓存有个致命问题多实例部署时数据不一致。所以只放那些改了也不影响正确性或者通过版本号能感知变更的数据。为什么不全用本地缓存因为 Agent 通常是无状态部署多个实例需要共享会话状态。用户第一次请求打到实例 A第二次打到实例 B如果会话只存在本地实例 B 就找不到上下文了。提示本地缓存的容量要设上限否则 Agent 跑久了内存会爆。我一般设 1000 条用 LRU 淘汰。2.2 Redis 部署模式怎么选Redis 的部署模式主要有单机、主从、哨兵、集群四种。对于 AI Agent 场景我的建议是部署模式适用场景优点缺点单机开发测试、小流量简单无高可用主从读多写少读写分离故障切换需人工哨兵生产环境中小规模自动故障切换配置稍复杂集群大规模、数据量大水平扩展运维复杂我自己的项目用的是哨兵模式一主两从三哨兵足够支撑日均百万级请求。如果你们的 Agent 还在早期单机加定期备份也能扛一阵但上线前一定要换成哨兵或集群。Docker 部署主从的配置我贴一下这是最常用的方式# 主节点 docker run -d --name redis-master \ -p 6379:6379 \ -v /data/redis/master:/data \ redis:7.2 redis-server --appendonly yes --requirepass yourpassword # 从节点 docker run -d --name redis-slave \ -p 6380:6379 \ -v /data/redis/slave:/data \ redis:7.2 redis-server --appendonly yes --requirepass yourpassword \ --slaveof redis-master 6379 --masterauth yourpasswordappendonly yes开启 AOF 持久化保证重启后数据不丢。requirepass设密码千万别裸奔。2.3 客户端选型Lettuce 还是 JedisJava 生态里主流是 Lettuce 和 Jedis。Lettuce 基于 Netty支持异步和响应式线程安全适合高并发Jedis 是同步阻塞每个线程一个连接简单但连接开销大。AI Agent 场景下我推荐 Lettuce因为 Agent 的调用链经常是异步的Lettuce 的RedisAsyncCommands能很好地配合。Python 生态用redis-py就行异步场景用redis.asyncio。有个坑要注意Lettuce 默认超时是 60 秒Agent 场景下这个太长了。我一般设成 2 秒配合重试机制。之前遇到过 Redis 网络抖动因为超时太长请求全部堆积最后线程池耗尽。改成 2 秒超时加两次重试后抖动时最多损失少量请求不会拖垮整个服务。3. 核心数据结构与缓存内容设计3.1 对话历史用什么结构存对话历史是 Agent 最核心的缓存内容。每条消息包含角色user/assistant/tool、内容、时间戳、工具调用 ID 等字段。我试过三种方案方案一String 存 JSON 数组。每次追加消息都要读出整个数组、反序列化、追加、再序列化写回。消息多了之后这个操作是 O(n)性能很差。方案二List 存消息。用RPUSH追加LRANGE读取。追加是 O(1)读取指定范围也很快。但问题是 List 不能单独更新某条消息而且取出来后还要反序列化每条消息。方案三Sorted Set 存消息。score 用时间戳member 存消息 JSON。追加用ZADD读取用ZRANGEBYSCORE。好处是可以按时间范围查询也支持分页。缺点是 member 不能重复如果两条消息内容完全一样会覆盖。最终我选了 List因为对话历史就是严格的追加和顺序读取List 最贴合。为了控制长度每次追加后用LTRIM保留最近 N 条import redis import json r redis.Redis(hostlocalhost, port6379, passwordyourpassword, decode_responsesTrue) def append_message(session_id, message, max_len50): key fagent:session:{session_id}:messages pipe r.pipeline() pipe.rpush(key, json.dumps(message, ensure_asciiFalse)) pipe.ltrim(key, -max_len, -1) pipe.expire(key, 3600) # 1小时过期 pipe.execute() def get_messages(session_id, count20): key fagent:session:{session_id}:messages raw r.lrange(key, -count, -1) return [json.loads(m) for m in raw]max_len50是我根据经验设的保留最近 50 条消息。太长了浪费内存太短了 Agent 会丢失上下文。你们可以根据模型上下文窗口调整。注意LTRIM和RPUSH放在同一个 pipeline 里保证原子性。否则并发追加时可能一个请求刚 push 完还没 trim另一个请求就读到了超长列表。3.2 工具调用结果怎么缓存Agent 调用工具比如查天气、搜网页、查数据库往往耗时较长而且同样的参数可能被重复调用。这类结果非常适合缓存。我用 String 存 JSONkey 是工具名加参数哈希import hashlib def cache_tool_result(tool_name, params, result, ttl300): param_str json.dumps(params, sort_keysTrue, ensure_asciiFalse) param_hash hashlib.md5(param_str.encode()).hexdigest() key fagent:tool:{tool_name}:{param_hash} r.setex(key, ttl, json.dumps(result, ensure_asciiFalse)) def get_tool_result(tool_name, params): param_str json.dumps(params, sort_keysTrue, ensure_asciiFalse) param_hash hashlib.md5(param_str.encode()).hexdigest() key fagent:tool:{tool_name}:{param_hash} cached r.get(key) return json.loads(cached) if cached else Nonesort_keysTrue很关键保证参数顺序不同但内容相同的请求能命中同一个缓存。TTL 设 300 秒是因为工具结果通常有时效性天气数据 5 分钟前的还能用但股票价格就不行了。不同工具要设不同 TTL这个后面细说。3.3 向量检索片段怎么缓存RAG 场景下Agent 每次都要把用户问题向量化然后去向量库检索。向量化本身要调模型检索也要时间。如果同样的问题被反复问缓存能省不少事。我用 String 存检索结果key 是问题文本的哈希def cache_retrieval(query, docs, ttl1800): query_hash hashlib.sha256(query.encode()).hexdigest()[:16] key fagent:retrieval:{query_hash} r.setex(key, ttl, json.dumps(docs, ensure_asciiFalse)) def get_retrieval(query): query_hash hashlib.sha256(query.encode()).hexdigest()[:16] key fagent:retrieval:{query_hash} cached r.get(key) return json.loads(cached) if cached else None这里用 SHA256 而不是 MD5因为问题文本可能较长SHA256 碰撞概率更低。截取前 16 位是为了控制 key 长度实际碰撞概率依然极低。TTL 设 1800 秒是因为知识库更新频率通常不高半小时内的检索结果基本可信。但如果你们的知识库是实时更新的这个值要调小或者用版本号做 key 的一部分。3.4 分布式锁保护关键操作Agent 有些操作必须串行比如更新用户余额、写入订单。这时候要用 Redis 分布式锁。import uuid import time def acquire_lock(lock_name, timeout10): token str(uuid.uuid4()) end time.time() timeout while time.time() end: if r.set(fagent:lock:{lock_name}, token, nxTrue, extimeout): return token time.sleep(0.01) return None def release_lock(lock_name, token): lua_script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end r.eval(lua_script, 1, fagent:lock:{lock_name}, token)释放锁必须用 Lua 脚本保证判断 token 是否匹配和删除 key是原子的。否则可能出现A 的锁过期了B 拿到锁A 执行完释放锁把 B 的锁删了。这个坑我踩过当时排查了半天。4. 缓存策略与失效治理4.1 TTL 设置的经验法则TTL 设太长数据陈旧设太短缓存命中率低。我的经验是按数据变化频率分档数据类型TTL理由Agent 配置3600s几乎不变对话历史3600s会话通常一小时内结束工具结果天气300s5分钟内可信工具结果股票10s秒级变化检索片段1800s知识库更新不频繁用户余额60s变化频繁但可短暂容忍关键是给每类数据单独设 TTL不要全局一个值。我见过有人所有缓存都设 300 秒结果股票数据陈旧导致 Agent 给出错误建议。4.2 缓存穿透、击穿、雪崩怎么防这三个是缓存经典问题AI Agent 场景下同样会遇到。缓存穿透查询一个不存在的数据缓存和数据库都没有每次请求都打到数据库。Agent 场景下用户可能问一个知识库里完全没有的问题每次都穿透。解决方案是缓存空值def get_with_null_cache(key, loader, ttl300, null_ttl60): cached r.get(key) if cached is not None: return json.loads(cached) if cached ! __NULL__ else None data loader() if data is None: r.setex(key, null_ttl, __NULL__) else: r.setex(key, ttl, json.dumps(data, ensure_asciiFalse)) return data空值 TTL 设短一点60 秒因为数据可能很快就被创建了。缓存击穿某个热点 key 过期瞬间大量请求同时打到数据库。Agent 场景下一个热门问题突然被很多人问就会击穿。解决方案是加互斥锁只让一个请求去加载def get_with_mutex(key, loader, ttl300): cached r.get(key) if cached: return json.loads(cached) lock_token acquire_lock(fload:{key}, timeout5) if lock_token: try: cached r.get(key) if cached: return json.loads(cached) data loader() r.setex(key, ttl, json.dumps(data, ensure_asciiFalse)) return data finally: release_lock(fload:{key}, lock_token) else: time.sleep(0.1) return get_with_mutex(key, loader, ttl)双重检查很关键拿到锁后再查一次缓存因为可能在等锁期间别的线程已经加载好了。缓存雪崩大量 key 同时过期请求全部打到数据库。解决方案是给 TTL 加随机抖动import random def setex_with_jitter(key, ttl, value): jitter random.randint(0, int(ttl * 0.1)) r.setex(key, ttl jitter, value)抖动范围设 TTL 的 10% 左右既打散了过期时间又不会让数据太早或太晚过期。4.3 缓存一致性怎么保证Agent 更新数据时是先更新数据库还是先更新缓存这是经典难题。我的策略是先更新数据库再删除缓存而不是更新缓存。原因是更新缓存可能写入一个中间状态的值而删除缓存下次读取时会从数据库加载最新值。但删除缓存也有并发问题请求 A 更新数据库后删除缓存请求 B 在 A 删除前读到旧缓存并写回。解决方案是延迟双删def update_with_double_delete(key, update_func, delay0.5): update_func() # 更新数据库 r.delete(key) # 第一次删除 time.sleep(delay) r.delete(key) # 延迟再删一次延迟时间设 0.5 秒足够覆盖大多数并发读的窗口。如果对一致性要求极高可以用 binlog 订阅的方式但那就复杂了Agent 场景下延迟双删够用。5. 实操过程与性能调优5.1 从零搭建缓存层的完整步骤假设你有一个基于 FastAPI 的 Agent 服务现在要加 Redis 缓存。完整步骤如下。第一步安装依赖。pip install redis5.0.1 hiredis2.2.3hiredis是 C 实现的解析器能显著提升 Redis 客户端性能实测吞吐量提升 30% 以上。第二步配置连接池。import redis from redis.connection import ConnectionPool pool ConnectionPool( hostlocalhost, port6379, passwordyourpassword, db0, max_connections50, socket_timeout2, socket_connect_timeout1, retry_on_timeoutTrue, health_check_interval30 ) r redis.Redis(connection_poolpool, decode_responsesTrue)max_connections50要根据服务并发量设太小会排队太大会浪费。一般设成最大并发数的 1.5 倍。health_check_interval30每 30 秒检查一次连接健康度避免用到已断开的连接。第三步封装缓存工具类。class AgentCache: def __init__(self, redis_client): self.r redis_client def get_json(self, key): val self.r.get(key) return json.loads(val) if val else None def set_json(self, key, value, ttl300): jitter random.randint(0, int(ttl * 0.1)) self.r.setex(key, ttl jitter, json.dumps(value, ensure_asciiFalse)) def delete(self, key): self.r.delete(key) def get_or_load(self, key, loader, ttl300): cached self.get_json(key) if cached is not None: return cached data loader() if data is not None: self.set_json(key, data, ttl) return data第四步在 Agent 关键路径接入缓存。对话历史读写、工具调用、检索结果都走AgentCache。改造时要注意缓存只是加速不能改变业务逻辑。所有缓存读取失败都要能回退到原始数据源。5.2 性能压测与调优记录改造完成后我做了压测用locust模拟 200 并发用户每个用户发起 10 轮对话。改造前QPS 45P99 响应 8200ms数据库连接池频繁打满。改造后QPS 380P99 响应 380ms数据库 QPS 从 1200 降到 350。关键调优点有三个一是 pipeline 批量操作。原来追加消息要 3 次 Redis 往返rpush、ltrim、expire改成 pipeline 后变成 1 次往返延迟降低 60%。二是连接池预热。服务启动时先创建 10 个连接避免冷启动时大量请求同时建连。三是大 key 拆分。有个用户的对话历史超过 5000 条单个 List 达到 2MB读取很慢。后来改成按会话分段每 100 条一个 key读取时按需加载。提示Redis 单 key 超过 10KB 就要警惕超过 1MB 基本要拆分。用redis-cli --bigkeys可以扫描大 key。5.3 监控指标怎么设缓存上线后必须监控否则出问题都不知道。我关注这几个指标命中率keyspace_hits / (keyspace_hits keyspace_misses)低于 80% 要排查。内存使用used_memory超过 maxmemory 的 80% 要告警。慢查询slowlog里超过 10ms 的命令要关注。连接数connected_clients接近 max_connections 要扩容。淘汰数evicted_keys持续大于 0 说明内存不够。redis-cli info stats | grep keyspace redis-cli info memory | grep used_memory redis-cli slowlog get 10这些命令我一般写成定时脚本每 5 分钟采集一次推到监控系统。6. 常见问题与排查技巧实录6.1 连接超时问题排查现象日志里频繁出现RedisCommandTimeoutException响应时间抖动。排查思路先看 Redis 服务端有没有慢查询redis-cli slowlog get 10。如果服务端正常那就是客户端或网络问题。客户端方面检查连接池是否耗尽。redis-cli info clients看connected_clients如果接近maxclients说明连接不够用。网络方面用redis-cli --latency测延迟正常应该在 1ms 以内。如果超过 10ms可能是网络抖动或 Redis 负载过高。我之前遇到过一次原因是 Agent 有个工具调用会执行KEYS *命令在数据量大时阻塞了 Redis。后来改成SCAN分批遍历问题解决。注意生产环境绝对禁止使用KEYS、FLUSHALL、FLUSHDB这些危险命令。可以在配置文件里用rename-command重命名它们。6.2 内存暴涨怎么处理现象Redis 内存持续增长触发淘汰甚至 OOM。排查思路先用redis-cli --bigkeys找大 key再用redis-cli memory usage key看具体 key 的内存占用。常见原因有三个一是对话历史没设上限越积越多二是工具结果缓存了超大响应比如整个网页 HTML三是 key 没有设 TTL永久驻留。解决方案对话历史用LTRIM限制长度工具结果超过 100KB 的不缓存或者压缩后缓存所有缓存 key 必须设 TTL用redis-cli --scan --pattern agent:* | head抽查。我一般会设maxmemory-policy allkeys-lru内存满了自动淘汰最久未使用的 key。但这是兜底不能依赖它该设的 TTL 还是要设。6.3 缓存与数据库不一致现象用户看到的数据和数据库里的不一致刷新后又对了。排查思路先确认是不是缓存没删干净。检查更新逻辑是不是先更新数据库再删缓存有没有漏删的 key。再看是不是并发导致。用MONITOR命令实时看 Redis 收到的命令能找到异常写入。如果是延迟双删的延迟时间不够可以适当加大。但根本上如果业务对一致性要求极高就不要用缓存或者用版本号机制每次更新数据库时版本号加一缓存 key 带上版本号旧版本自然失效。6.4 常见问题速查表问题可能原因排查命令解决方案连接超时连接池耗尽/慢查询info clients、slowlog get扩容连接池、优化慢命令内存暴涨大 key/无 TTL--bigkeys、memory usage拆分大 key、设 TTL命中率低TTL 太短/key 设计不合理info stats调整 TTL、优化 key数据不一致缓存未删/并发写MONITOR延迟双删、版本号主从延迟网络/从库负载info replication检查网络、扩容从库6.5 几个我踩过的坑坑一用decode_responsesTrue后忘了处理 bytes。有些场景下还是返回 bytes比如hgetall的 field。统一用decode_responsesTrue能避免大部分问题但要注意二进制数据不能这么用。坑二pipeline 里混用了需要立即返回结果的命令。pipeline 是批量发送所有命令的返回值要等execute()才拿到。如果在 pipeline 中间需要根据前一个命令的结果决定下一个命令就不能用 pipeline。坑三分布式锁的过期时间设太短。业务还没执行完锁就过期了导致并发问题。锁的过期时间要大于业务最大执行时间同时业务里要有续期机制。坑四缓存 key 没有统一前缀。多个项目共用一个 Redis 时key 冲突了。所有 key 都要带项目前缀比如agent:。坑五忘了处理缓存序列化。Python 的json.dumps默认不支持datetime、Decimal等类型要自定义default函数。我一般统一转成字符串读取时再转回来。这套缓存方案在我自己的 Agent 项目里跑了半年多日均处理百万级请求稳定性还不错。核心经验就是分层缓存、按需设 TTL、做好监控、留好降级。缓存不是银弹它解决的是性能问题不是正确性问题。任何缓存失效时业务都要能回退到原始数据源正常工作。