企业官网建设流程全解析

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 String：LLM 响应和简单结果的首选

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(host='localhost', port=6379, db=0, decode_responses=False) def cache_llm_response(prompt, model, temperature, response, ttl=3600): key_raw = f"llm:{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, b"Z" + data) else: r.setex(key, ttl, b"J" + data) def get_llm_response(prompt, model, temperature): key_raw = f"llm:{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 == b"Z": 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, ttl=1800): key = f"session:{session_id}" r.hset(key, field, json.dumps(value)) r.expire(key, ttl) def get_session_field(session_id, field): key = f"session:{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 里写一条事件，方便追踪和回放。相比 List，Stream 支持消费者组、消息确认、按 ID 范围查询，做 Agent 的可观测性特别合适。

def append_agent_event(session_id, event_type, payload): key = f"events:{session_id}" r.xadd(key, {"type": event_type, "payload": json.dumps(payload)}, maxlen=1000) def read_agent_events(session_id, count=100): key = f"events:{session_id}" events = r.xrevrange(key, count=count) return [{"id": eid, **{k.decode(): v.decode() for k, v in fields.items()}} for eid, fields in events]

maxlen=1000是防止 Stream 无限增长，超过就自动裁剪最老的。这个参数要根据实际事件量调，太小会丢历史，太大占内存。

3.4 Sorted Set：带权重的记忆和优先级队列

Agent 的长期记忆如果要做相关性排序，Sorted Set 很好用。score 存时间戳或者重要性分数，member 存记忆内容或者记忆 ID。取最近记忆用ZREVRANGE，取某个分数区间的用ZRANGEBYSCORE。

任务队列也可以用 Sorted Set 实现，score 是优先级或者计划执行时间，Agent 从队列里取任务时按 score 排序。这比 List 的 FIFO 灵活得多，支持延迟任务和优先级调度。

3.5 数据类型选型速查表

这张表是我几个项目总结下来的，基本覆盖了 Agent 场景 90% 的缓存需求。选型的时候先想清楚数据的读写模式，是整体读写还是字段级读写，是顺序访问还是随机访问，是短期还是长期，答案自然就出来了。

4. AI Agent 怎么扛并发：缓存层的并发控制

4.1 缓存击穿：同一个热点 key 被并发打爆

这是 Agent 系统最典型的并发问题。假设某个热门 prompt 突然被大量用户同时请求，缓存里没有，所有请求同时去调 LLM，瞬间把配额打满，响应时间爆炸。这就是缓存击穿。

解决方案是分布式锁加双重检查。第一个拿到锁的请求去调模型，其他请求等待或者返回旧值。等第一个请求写回缓存后，其他请求直接读缓存。

import time import uuid def get_or_compute_with_lock(key, compute_fn, ttl=3600, lock_timeout=30): val = r.get(key) if val is not None: return val lock_key = f"lock:{key}" lock_value = str(uuid.uuid4()) acquired = r.set(lock_key, lock_value, nx=True, ex=lock_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()

这段代码有几个关键点。nx=True保证只有一个请求能拿到锁，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, ttl=3600, null_ttl=60): 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 的热点 key，TTL 设 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, ttl=3600): version = r.get(f"version:{namespace}") or b"1" 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(f"version:{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 常见问题速查表

这张表基本覆盖了我遇到过的所有缓存问题，排查的时候按现象对号入座，能省不少时间。

7. 从零搭建：一个可复现的 Agent 缓存层

7.1 环境准备与 Redis 安装

先装 Redis。Linux 上用包管理器最省事，macOS 用 Homebrew，Windows 建议用 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-alpine

Docker 方式我推荐用 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( host='localhost', port=6379, db=0, max_connections=50, socket_timeout=2, socket_connect_timeout=2, retry_on_timeout=True, health_check_interval=30 ) r = redis.Redis(connection_pool=pool)

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, host='localhost', port=6379, db=0, max_connections=50): pool = ConnectionPool( host=host, port=port, db=db, max_connections=max_connections, socket_timeout=2, socket_connect_timeout=2, retry_on_timeout=True, health_check_interval=30 ) self.r = redis.Redis(connection_pool=pool) 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, ttl=3600, lock_timeout=30): 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 = f"lock:{key}" lock_value = str(uuid.uuid4()) acquired = self.r.set(lock_key, lock_value, nx=True, ex=lock_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 b"Z" + zlib.compress(raw) return b"J" + raw def _decode(self, data): flag, payload = data[:1], data[1:] if flag == b"Z": payload = zlib.decompress(payload) return json.loads(payload) def update_session(self, session_id, field, value, ttl=1800): key = f"session:{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(f"session:{session_id}", field) return json.loads(val) if val else None def append_event(self, session_id, event_type, payload, maxlen=1000): self.r.xadd(f"events:{session_id}", {"type": event_type, "payload": json.dumps(payload)}, maxlen=maxlen)

这个类把 key 生成、压缩、抖动、锁、空值缓存、会话管理、事件流都封装好了。用的时候直接实例化，调get_or_compute就行。

7.4 接入 Agent 的完整示例

假设有一个简单的 Agent，需要调 LLM 和搜索工具，用缓存包一层。

cache = AgentCache() def call_llm(prompt, model="gpt-4", temperature=0.7): def compute(): # 实际调用 LLM 的逻辑 return {"text": "模型返回的内容", "tokens": 100} return cache.get_or_compute( namespace="llm", raw_key=f"{model}:{temperature}:{prompt}", compute_fn=compute, ttl=3600 ) def search_tool(query): def compute(): # 实际调用搜索的逻辑 return {"results": ["结果1", "结果2"]} return cache.get_or_compute( namespace="tool:search", raw_key=query, compute_fn=compute, ttl=300 ) 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 加 Grafana，redis_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。这些等有机会再展开聊。