1. 为什么 AI Agent 需要 Redis 缓存1.1 从一次线上事故说起去年冬天我负责的一个 AI Agent 项目在凌晨两点突然告警。用户反馈对话响应时间从平均 1.2 秒飙升到 8 秒以上部分请求直接超时。排查后发现Agent 在处理多轮对话时每一轮都要重新从向量数据库拉取历史上下文、重新计算用户意图、重新调用大模型做推理。并发量一上来后端服务像被掐住脖子一样喘不过气。这个场景其实很典型。AI Agent 和普通 Web 应用最大的区别在于它的每一次“思考”都伴随着昂贵的计算和外部 API 调用。大模型推理按 token 计费向量检索消耗 CPU 和内存工具调用可能涉及第三方接口。如果每个请求都从头走一遍完整链路成本和延迟都会失控。Redis 在这里扮演的角色就是 Agent 的“短期记忆”和“高速缓存层”。它把那些频繁访问、变化不频繁、计算代价高的数据暂存起来让 Agent 在毫秒级拿到结果而不是每次都去“翻箱倒柜”。1.2 缓存到底缓存什么很多人一提到 Redis 缓存第一反应就是缓存数据库查询结果。但在 AI Agent 场景下需要缓存的东西要丰富得多。我把它分成四类第一类是会话上下文。多轮对话中用户的历史消息、Agent 的回复、当前对话状态这些数据如果每次都从关系型数据库读取延迟高且数据库压力大。Redis 的 Hash 结构非常适合存储会话级别的键值对一个session:{id}的 Hash 就能装下整轮对话的上下文。第二类是向量检索结果。RAG 架构中用户问题经过 Embedding 后去向量库检索相似文档。同一个问题或相似问题在短时间内可能被多次问到把检索结果缓存下来能省掉大量向量计算和 IO。第三类是大模型推理结果。对于确定性较强的问答场景相同输入的大模型输出可以缓存。虽然大模型有随机性但在 temperature 设为 0 或极低时缓存命中率相当可观。第四类是工具调用结果。Agent 调用天气 API、搜索 API、数据库查询等这些外部依赖的返回结果在短时间内是稳定的缓存起来能显著降低外部依赖故障带来的影响。1.3 不缓存的代价有多大我做过一个粗略的测算。假设一个 Agent 服务每天处理 10 万次请求其中 40% 是重复或高度相似的问题。如果不做缓存这 4 万次请求每次都要走完整的 Embedding 向量检索 大模型推理链路。按每次推理 0.01 元、检索 0.005 元计算一天多花 600 元一个月就是 1.8 万。这还只是直接成本没算上延迟带来的用户体验损失和服务器扩容成本。缓存命中后响应时间从秒级降到毫秒级用户感知到的就是“这个 Agent 反应真快”。在竞争激烈的 AI 应用市场响应速度往往就是留存率。注意缓存不是银弹。缓存穿透、缓存雪崩、缓存击穿这三个经典问题在 Agent 场景下同样存在而且因为数据结构的特殊性处理方式需要调整。后面会专门讲。2. Redis 在 AI Agent 中的核心数据结构选型2.1 String最简单的缓存载体String 是 Redis 最基础的数据类型适合缓存序列化后的 JSON 对象。比如把整个 Agent 配置、单次工具调用结果序列化成 JSON 字符串存进去。import redis import json r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) # 缓存工具调用结果 tool_result {weather: sunny, temperature: 25, city: Beijing} r.setex(tool:weather:beijing, 300, json.dumps(tool_result)) # 读取 cached r.get(tool:weather:beijing) if cached: data json.loads(cached)String 的优点是简单直接缺点是更新部分字段时需要整体重写。对于结构复杂、字段频繁单独更新的数据Hash 更合适。2.2 Hash会话上下文的最佳拍档Hash 适合存储对象的多个字段而且可以单独更新某个字段而不影响其他字段。会话上下文就是典型场景。# 存储会话上下文 session_id sess_12345 r.hset(fsession:{session_id}, mapping{ user_id: u_001, last_message: 帮我查一下明天天气, intent: weather_query, context: json.dumps([历史消息1, 历史消息2]), updated_at: 2025-01-15T10:30:00 }) # 只更新最后一条消息 r.hset(fsession:{session_id}, last_message, 那后天呢) # 设置过期时间 r.expire(fsession:{session_id}, 1800)Hash 的另一个好处是内存利用率高。当 Hash 字段较少时Redis 会用 ziplist 编码内存占用比多个 String 键小得多。2.3 List消息队列与对话历史List 可以做简单的消息队列也可以存储对话历史。用LPUSH插入新消息LRANGE读取最近 N 条。# 追加对话消息 r.lpush(fchat:{session_id}, json.dumps({role: user, content: 你好})) r.lpush(fchat:{session_id}, json.dumps({role: assistant, content: 你好有什么可以帮你})) # 读取最近 10 条 history r.lrange(fchat:{session_id}, 0, 9)但要注意List 做消息队列时没有 ACK 机制消息可能丢失。如果 Agent 的任务队列要求可靠性建议用 Redis Stream 或者专业消息队列。2.4 Set 与 ZSet去重与排序Set 适合做去重比如记录用户已经看过的文档 ID避免重复推荐。ZSet 适合做排序比如按时间戳排序的对话摘要、按相关性排序的检索结果。# 记录用户已读文档 r.sadd(fuser:{user_id}:read_docs, doc_001, doc_002) # 检查是否已读 is_read r.sismember(fuser:{user_id}:read_docs, doc_001) # ZSet 按时间排序的对话摘要 r.zadd(fuser:{user_id}:summaries, {summary_1: 1700000000, summary_2: 1700000100})2.5 数据结构选型对照表数据类型适用场景优点注意事项String工具调用结果、配置项、简单 KV简单、通用部分更新需整体重写Hash会话上下文、对象属性字段独立更新、内存高效字段过多时性能下降List对话历史、简单队列有序、插入快无 ACK可靠性有限Set去重、标签去重、集合运算无序ZSet排序、优先级队列自动排序、范围查询内存占用相对较高Stream可靠消息队列持久化、ACK结构复杂学习成本高选型时不要追求“最先进”而是看业务需求。会话上下文用 Hash工具结果用 String对话历史用 List大多数场景就够用了。3. 缓存策略设计与实现细节3.1 缓存粒度粗粒度还是细粒度缓存粒度决定了缓存的灵活性和命中率。粗粒度缓存整个响应命中率高但更新麻烦细粒度缓存单个字段更新灵活但命中率可能下降。我的经验是会话上下文用粗粒度工具结果用细粒度。会话上下文整体变化频繁细粒度更新反而增加复杂度工具结果相对独立按工具名和参数哈希做键细粒度缓存更合适。import hashlib def cache_key_for_tool(tool_name, params): param_str json.dumps(params, sort_keysTrue) param_hash hashlib.md5(param_str.encode()).hexdigest()[:8] return ftool:{tool_name}:{param_hash}3.2 过期时间TTL 设置的艺术TTL 设置太短缓存频繁失效等于没缓存设置太长数据陈旧用户拿到过期信息。不同数据的 TTL 策略应该不同会话上下文30 分钟到 2 小时。用户对话通常不会间隔太久超时后重新建立上下文成本可接受。工具调用结果5 到 15 分钟。天气、股价这类数据变化快但短时间内重复查询概率高。向量检索结果1 到 6 小时。文档库更新不频繁检索结果相对稳定。大模型推理结果根据业务定FAQ 类可以 24 小时创意生成类不建议缓存。# 不同数据的 TTL 配置 TTL_CONFIG { session: 1800, # 30 分钟 tool_result: 600, # 10 分钟 vector_search: 3600, # 1 小时 llm_response: 86400 # 24 小时 }实操心得TTL 最好加一点随机抖动避免大量缓存同时失效造成雪崩。比如基础 TTL 加 0 到 300 秒的随机值。3.3 缓存更新写穿、写回还是失效缓存更新有三种常见模式Cache-Aside旁路缓存读时先查缓存没有则查数据库并写入缓存写时更新数据库然后删除缓存。这是最常用的模式实现简单但存在短暂的数据不一致窗口。Write-Through写穿写时同时更新缓存和数据库。一致性较好但写延迟增加。Write-Behind写回写时只更新缓存异步批量写数据库。写性能最好但宕机可能丢数据。Agent 场景下我推荐Cache-Aside。会话上下文和工具结果对一致性要求不是极端高短暂不一致可以接受。实现时注意先更新数据库再删除缓存而不是先删缓存再更新数据库。后者在并发下更容易出现脏数据。def update_session(session_id, data): # 1. 更新数据库 db.update_session(session_id, data) # 2. 删除缓存 r.delete(fsession:{session_id}) # 下次读取时自然回填3.4 缓存穿透、击穿、雪崩的应对缓存穿透查询不存在的数据缓存和数据库都没有每次请求都打到数据库。解决方案是缓存空值或使用布隆过滤器。def get_with_null_cache(key): value r.get(key) if value is not None: return None if value __NULL__ else json.loads(value) # 查数据库 db_value db.query(key) if db_value is None: r.setex(key, 60, __NULL__) # 缓存空值短 TTL return None r.setex(key, 3600, json.dumps(db_value)) return db_value缓存击穿热点 key 失效瞬间大量请求同时打到数据库。解决方案是加互斥锁只让一个请求去回填。def get_with_mutex(key): value r.get(key) if value: return json.loads(value) # 尝试获取锁 lock_key flock:{key} if r.set(lock_key, 1, nxTrue, ex10): try: db_value db.query(key) r.setex(key, 3600, json.dumps(db_value)) return db_value finally: r.delete(lock_key) else: # 等待片刻后重试 time.sleep(0.1) return get_with_mutex(key)缓存雪崩大量 key 同时失效。解决方案是 TTL 加随机值或者用多级缓存。注意互斥锁方案在极高并发下可能造成请求堆积建议配合超时和降级策略。如果拿不到锁且等待超时直接查数据库并返回不要无限等待。4. 高并发场景下的 Redis 实战4.1 连接池配置别让连接成为瓶颈很多线上问题不是 Redis 本身慢而是连接池配置不合理。默认配置下连接数太少请求排队连接数太多Redis 服务端压力大。import redis pool redis.ConnectionPool( hostlocalhost, port6379, max_connections50, # 根据并发量调整 socket_timeout2, # 读写超时 socket_connect_timeout1,# 连接超时 retry_on_timeoutTrue, health_check_interval30 ) r redis.Redis(connection_poolpool)max_connections怎么定一个经验公式并发请求数 × 平均 Redis 操作耗时 / 1000。假设 1000 并发每次操作 2ms那需要 2 个连接就够。但实际要考虑峰值和网络抖动一般设为理论值的 3 到 5 倍。4.2 管道与批量操作减少网络往返Agent 一次请求可能需要读写多个缓存键。如果逐个操作网络往返时间累积起来很可观。用 Pipeline 批量发送能显著降低延迟。def get_session_bundle(session_id): pipe r.pipeline() pipe.hgetall(fsession:{session_id}) pipe.lrange(fchat:{session_id}, 0, 9) pipe.get(ftool:weather:{session_id}) results pipe.execute() return { session: results[0], history: results[1], weather: results[2] }实测下来10 个键的批量读取用 Pipeline 比逐个读取快 5 到 8 倍。但注意 Pipeline 不是原子操作中间可能插入其他命令。4.3 分布式锁Agent 任务互斥多个 Agent 实例同时处理同一个用户请求时需要互斥锁避免重复计算。Redis 的SET NX EX是实现分布式锁的常用方式。import uuid def acquire_lock(lock_name, timeout10): token str(uuid.uuid4()) acquired r.set(flock:{lock_name}, token, nxTrue, extimeout) return token if acquired else None def release_lock(lock_name, token): # Lua 脚本保证原子性 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, flock:{lock_name}, token)释放锁必须用 Lua 脚本保证“判断 token 是否匹配”和“删除锁”两个操作的原子性。否则可能出现A 判断 token 匹配后锁刚好过期B 拿到锁A 删除了 B 的锁。4.4 主从与集群容量和可用性扩展单机 Redis 有内存上限和单点故障风险。生产环境建议至少主从架构读写分离主节点写从节点读。# Docker 启动主从 docker run -d --name redis-master -p 6379:6379 redis:7 docker run -d --name redis-slave -p 6380:6379 redis:7 \ redis-server --slaveof redis-master 6379数据量超过单机内存时需要 Redis Cluster 分片。但 Cluster 有一些限制不支持多键操作跨槽、事务受限。Agent 场景下如果会话数据按 session_id 分片大多数操作是单键的Cluster 问题不大。实操心得主从复制有延迟对一致性要求高的读操作比如刚写入就读取建议走主节点。可以在代码里区分读写连接。5. 常见问题与排查技巧实录5.1 连接超时Redis command timed out这是最常见的报错之一。io.lettuce.core.RedisCommandTimeoutException或者redis.exceptions.TimeoutError。排查思路可能原因排查方法解决方案网络延迟ping Redis 服务器检查网络链路同机房部署慢查询SLOWLOG GET优化大 key、复杂命令连接池耗尽监控连接数增大 max_connections内存不足INFO memory扩容或清理数据主从切换查看哨兵日志配置合理的超时和重试我遇到过一次原因是某个 key 存了 10MB 的大 JSON每次读取都阻塞 Redis 主线程。后来把大 key 拆成多个小 key问题解决。5.2 内存告警used_memory 持续增长Redis 内存增长通常有几个来源缓存数据没有设置 TTL、大 key 堆积、客户端输出缓冲区膨胀。# 查看内存概况 redis-cli INFO memory # 查看大 key redis-cli --bigkeys # 查看内存碎片率 redis-cli INFO memory | grep mem_fragmentation_ratiomem_fragmentation_ratio大于 1.5 说明碎片较多可以开启activedefrag yes自动整理。但整理会消耗 CPU建议在低峰期进行。5.3 缓存与数据库不一致Cache-Aside 模式下不一致通常来自并发写。比如请求 A 更新数据库后删除缓存请求 B 在 A 删除缓存前读取了旧缓存并回填导致缓存里是旧数据。解决方案有几种延迟双删更新数据库后删缓存延迟几百毫秒再删一次、订阅数据库 binlog 异步删除缓存、或者对一致性要求高的数据直接读数据库。Agent 场景下会话上下文的不一致影响不大用户下一轮对话会覆盖。但工具结果如果缓存了旧数据可能给用户错误信息。我的做法是对时效敏感的工具结果TTL 设短一点或者不缓存。5.4 序列化问题JSON 还是 MessagePackPython 的json.dumps和json.loads简单通用但性能一般且不支持 bytes 类型。MessagePack 更紧凑、更快但可读性差。import msgpack # 序列化 packed msgpack.packb(data, use_bin_typeTrue) r.setex(key, ttl, packed) # 反序列化 unpacked msgpack.unpackb(r.get(key), rawFalse)实测下来MessagePack 的序列化速度比 JSON 快 2 到 3 倍体积小 30% 左右。如果缓存数据量大值得切换。但调试时不如 JSON 直观建议在开发环境用 JSON生产环境用 MessagePack。5.5 常见问题速查表现象可能原因快速处理响应变慢大 key、慢查询--bigkeys、SLOWLOG连接超时连接池小、网络抖动增大连接池、检查网络内存满无 TTL、数据堆积设置 TTL、清理冷数据缓存命中率低TTL 太短、键设计不合理调整 TTL、优化键结构主从延迟写入量大、网络带宽监控延迟、读写分离集群槽迁移扩容缩容低峰期操作、监控迁移进度6. 从零搭建一个带缓存的 AI Agent 服务6.1 环境准备与依赖安装先装 Redis。macOS 用 HomebrewLinux 用 apt 或 yumWindows 建议用 Docker。# macOS brew install redis brew services start redis # Ubuntu sudo apt update sudo apt install redis-server sudo systemctl start redis # Docker推荐跨平台一致 docker run -d --name redis -p 6379:6379 redis:7-alpinePython 依赖pip install redis fastapi uvicorn openai langchain6.2 缓存层封装我习惯把缓存操作封装成一个类统一管理键前缀、TTL 和序列化。import redis import json import hashlib from typing import Any, Optional class AgentCache: def __init__(self, hostlocalhost, port6379, db0): self.r redis.Redis( hosthost, portport, dbdb, decode_responsesTrue, socket_timeout2, socket_connect_timeout1 ) def _key(self, prefix: str, identifier: str) - str: return fagent:{prefix}:{identifier} def get_session(self, session_id: str) - Optional[dict]: key self._key(session, session_id) data self.r.hgetall(key) return data if data else None def set_session(self, session_id: str, data: dict, ttl: int 1800): key self._key(session, session_id) self.r.hset(key, mappingdata) self.r.expire(key, ttl) def get_tool_result(self, tool_name: str, params: dict) - Optional[Any]: param_hash hashlib.md5( json.dumps(params, sort_keysTrue).encode() ).hexdigest()[:8] key self._key(ftool:{tool_name}, param_hash) data self.r.get(key) return json.loads(data) if data else None def set_tool_result(self, tool_name: str, params: dict, result: Any, ttl: int 600): param_hash hashlib.md5( json.dumps(params, sort_keysTrue).encode() ).hexdigest()[:8] key self._key(ftool:{tool_name}, param_hash) self.r.setex(key, ttl, json.dumps(result))6.3 集成到 FastAPI Agent 服务from fastapi import FastAPI from pydantic import BaseModel from agent_cache import AgentCache app FastAPI() cache AgentCache() class ChatRequest(BaseModel): session_id: str message: str app.post(/chat) async def chat(req: ChatRequest): # 1. 读取会话上下文 session cache.get_session(req.session_id) or {} history session.get(history, []) # 2. 检查是否有缓存的大模型响应 cache_key_params {message: req.message, history_len: len(history)} cached_response cache.get_tool_result(llm, cache_key_params) if cached_response: return {response: cached_response, cached: True} # 3. 调用大模型这里用伪代码 response await call_llm(req.message, history) # 4. 更新会话上下文 history.append({role: user, content: req.message}) history.append({role: assistant, content: response}) cache.set_session(req.session_id, { history: json.dumps(history[-20:]), # 只保留最近 20 条 updated_at: str(time.time()) }) # 5. 缓存响应 cache.set_tool_result(llm, cache_key_params, response, ttl3600) return {response: response, cached: False}6.4 压测与调优用wrk或locust压测观察缓存命中率和响应时间。# 安装 wrk brew install wrk # 压测 wrk -t4 -c100 -d30s -s post.lua http://localhost:8000/chat调优时重点关注几个指标缓存命中率目标 60% 以上、平均响应时间目标 200ms 以内、Redis 内存使用率目标 70% 以下。如果命中率低检查键设计是否合理、TTL 是否太短如果响应时间高检查是否有大 key 或慢查询。实操心得压测时先用真实流量回放而不是随机生成请求。Agent 场景下用户问题的分布很不均匀热门问题可能占 80% 流量用真实数据才能测出真实命中率。7. 缓存治理与长期维护7.1 监控指标别等出事才看Redis 需要监控的核心指标used_memory、connected_clients、instantaneous_ops_per_sec、keyspace_hits、keyspace_misses、evicted_keys。# 实时监控 redis-cli --stat # 查看命中率 redis-cli INFO stats | grep keyspace命中率计算公式keyspace_hits / (keyspace_hits keyspace_misses)。低于 50% 说明缓存策略有问题需要调整。7.2 键命名规范团队协作的基础多人协作时键命名混乱是灾难。我建议统一格式{业务}:{模块}:{标识}:{版本}。agent:session:sess_12345:v1 agent:tool:weather:abc12345:v1 agent:vector:search:def67890:v1版本号的作用是当数据结构变更时直接升版本旧缓存自然淘汰不用手动清理。7.3 定期清理与容量规划即使设置了 TTL也可能有漏网之鱼。建议每周跑一次扫描找出没有 TTL 的键。# 扫描没有 TTL 的键生产环境慎用建议用 SCAN redis-cli --scan --pattern agent:* | while read key; do ttl$(redis-cli TTL $key) if [ $ttl -eq -1 ]; then echo No TTL: $key fi done容量规划按“峰值内存 × 1.5”预留。如果当前用了 4GB峰值可能到 6GB那至少配 8GB 内存。Redis 的maxmemory-policy建议设为allkeys-lru内存满时自动淘汰最久未使用的键。7.4 缓存治理的常见误区误区一缓存越多越好。缓存占用内存内存是成本。只缓存那些“计算贵、访问频、变化慢”的数据。误区二TTL 设长一点省事。TTL 太长数据陈旧用户拿到过期信息反而损害体验。宁可短一点配合回填机制。误区三忽略序列化开销。大对象的序列化和反序列化本身消耗 CPU。如果缓存的数据很大先压缩再存或者拆分存储。误区四不做降级。Redis 挂了怎么办必须有降级策略直接查数据库、返回默认值、或者限流。不能因为缓存故障导致整个 Agent 不可用。我在实际项目中踩过最深的坑是早期没有做缓存降级。一次 Redis 主从切换整个 Agent 服务雪崩用户端全部超时。后来加了降级开关Redis 不可用时自动绕过缓存直接查库虽然慢但至少可用。这个教训让我明白缓存是加速器不是必需品。系统设计时永远要假设缓存会失效。