1. 为什么 AI Agent 必须认真对待缓存这件事做过 AI Agent 项目的人大概都有过这种体验本地跑一个对话流程丝滑得不行一上线用户量稍微起来一点响应时间就从几百毫秒飙到十几秒账单也跟着往上翻。问题往往不在模型本身而在于 Agent 的每一次思考都在重复做大量无意义的计算和请求。这时候 Redis 缓存就不是一个可选项而是决定项目能不能活下去的基础设施。我这两年陆续搭过几个基于大模型的 Agent 系统从最简单的问答机器人到带工具调用、多轮规划、长期记忆的复杂智能体踩过的坑基本都跟缓存有关。这篇文章就把 AI Agent 和 Redis 缓存结合这件事讲透从架构设计、数据类型选型、并发扛压、缓存失效策略到线上排查的实战经验全部摊开来说。不管你是刚接触 Agent 开发的新手还是已经在做线上系统的工程师应该都能从里面找到能直接抄作业的东西。先说清楚这篇文章适合谁看。如果你正在用 Python、Java 或者 Rust 搭建 AI Agent遇到了响应慢、成本高、并发上不去的问题那这篇就是写给你的。如果你还在纠结 Agent 到底该怎么设计缓存层或者被 Redis 的一堆数据类型搞得眼花缭乱那也能在这里找到答案。我会尽量用大白话把原理讲明白同时给出可以直接复现的代码和配置。核心关键词就三个AI Agent、Redis、缓存。这三个词串起来本质上要解决的是一个工程问题——如何让一个计算密集、IO 密集、状态复杂的智能体系统在有限的资源下稳定高效地服务大量请求。下面我按自己的实战思路一层一层拆开讲。2. AI Agent 的缓存需求到底特殊在哪2.1 普通 Web 缓存和 Agent 缓存的本质区别很多人第一反应是缓存嘛不就是把结果存起来下次直接取。传统 Web 应用的缓存确实相对简单一个 HTTP 请求对应一个响应key 就是 URL 加参数value 就是响应体过期时间设个几分钟命中率能到八九成就算成功。但 AI Agent 完全不是这个逻辑。Agent 的一次完整执行可能包含多轮 LLM 调用、多次工具调用、向量检索、状态更新。每一轮调用的输入都依赖上一轮的输出形成一个有向图甚至是有环的决策流程。这意味着缓存的对象不是单一的请求-响应而是中间步骤的产物。比如一次工具调用的结果、一次向量检索的 top-k 文档、一段对话的摘要、一个子任务的规划结果这些都可以独立缓存而且缓存它们的收益往往比缓存最终答案还大。我做过一个对比测试一个带 5 步规划的 Agent如果只缓存最终答案命中率大概 30%如果把中间的工具调用结果和检索结果也缓存起来整体命中率能到 70% 以上平均响应时间从 8 秒降到 2 秒出头。这个差距就是 Agent 缓存设计的价值所在。2.2 Agent 场景下缓存的四类核心对象在实际项目里我把 Agent 需要缓存的东西分成四类每一类的 key 设计、过期策略、数据类型都不一样。第一类是LLM 响应缓存。同样的 prompt 和参数如果短时间内重复请求完全没必要再调一次模型。这类缓存 key 通常是 prompt 的哈希加上模型名和温度参数value 就是模型返回的文本。它的特点是写入成本高一次调用可能几毛钱读取频繁过期时间可以设得长一些比如几小时到几天。第二类是工具调用结果缓存。Agent 调用搜索、数据库查询、API 接口这些外部工具时结果往往在一段时间内是稳定的。比如查天气、查汇率、查商品库存几分钟内重复查没意义。这类缓存 key 是工具名加参数哈希过期时间短通常几十秒到几分钟。第三类是向量检索结果缓存。RAG 场景下同一个 query 的 embedding 和检索结果可以复用。key 是 query 文本的哈希value 是检索到的文档 ID 列表和相似度分数。这类缓存对降低向量数据库压力特别有效。第四类是会话状态和记忆缓存。多轮对话中用户的上下文、Agent 的中间状态、长期记忆的摘要都需要快速读写。这类数据用 Redis 的 Hash 或者 Stream 结构存最合适过期时间根据会话活跃度动态调整。把这四类分开设计而不是一股脑塞进一个缓存池是 Agent 缓存架构的第一个关键决策。混在一起会导致 key 冲突、过期策略互相干扰、排查困难。2.3 为什么是 Redis 而不是本地缓存有人会问用进程内的本地缓存比如 Python 的 lru_cache 或者 Java 的 Caffeine不行吗快是快但 Agent 系统通常是无状态多实例部署的本地缓存命中率会被实例数稀释。假设你有 10 个实例本地缓存命中率理论上限就只有单实例的十分之一左右而且实例重启缓存就没了。Redis 作为独立的缓存层所有实例共享同一份缓存命中率不受实例数影响。同时它支持丰富的数据结构、原子操作、过期策略、持久化这些对 Agent 的复杂状态管理非常关键。特别是分布式锁和原子计数器在 Agent 的并发控制里几乎是刚需。当然Redis 也不是没有代价网络往返会带来额外延迟。我的经验是对于毫秒级的本地计算本地缓存更合适对于 LLM 调用、工具调用这种动辄几百毫秒到几秒的操作Redis 的网络开销完全可以忽略。所以实际项目里通常是两级缓存本地缓存挡最热的数据Redis 挡大部分请求模型和数据库兜底。3. Redis 数据类型在 Agent 缓存中的选型实战3.1 StringLLM 响应和简单结果的首选String 是 Redis 最基础也最常用的类型。在 Agent 里LLM 的响应文本、工具调用的 JSON 结果、embedding 向量序列化后都适合用 String 存。这里有个细节值得说LLM 响应通常比较大几 KB 到几十 KB 都正常。直接存 String 没问题但要注意 Redis 单 value 最大 512MB虽然远够用但大 value 会影响网络传输和内存碎片。我的做法是对超过 100KB 的响应做压缩用 zlib 或者 snappy压缩率通常能到 3 到 5 倍读取时解压整体反而更快。import redis import json import zlib import hashlib r redis.Redis(hostlocalhost, port6379, db0, decode_responsesFalse) def cache_llm_response(prompt, model, temperature, response, ttl3600): key_raw fllm:{model}:{temperature}:{prompt} key llm: hashlib.sha256(key_raw.encode()).hexdigest() data json.dumps({response: response}).encode() if len(data) 100 * 1024: data zlib.compress(data) r.setex(key, ttl, bZ data) else: r.setex(key, ttl, bJ data) def get_llm_response(prompt, model, temperature): key_raw fllm:{model}:{temperature}:{prompt} key llm: hashlib.sha256(key_raw.encode()).hexdigest() data r.get(key) if data is None: return None flag, payload data[:1], data[1:] if flag bZ: payload zlib.decompress(payload) return json.loads(payload)[response]注意 key 的设计我用llm:前缀加模型名加温度加 prompt 哈希。前缀是为了方便按业务清理模型和温度是必须的因为同样的 prompt 在不同模型或不同温度下结果完全不同混用会导致返回错误结果。哈希是为了控制 key 长度prompt 可能很长直接当 key 会浪费内存。3.2 Hash会话状态和 Agent 中间状态多轮对话的上下文、Agent 执行过程中的中间变量用 Hash 存最合适。Hash 的字段可以单独读写不用整个 value 取出来再写回去这在并发更新时特别重要。比如一个会话的状态我用一个 Hash 存session:{session_id}字段包括history对话历史、current_task当前任务、tool_results工具结果缓存、last_active最后活跃时间。更新某个字段时用HSET读取时用HGET互不干扰。def update_session_state(session_id, field, value, ttl1800): key fsession:{session_id} r.hset(key, field, json.dumps(value)) r.expire(key, ttl) def get_session_field(session_id, field): key fsession:{session_id} val r.hget(key, field) return json.loads(val) if val else None这里有个坑我踩过Hash 的过期时间是整个 key 的不能给单个字段设过期。所以如果某些字段需要更短的过期时间得单独拆成独立的 key。比如工具结果缓存我就单独用tool:{session_id}:{tool_name}存跟会话状态分开。3.3 List 和 Stream对话历史和事件流对话历史用 List 存LPUSH加新消息LRANGE取最近 N 条LTRIM控制长度。这个模式很成熟但要注意 List 只能从头尾操作中间查询效率低。如果需要对历史做复杂查询还是得落到数据库。Stream 是 Redis 5.0 引入的适合做事件流。Agent 的每一步执行都可以往 Stream 里写一条事件方便追踪和回放。相比 ListStream 支持消费者组、消息确认、按 ID 范围查询做 Agent 的可观测性特别合适。def append_agent_event(session_id, event_type, payload): key fevents:{session_id} r.xadd(key, {type: event_type, payload: json.dumps(payload)}, maxlen1000) def read_agent_events(session_id, count100): key fevents:{session_id} events r.xrevrange(key, countcount) return [{id: eid, **{k.decode(): v.decode() for k, v in fields.items()}} for eid, fields in events]maxlen1000是防止 Stream 无限增长超过就自动裁剪最老的。这个参数要根据实际事件量调太小会丢历史太大占内存。3.4 Sorted Set带权重的记忆和优先级队列Agent 的长期记忆如果要做相关性排序Sorted Set 很好用。score 存时间戳或者重要性分数member 存记忆内容或者记忆 ID。取最近记忆用ZREVRANGE取某个分数区间的用ZRANGEBYSCORE。任务队列也可以用 Sorted Set 实现score 是优先级或者计划执行时间Agent 从队列里取任务时按 score 排序。这比 List 的 FIFO 灵活得多支持延迟任务和优先级调度。3.5 数据类型选型速查表缓存对象推荐类型key 示例过期策略LLM 响应Stringllm:{hash}1-24 小时工具调用结果Stringtool:{name}:{hash}30 秒-5 分钟向量检索结果String/Listrag:{query_hash}10-60 分钟会话状态Hashsession:{id}30 分钟滑动对话历史Listhistory:{id}1-7 天事件流Streamevents:{id}按 maxlen 裁剪长期记忆Sorted Setmemory:{user_id}按需清理分布式锁Stringlock:{resource}秒级这张表是我几个项目总结下来的基本覆盖了 Agent 场景 90% 的缓存需求。选型的时候先想清楚数据的读写模式是整体读写还是字段级读写是顺序访问还是随机访问是短期还是长期答案自然就出来了。4. AI Agent 怎么扛并发缓存层的并发控制4.1 缓存击穿同一个热点 key 被并发打爆这是 Agent 系统最典型的并发问题。假设某个热门 prompt 突然被大量用户同时请求缓存里没有所有请求同时去调 LLM瞬间把配额打满响应时间爆炸。这就是缓存击穿。解决方案是分布式锁加双重检查。第一个拿到锁的请求去调模型其他请求等待或者返回旧值。等第一个请求写回缓存后其他请求直接读缓存。import time import uuid def get_or_compute_with_lock(key, compute_fn, ttl3600, lock_timeout30): val r.get(key) if val is not None: return val lock_key flock:{key} lock_value str(uuid.uuid4()) acquired r.set(lock_key, lock_value, nxTrue, exlock_timeout) if acquired: try: val r.get(key) if val is not None: return val result compute_fn() r.setex(key, ttl, result) return result finally: if r.get(lock_key) lock_value.encode(): r.delete(lock_key) else: for _ in range(50): time.sleep(0.1) val r.get(key) if val is not None: return val return compute_fn()这段代码有几个关键点。nxTrue保证只有一个请求能拿到锁ex设置锁超时防止死锁。释放锁时先检查 value 是不是自己的避免误删别人的锁。等待的请求轮询 5 秒超时后降级为直接计算保证可用性。注意锁的 value 一定要用唯一标识不能用固定值。否则 A 请求的锁超时释放后B 请求拿到锁A 执行完去删锁会把 B 的锁删掉导致并发失控。4.2 缓存雪崩大量 key 同时过期如果一批 key 设置了相同的过期时间到点同时失效所有请求一起打到后端就是雪崩。Agent 场景下比如批量预热的 prompt 缓存很容易踩这个坑。解决办法是给过期时间加随机抖动。基础 TTL 加上一个随机偏移比如 3600 秒的缓存实际过期时间在 3000 到 4200 秒之间随机分布避免同时失效。import random def set_with_jitter(key, value, base_ttl): jitter random.randint(-int(base_ttl * 0.2), int(base_ttl * 0.2)) ttl max(60, base_ttl jitter) r.setex(key, ttl, value)抖动范围我一般设基础 TTL 的 20%太小起不到分散作用太大又不好预估缓存寿命。4.3 缓存穿透查询不存在的数据恶意请求或者逻辑 bug 导致查询大量不存在的 key每次都穿透到后端。Agent 场景下比如用户输入了乱七八糟的 prompt缓存里没有每次都去调模型。解决办法是缓存空值。查询结果为空时也往缓存里写一个特殊标记TTL 设短一点比如 60 秒。这样后续同样的请求直接命中空值缓存不会打到后端。NULL_MARKER b__NULL__ def get_with_null_cache(key, compute_fn, ttl3600, null_ttl60): val r.get(key) if val NULL_MARKER: return None if val is not None: return val result compute_fn() if result is None: r.setex(key, null_ttl, NULL_MARKER) else: r.setex(key, ttl, result) return result空值缓存的 TTL 不能太长否则数据真的出现了也读不到。60 秒是个比较平衡的值。4.4 热点 key 的本地缓存兜底即使有 Redis单个热点 key 的 QPS 太高也会把 Redis 打满。这时候可以在应用层加一层本地缓存用 Caffeine 或者简单的字典加过期时间挡住最热的那部分请求。我的做法是本地缓存只存 top 100 的热点 keyTTL 设 5 到 10 秒。这样即使 Redis 挂了最热的请求还能撑一会儿给恢复争取时间。本地缓存和 Redis 的一致性靠短 TTL 保证10 秒内的数据不一致在 Agent 场景下通常可以接受。5. 缓存失效与一致性Agent 场景下的取舍5.1 主动失效还是被动过期缓存失效有两种策略主动删除和被动过期。主动删除是数据更新时立即删缓存下次读时重建。被动过期是设 TTL到点自动失效。Agent 场景下我倾向于被动过期为主主动失效为辅。因为 Agent 的数据大多是生成后一段时间内有效比如 LLM 响应、检索结果它们不会因为外部数据变化而失效只是随时间变得不那么相关。这类数据用 TTL 就够了。但有些数据必须主动失效比如用户的配置、工具的可用状态、模型的切换。这些变化后如果还读旧缓存会导致行为错误。这类数据在更新时同步删除缓存 key。5.2 先删缓存还是先更新数据库这是缓存一致性的经典问题。在 Agent 场景下如果缓存背后是数据库标准做法是先更新数据库再删除缓存。这样即使删除失败下次读时也会因为缓存不存在而重建最终一致。但要注意并发场景A 更新数据库后删缓存B 在 A 删缓存前读了旧缓存并写回导致脏数据。这个概率很低通常可以接受。如果业务不能容忍可以用延迟双删更新数据库后删一次缓存延迟几百毫秒再删一次。5.3 版本号机制避免脏读更稳妥的做法是给缓存加版本号。每次数据更新版本号加一。缓存 key 里带上版本号读的时候先读版本号再读对应版本的缓存。版本号变了旧缓存自然失效。def get_versioned_cache(namespace, key, compute_fn, ttl3600): version r.get(fversion:{namespace}) or b1 version version.decode() cache_key f{namespace}:v{version}:{key} val r.get(cache_key) if val is not None: return val result compute_fn() r.setex(cache_key, ttl, result) return result def bump_version(namespace): r.incr(fversion:{namespace})这个机制的好处是失效是原子的不用遍历删除旧 key旧 key 靠 TTL 自然清理。缺点是会短暂占用双倍内存但通常可以接受。5.4 缓存失效的监控指标缓存失效策略好不好得靠数据说话。我一般监控这几个指标命中率、平均响应时间、后端调用量、缓存内存使用率、key 数量增长趋势。命中率低于 60% 说明缓存设计有问题要么 key 设计不合理要么 TTL 太短。后端调用量突然上升可能是缓存大面积失效。内存使用率持续上涨可能是 key 没有正确过期有内存泄漏。这些指标用 Redis 的INFO命令就能拿到配合 Prometheus 或者自建监控都行。关键是要有告警命中率跌破阈值、内存超过 80% 这些情况要第一时间知道。6. 线上排查实录那些让我熬夜的缓存问题6.1 连接超时Redis command timed out这个报错我见过太多次了。io.lettuce.core.RedisCommandTimeoutException表面看是 Redis 慢实际原因可能有好几种。第一种是大 key 阻塞。某个 key 的 value 特别大比如几 MB 的 LLM 响应读取时阻塞了其他请求。排查方法是redis-cli --bigkeys扫描大 key找到后拆分或者压缩。第二种是慢命令。KEYS *、SMEMBERS大集合、ZRANGE大范围查询这些命令在生产环境要禁用或者限制。用SLOWLOG GET查看慢命令记录。第三种是网络问题。客户端和 Redis 之间的网络抖动或者连接池配置不合理。连接池太小请求排队太大Redis 连接数爆掉。一般连接池大小设为 CPU 核数的 2 到 4 倍比较合适。第四种是持久化阻塞。RDB 快照或者 AOF 重写时Redis 会 fork 子进程如果数据量大fork 会阻塞主线程。这种情况要调整持久化策略或者用主从架构把持久化放到从节点。6.2 内存告警缓存把内存吃满了Redis 内存满了会触发淘汰策略如果策略配置不当可能把重要数据淘汰掉。Agent 场景下LLM 响应缓存通常很大最容易吃内存。首先要设置maxmemory和maxmemory-policy。我一般用allkeys-lru所有 key 按最近最少使用淘汰。如果有些 key 绝对不能淘汰用volatile-lru只淘汰设置了过期时间的 key。其次要定期清理无用 key。比如过期的会话状态、不再需要的中间结果。可以用SCAN配合TTL批量清理注意不要用KEYS会阻塞。redis-cli --scan --pattern session:* | while read key; do ttl$(redis-cli ttl $key) if [ $ttl -eq -1 ]; then redis-cli del $key fi done这段脚本清理没有设置过期时间的 session key防止它们永久占用内存。生产环境要小心执行最好在从节点先试。6.3 缓存与数据库不一致用户反馈我明明改了配置Agent 还是用旧的。这种问题通常是缓存没失效。排查思路是先确认数据库是否更新成功再确认缓存是否删除最后确认读的时候是不是命中了旧缓存。我遇到过一次是因为更新数据库和删缓存不在同一个事务里删缓存失败了但没重试。后来改成用消息队列异步删缓存失败自动重试问题就解决了。还有一种情况是本地缓存没失效。应用层的本地缓存 TTL 设太长Redis 删了但本地还有。解决办法是本地缓存 TTL 设短或者用发布订阅通知所有实例清本地缓存。6.4 常见问题速查表现象可能原因排查方法解决方向响应突然变慢大 key 阻塞--bigkeys扫描拆分或压缩大 key连接超时连接池不足查看连接数调整连接池大小内存告警key 未过期INFO memory设置淘汰策略清理无用 key命中率低key 设计不合理监控命中率优化 key调整 TTL数据不一致缓存未失效对比缓存和数据库主动失效加版本号并发打爆后端缓存击穿查看后端 QPS分布式锁加双重检查批量失效雪崩查看 key 过期分布TTL 加随机抖动这张表基本覆盖了我遇到过的所有缓存问题排查的时候按现象对号入座能省不少时间。7. 从零搭建一个可复现的 Agent 缓存层7.1 环境准备与 Redis 安装先装 Redis。Linux 上用包管理器最省事macOS 用 HomebrewWindows 建议用 Docker。# Ubuntu/Debian sudo apt update sudo apt install redis-server # macOS brew install redis # Docker推荐跨平台一致 docker run -d --name redis -p 6379:6379 redis:7-alpineDocker 方式我推荐用 alpine 镜像体积小启动快。生产环境要挂载数据卷配置持久化。docker run -d --name redis \ -p 6379:6379 \ -v /data/redis:/data \ redis:7-alpine \ redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru--appendonly yes开启 AOF 持久化--maxmemory限制内存--maxmemory-policy设置淘汰策略。这三个参数是生产环境必配的。7.2 Python 客户端连接与连接池Python 用 redis-py一定要用连接池不要每次操作都新建连接。import redis from redis import ConnectionPool pool ConnectionPool( hostlocalhost, port6379, db0, max_connections50, socket_timeout2, socket_connect_timeout2, retry_on_timeoutTrue, health_check_interval30 ) r redis.Redis(connection_poolpool)max_connections根据并发量调一般 50 到 200。socket_timeout设 2 秒避免请求卡死。health_check_interval定期检查连接健康防止用到失效连接。7.3 完整的 Agent 缓存封装把前面讲的都整合起来封装成一个 AgentCache 类。import redis import json import hashlib import zlib import random import time import uuid from redis import ConnectionPool class AgentCache: def __init__(self, hostlocalhost, port6379, db0, max_connections50): pool ConnectionPool( hosthost, portport, dbdb, max_connectionsmax_connections, socket_timeout2, socket_connect_timeout2, retry_on_timeoutTrue, health_check_interval30 ) self.r redis.Redis(connection_poolpool) self.null_marker b__NULL__ def _make_key(self, namespace, raw): h hashlib.sha256(raw.encode()).hexdigest() return f{namespace}:{h} def _set_with_jitter(self, key, value, base_ttl): jitter random.randint(-int(base_ttl * 0.2), int(base_ttl * 0.2)) ttl max(60, base_ttl jitter) self.r.setex(key, ttl, value) def get_or_compute(self, namespace, raw_key, compute_fn, ttl3600, lock_timeout30): key self._make_key(namespace, raw_key) val self.r.get(key) if val self.null_marker: return None if val is not None: return self._decode(val) lock_key flock:{key} lock_value str(uuid.uuid4()) acquired self.r.set(lock_key, lock_value, nxTrue, exlock_timeout) if acquired: try: val self.r.get(key) if val is not None: return self._decode(val) if val ! self.null_marker else None result compute_fn() if result is None: self.r.setex(key, 60, self.null_marker) else: self._set_with_jitter(key, self._encode(result), ttl) return result finally: if self.r.get(lock_key) lock_value.encode(): self.r.delete(lock_key) else: for _ in range(50): time.sleep(0.1) val self.r.get(key) if val is not None: return self._decode(val) if val ! self.null_marker else None return compute_fn() def _encode(self, data): raw json.dumps(data).encode() if len(raw) 100 * 1024: return bZ zlib.compress(raw) return bJ raw def _decode(self, data): flag, payload data[:1], data[1:] if flag bZ: payload zlib.decompress(payload) return json.loads(payload) def update_session(self, session_id, field, value, ttl1800): key fsession:{session_id} self.r.hset(key, field, json.dumps(value)) self.r.expire(key, ttl) def get_session(self, session_id, field): val self.r.hget(fsession:{session_id}, field) return json.loads(val) if val else None def append_event(self, session_id, event_type, payload, maxlen1000): self.r.xadd(fevents:{session_id}, {type: event_type, payload: json.dumps(payload)}, maxlenmaxlen)这个类把 key 生成、压缩、抖动、锁、空值缓存、会话管理、事件流都封装好了。用的时候直接实例化调get_or_compute就行。7.4 接入 Agent 的完整示例假设有一个简单的 Agent需要调 LLM 和搜索工具用缓存包一层。cache AgentCache() def call_llm(prompt, modelgpt-4, temperature0.7): def compute(): # 实际调用 LLM 的逻辑 return {text: 模型返回的内容, tokens: 100} return cache.get_or_compute( namespacellm, raw_keyf{model}:{temperature}:{prompt}, compute_fncompute, ttl3600 ) def search_tool(query): def compute(): # 实际调用搜索的逻辑 return {results: [结果1, 结果2]} return cache.get_or_compute( namespacetool:search, raw_keyquery, compute_fncompute, ttl300 ) def agent_run(session_id, user_input): cache.append_event(session_id, user_input, {text: user_input}) history cache.get_session(session_id, history) or [] history.append({role: user, content: user_input}) llm_result call_llm(str(history)) search_result search_tool(user_input) history.append({role: assistant, content: llm_result[text]}) cache.update_session(session_id, history, history) cache.append_event(session_id, agent_output, llm_result) return llm_result[text]这个示例虽然简化了但结构是完整的。LLM 调用和搜索都走了缓存会话状态和事件流也管理起来了。实际项目里把 compute 函数换成真实的调用逻辑就行。8. 几个容易被忽略的实操心得8.1 key 命名规范比你想的重要我见过太多项目 key 命名乱七八糟user_123_cache、cache_user_123、USER:123混着用排查问题时根本不知道哪个 key 是什么。统一用业务:对象:标识的格式比如llm:gpt4:abc123、session:user_456、tool:search:xyz。冒号分隔全小写前缀固定。这样用SCAN按前缀清理的时候特别方便。8.2 序列化方式影响性能和兼容性JSON 可读性好但体积大pickle 快但跨语言不兼容msgpack 折中。Agent 场景下我一般用 JSON因为调试方便而且压缩后体积差距不大。如果对性能极致要求可以用 msgpack 或者 protobuf。但要注意序列化方式一旦定了改起来很麻烦因为旧缓存反序列化会失败。所以一开始就要想清楚。8.3 缓存预热能显著提升冷启动体验系统刚上线或者重启后缓存是空的第一批请求会很慢。可以在启动时预热热点数据比如常用的 prompt、高频的检索 query。预热脚本从数据库或者日志里捞出 top N 的热点提前调一遍写入缓存。这样上线后第一批用户也能享受缓存加速。8.4 监控和告警是缓存的生命线没有监控的缓存就是定时炸弹。至少要监控命中率、内存使用率、连接数、慢命令数、key 数量。命中率跌破 60%、内存超过 80%、慢命令突然增多都要告警。我一般用 Prometheus 加 Grafanaredis_exporter 采集指标配置几条核心告警规则。8.5 定期做缓存容量规划缓存不是越大越好内存是有限的。要根据业务量估算缓存容量。比如每天 10 万次 LLM 调用平均响应 10KB缓存 24 小时那就是 10 万乘 10KB 等于 1GB。加上工具调用、会话状态总共可能 2 到 3GB。留一倍余量Redis 内存设 6GB 比较稳妥。这个估算要定期更新业务涨了缓存也要跟着扩。9. 关于 AI Agent 缓存治理的一些个人体会做 Agent 缓存这两年我最大的感受是缓存不是加一层就完事而是一套需要持续治理的机制。上线只是开始后面要不断看数据、调参数、清垃圾、优化 key 设计。一个健康的缓存系统命中率应该稳定在 70% 以上内存增长平稳没有大 key慢命令几乎为零。另一个体会是不要过度缓存。有些数据变化频繁缓存了反而增加不一致风险还不如直接查。判断标准是这个数据的读取频率是否远高于更新频率读取成本是否远高于缓存成本。两个都是是才值得缓存。最后分享一个小技巧给缓存 key 加一个来源标记比如llm:、tool:、rag:然后在监控里按来源统计命中率和内存占用。这样能快速定位是哪类缓存出了问题比笼统看整体指标有用得多。我靠这个技巧定位过好几次问题比如发现是某个工具的缓存 key 设计有问题导致命中率低改完命中率直接涨了 20 个百分点。这套东西后续还可以往几个方向扩展一是接入多级缓存本地加 Redis 加 CDN二是做缓存的分层过期热数据长 TTL冷数据短 TTL三是用 Redis 的模块能力比如 RedisJSON、RediSearch把向量检索也放进 Redis。这些等有机会再展开聊。
