企业官网建设流程全解析

1. 为什么 AI Agent 需要 Redis 缓存

1.1 从一次线上事故说起

去年冬天，我负责的一个 AI Agent 项目在凌晨两点突然告警。这个 Agent 主要做两件事：帮用户查资料、生成摘要，然后通过企业微信推送给团队。平时 QPS 也就几十，那天因为一个内部活动，瞬间涌进来两千多个请求。结果不到三分钟，后端的大模型接口开始大面积超时，数据库连接池直接被打满，整个服务雪崩。

事后复盘，问题出在一个很朴素的地方：同一个用户在同一分钟内重复提交了完全相同的查询。Agent 每次都老老实实调用大模型、查向量库、拼上下文，一遍又一遍地烧钱、烧时间。如果当时有一层缓存挡在前面，哪怕只缓存 60 秒，也能把 90% 的重复请求拦下来。

这就是我今天想聊的主题：AI Agent 和 Redis 缓存的结合。不是那种“Hello World”式的演示，而是真正在生产环境里扛过并发、踩过坑、调过参的实战经验。如果你正在搭建 AI Agent，或者已经在跑但被并发和成本折磨得够呛，这篇内容应该能帮你少走不少弯路。

1.2 AI Agent 的缓存需求到底特殊在哪

传统 Web 应用的缓存，思路很直接：查数据库慢，就在前面加一层 Redis，把热点数据放进去，下次直接读内存。但 AI Agent 的缓存场景要复杂得多，我把它归纳为三个“不一样”。

第一，缓存的对象不一样。普通应用缓存的是结构化数据，比如用户信息、商品详情，key 和 value 都很明确。AI Agent 缓存的是“一次完整的推理结果”——可能是一段自然语言回答、一个工具调用链的执行结果、甚至是一组多轮对话的中间状态。这些东西往往是非结构化的、长度不固定的，序列化和反序列化的成本不能忽略。

第二，失效策略不一样。商品价格变了要立刻失效，但 AI Agent 的回答呢？同一个问题，今天问和明天问，答案可能因为模型更新、知识库更新而不同。你不能简单设个 TTL 就完事，得考虑“语义缓存”和“精确缓存”的区分。

第三，并发模型不一样。AI Agent 的请求往往是长耗时操作，一次推理可能几秒到几十秒。这期间如果同一个请求重复进来，你是让它排队、还是直接返回缓存、还是做请求合并？每种选择背后的取舍都不一样。

提示：如果你的 Agent 目前 QPS 低于 10，且对成本不敏感，可以先不上缓存。但只要出现“同一问题被反复问”“大模型调用费用超预期”“高峰期响应变慢”这三个信号中的任意一个，就该认真考虑 Redis 了。

1.3 本文能帮你解决什么

接下来我会从架构设计、Redis 数据结构选型、缓存 key 的设计、并发控制、失效策略、监控排查这几个维度，把 AI Agent 缓存这件事拆开讲透。每一部分都会给出可直接参考的代码片段和配置参数，也会分享我在实际项目里踩过的坑。

适合的读者：正在用 Python、Java 或 Rust 搭建 AI Agent 的后端工程师；负责 AI 应用稳定性与成本优化的技术负责人；对 Redis 有一定了解但没在 AI 场景下用过的开发者。不需要你是 Redis 专家，但至少要能看懂基本的命令和数据结构。

2. 整体架构设计与方案选型

2.1 缓存层应该放在 Agent 的哪个位置

这是第一个要拍板的问题。AI Agent 的典型链路是：接收请求 → 意图识别 → 工具调用/知识检索 → 大模型推理 → 结果后处理 → 返回。缓存可以放在多个位置，但效果差别很大。

我试过三种方案，直接说结论：

请求入口层缓存是我最推荐的起步方案。它的逻辑很简单：用户发来一个问题，先对问题做归一化处理（去空格、转小写、去掉无意义的标点），然后拿这个归一化后的文本去 Redis 里查。命中就直接返回，没命中才走完整链路，走完再把结果写回 Redis。

这个方案的好处是拦截率最高。根据我在几个项目里的统计，问答类 Agent 的重复问题比例通常在 40% 到 70% 之间，尤其是企业内部知识助手这种场景，大家问来问去就那么些问题。入口层缓存能把这部分全部挡掉，省下的不只是大模型费用，还有整条链路的耗时。

工具调用层缓存适合那些依赖外部服务的 Agent。比如你的 Agent 要调用天气 API、股票 API、或者某个 SaaS 平台的接口，这些接口往往有速率限制或者按次收费。在工具调用前加一层缓存，key 用“工具名 + 参数哈希”，能有效减少外部调用次数。我一般会给这类缓存设一个较短的 TTL，比如 5 到 15 分钟，因为外部数据的新鲜度要求比较高。

大模型推理层缓存是最复杂的，也是最近比较火的“语义缓存”。它的思路是：即使用户问法不同，只要语义相近，就返回缓存结果。比如“今天天气怎么样”和“今天天气如何”，字面不同但意思一样。实现上通常要用向量化模型把问题转成 embedding，然后在向量库里做相似度检索。这套方案命中率提升有限，但引入的复杂度和误判风险不小，我建议除非有明确的业务需求，否则不要一上来就搞。

2.2 为什么选 Redis 而不是本地缓存

有人会问：我用 Python 的functools.lru_cache或者 Java 的 Caffeine 不就行了，为什么要引入 Redis？

本地缓存的问题在于多实例不一致。你的 Agent 服务通常不会只跑一个进程，可能是 4 个、8 个甚至更多。每个实例维护自己的本地缓存，命中率会被稀释，而且数据更新时没法同步失效。更麻烦的是，Agent 服务经常需要滚动发布，一重启本地缓存就全没了，缓存雪崩的风险很高。

Redis 作为集中式缓存，解决了这几个问题：所有实例共享同一份缓存，命中率最大化；支持丰富的过期策略和淘汰策略；可以持久化，重启不丢数据；还能顺便做分布式锁、限流、消息队列这些事。

当然，Redis 也不是没有代价。网络往返会增加几毫秒的延迟，对于超低延迟场景可能不划算。但在 AI Agent 这个场景下，一次大模型推理动辄几秒，几毫秒的网络开销完全可以忽略。

2.3 Redis 部署形态怎么选

单机 Redis 适合开发和测试，生产环境我一般推荐主从 + 哨兵或者集群模式。

主从 + 哨兵的好处是配置简单，故障转移自动化程度高，适合数据量不大（比如几 GB 以内）的场景。集群模式适合数据量大、写入吞吐高的场景，但运维复杂度会上升一个台阶，而且有些命令在集群模式下有限制，比如跨 slot 的多 key 操作。

对于大多数 AI Agent 项目，我的建议是：先用主从 + 哨兵，等缓存数据量超过 10GB 或者 QPS 超过 5 万再考虑集群。别为了“看起来专业”而过早引入集群，运维成本会让你后悔。

如果你用 Docker 部署，一个典型的主从配置大概是这样：

version: '3.8' services: redis-master: image: redis:7.2-alpine ports: - "6379:6379" command: redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru volumes: - ./master-data:/data redis-slave: image: redis:7.2-alpine ports: - "6380:6379" command: redis-server --appendonly yes --slaveof redis-master 6379 volumes: - ./slave-data:/data

这里有两个参数值得说明。maxmemory 2gb限制了 Redis 最大内存，防止它把服务器内存吃光。maxmemory-policy allkeys-lru是淘汰策略，意思是内存满了之后，优先淘汰最近最少使用的 key。对于缓存场景，LRU 是最常用的策略，但如果你希望缓存的数据尽量保留久一点，可以用allkeys-lfu，它基于访问频率淘汰，对热点数据更友好。

注意：千万不要用noeviction策略。这个策略在内存满时会拒绝写入，导致你的 Agent 写缓存失败，进而影响主流程。我见过有人因为这个配置，在高峰期缓存写入全部报错，日志刷屏。

3. 核心细节解析与实操要点

3.1 缓存 key 怎么设计才合理

Key 的设计看起来简单，但实际项目里最容易出问题。我见过用整个 JSON 请求体做 key 的，长度几百个字符，Redis 内存被 key 本身吃掉一大半；也见过用自增 ID 做 key 的，结果不同用户之间缓存串了。

我的经验是遵循三个原则：可读、可控、可清理。

可读是指 key 要能一眼看出是什么内容。我通常用冒号分隔的命名空间，比如agent:qa:{hash}表示问答缓存，agent:tool:{tool_name}:{param_hash}表示工具调用缓存。这样在 Redis 客户端里一看就知道是什么。

可控是指 key 的长度要控制。Redis 的 key 本身也占内存，太长的 key 会浪费空间。我的做法是对原始内容做哈希，用 SHA256 或者 MD5，取前 16 个字符就够了。碰撞概率极低，长度也可控。

可清理是指要能按业务维度批量删除。比如某个知识库更新了，需要清掉所有相关的缓存。如果 key 里带了知识库版本号，就可以用SCAN命令匹配删除。但注意，生产环境不要用KEYS命令，它会阻塞 Redis，用SCAN分批处理。

一个实际的 key 生成函数大概长这样：

import hashlib import json def build_cache_key(prefix: str, content: str, version: str = "v1") -> str: normalized = content.strip().lower() digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16] return f"agent:{prefix}:{version}:{digest}"

这里的version字段很关键。当你调整了 Agent 的 prompt 或者换了模型，旧缓存就不应该再被命中。改一下 version，所有旧 key 自然失效，不用手动清理。

3.2 缓存内容用什么数据结构存

Redis 支持 String、Hash、List、Set、ZSet 等多种数据结构。AI Agent 的缓存内容通常是“问题 + 答案 + 元数据”这样的组合，我推荐两种存法。

方案一：String 存 JSON。把整个结果序列化成 JSON 字符串，直接SET进去。优点是简单直接，读取时一次GET就拿到全部。缺点是如果只想更新其中某个字段，得整个读出来改完再写回去。

方案二：Hash 存字段。用HSET把答案、模型版本、生成时间、token 消耗等字段分开存。优点是灵活，可以单独更新某个字段，也方便用HGET只取需要的部分。缺点是字段多时内存开销略大。

我一般用方案一，因为 Agent 的缓存结果通常是整体使用的，很少需要单独更新某个字段。而且 JSON 序列化在 Python 和 Java 里都很成熟，性能不是瓶颈。

但有一个细节要注意：序列化方式的选择。Python 里默认的json.dumps对中文会转成\uXXXX形式，体积会膨胀。用ensure_ascii=False可以保留中文原样，节省内存。如果对性能要求更高，可以用orjson或者msgpack，序列化速度能快好几倍。

import orjson def serialize_result(result: dict) -> bytes: return orjson.dumps(result) def deserialize_result(data: bytes) -> dict: return orjson.loads(data)

3.3 TTL 设多长才合适

TTL 的设置是个权衡：设太短，缓存命中率低，起不到作用；设太长，数据陈旧，用户拿到过时答案。

我的经验是按业务场景分档：

除了固定 TTL，我还建议加随机抖动。什么意思呢？如果一批缓存同时写入、TTL 又完全一样，它们会在同一时刻集体失效，导致大量请求同时穿透到后端，这就是缓存雪崩。解决办法是在基础 TTL 上加一个随机偏移，比如基础 3600 秒，实际 TTL 在 3300 到 3900 之间随机。

import random def get_ttl_with_jitter(base_ttl: int, jitter_ratio: float = 0.1) -> int: jitter = int(base_ttl * jitter_ratio) return base_ttl + random.randint(-jitter, jitter)

3.4 缓存穿透、击穿、雪崩怎么防

这三个词是缓存领域的经典问题，AI Agent 场景下同样存在，而且因为大模型调用昂贵，后果更严重。

缓存穿透是指查询一个不存在的数据，缓存里没有，数据库里也没有，每次请求都打到后端。在 Agent 场景下，可能是用户问了一个知识库里完全没有的问题。防御手段是缓存空结果：即使没查到，也往 Redis 里写一个特殊标记，TTL 设短一点，比如 60 秒。这样短时间内重复问同一个问题，就不会反复穿透。

缓存击穿是指某个热点 key 突然失效，大量请求同时涌向后端。防御手段是互斥锁：只允许一个请求去重建缓存，其他请求等待或者返回旧值。用 Redis 的SET NX命令可以实现一个简单的分布式锁。

缓存雪崩是指大量 key 同时失效。前面说的 TTL 加抖动就是防御手段之一。另外还可以做多级缓存，本地缓存作为第一层，Redis 作为第二层，即使 Redis 全挂了，本地缓存还能撑一阵。

import redis import time r = redis.Redis(host='localhost', port=6379, decode_responses=False) def get_with_lock(key: str, rebuild_func, ttl: int = 3600): value = r.get(key) if value is not None: return value lock_key = f"{key}:lock" acquired = r.set(lock_key, "1", nx=True, ex=10) if acquired: try: value = rebuild_func() r.set(key, value, ex=ttl) return value finally: r.delete(lock_key) else: for _ in range(50): time.sleep(0.1) value = r.get(key) if value is not None: return value return rebuild_func()

这段代码的逻辑是：先查缓存，没有就尝试拿锁。拿到锁的请求负责重建缓存，没拿到锁的请求轮询等待，最多等 5 秒。如果还没等到，就自己走一遍重建逻辑，保证不会无限阻塞。

提示：锁的过期时间要设得比重建逻辑的最长耗时略长。如果重建要 30 秒，锁只设 10 秒，那锁提前释放了，其他请求又会涌进来。我一般设重建超时时间的 1.5 倍。

4. 实操过程与核心环节实现

4.1 环境准备与 Redis 安装

先把环境搭起来。Linux 上用包管理器装最省事，Ubuntu 下是apt install redis-server，CentOS 下是yum install redis。macOS 上用 Homebrew，brew install redis然后brew services start redis。

装完之后验证一下：

redis-cli ping # 返回 PONG 就说明服务正常

如果你不想在本地装，用 Docker 是最干净的：

docker run -d --name redis-agent \ -p 6379:6379 \ -v /data/redis:/data \ redis:7.2-alpine \ redis-server --appendonly yes --maxmemory 1gb --maxmemory-policy allkeys-lru

Python 客户端我推荐用redis-py，它同时支持同步和异步，和 FastAPI、LangChain 这些框架配合得很好。

pip install redis

连接池的配置有讲究。不要每次操作都新建连接，那样开销很大。用连接池复用连接：

import redis pool = redis.ConnectionPool( host='localhost', port=6379, db=0, max_connections=50, socket_timeout=3, socket_connect_timeout=2, retry_on_timeout=True, health_check_interval=30 ) r = redis.Redis(connection_pool=pool)

max_connections要根据你的并发量来定。太小会导致请求排队等连接，太大又浪费资源。我的经验值是：峰值 QPS 乘以平均操作耗时（秒），再留 50% 余量。比如峰值 1000 QPS，每次操作平均 2 毫秒，那就是 1000 × 0.002 × 1.5 = 3，取整至少 10 个连接。实际我会设 50，留足缓冲。

socket_timeout设 3 秒是个经验值。太短容易误判超时，太长会导致请求堆积。health_check_interval让连接池定期检查连接健康度，避免拿到已经断开的连接。

4.2 在 Agent 主流程中嵌入缓存

假设你的 Agent 主流程是一个函数run_agent(question)，嵌入缓存的改造大概是这样：

def run_agent_with_cache(question: str, user_id: str) -> dict: cache_key = build_cache_key("qa", question) cached = r.get(cache_key) if cached is not None: result = deserialize_result(cached) result["from_cache"] = True return result result = run_agent(question) result["from_cache"] = False ttl = get_ttl_with_jitter(3600) r.set(cache_key, serialize_result(result), ex=ttl) return result

这段代码看起来简单，但有几个细节值得展开。

第一，缓存写入要异步化。如果r.set耗时较长，会拖慢主流程的响应。可以用后台线程或者异步任务来写缓存。Python 里可以用ThreadPoolExecutor，或者如果你的框架支持 async，直接用await r.set(...)。

第二，缓存失败不能影响主流程。Redis 偶尔会抖动，网络会超时。如果写缓存失败导致整个请求报错，那就本末倒置了。所以缓存操作要包在 try-except 里，失败就记日志，继续返回结果。

def safe_cache_set(key: str, value: bytes, ttl: int): try: r.set(key, value, ex=ttl) except redis.RedisError as e: logger.warning(f"cache set failed: {e}")

第三，要记录缓存命中率。这是评估缓存效果的核心指标。我一般在返回结果里带一个from_cache字段，然后在监控系统里统计命中率。如果命中率长期低于 20%，说明缓存策略有问题，要么 key 设计不合理，要么 TTL 太短，要么业务本身重复率就低。

4.3 工具调用层的缓存实现

工具调用缓存和问答缓存略有不同，因为工具的参数往往是结构化的。我的做法是把参数按固定顺序序列化，再哈希。

def build_tool_cache_key(tool_name: str, params: dict) -> str: sorted_params = json.dumps(params, sort_keys=True, ensure_ascii=False) digest = hashlib.sha256(sorted_params.encode()).hexdigest()[:16] return f"agent:tool:{tool_name}:{digest}"

sorted_keys=True很关键。如果参数字典的顺序不同，序列化结果就不同，哈希也不同，会导致明明参数一样却缓存不命中。排序之后，只要参数内容相同，不管传入顺序如何，key 都一致。

工具缓存的 TTL 我一般设得比较短，5 到 15 分钟。因为外部数据变化快，缓存太久可能返回过时信息。但有些工具的结果是稳定的，比如“查询某个城市的经纬度”，这种可以设长一点，几小时甚至几天都行。具体要看工具的性质。

4.4 多轮对话场景的缓存处理

多轮对话是 AI Agent 里比较难处理缓存的场景。因为每一轮的输入都依赖上一轮的上下文，同样的用户问题在不同上下文下答案可能完全不同。

我的处理方式是只缓存单轮独立问答，多轮对话不缓存，或者只缓存最后一轮的完整上下文哈希。具体来说，如果用户的问题是独立的、不依赖历史，就走缓存；如果依赖历史，就把历史对话也纳入 key 的计算。

def build_conversation_cache_key(question: str, history: list) -> str: if not history: return build_cache_key("qa", question) context = "|".join([f"{h['role']}:{h['content']}" for h in history[-3:]]) combined = f"{context}||{question}" return build_cache_key("qa-ctx", combined)

这里只取最近 3 轮历史，是因为更早的对话对当前回答的影响通常很小，全部纳入会让 key 变得很长，而且命中率会大幅下降。3 轮是个经验值，你可以根据业务特点调整。

注意：多轮对话缓存要特别小心隐私问题。如果对话内容涉及敏感信息，缓存时要做脱敏处理，或者干脆不缓存。我一般会在 key 生成前过滤掉手机号、身份证号这类信息。

5. 常见问题与排查技巧实录

5.1 缓存命中率低怎么办

命中率低是最常见的问题。排查思路我总结成一个清单：

我遇到过一个典型案例：key 里带了用户 ID，导致同一个问题不同用户问就是不同的 key，命中率极低。后来把用户 ID 从 key 里去掉，只在返回结果时做权限过滤，命中率从 8% 涨到了 55%。

5.2 Redis 连接超时怎么排查

redis command timed out这个报错我在好几个项目里都见过。原因通常有三种：网络抖动、Redis 负载过高、慢查询阻塞。

排查步骤是这样的。先用redis-cli --latency看 Redis 的响应延迟，正常应该在 1 毫秒以内。如果超过 10 毫秒，说明 Redis 本身有问题。再用redis-cli slowlog get 10看慢查询日志，找出执行时间长的命令。常见的慢查询包括KEYS *、大 key 的HGETALL、SMEMBERS等。

如果是网络问题，检查客户端和 Redis 之间的网络质量，看是否有丢包。如果是负载问题，看 Redis 的 CPU 和内存使用率，考虑扩容或者优化命令。

还有一个容易被忽略的点：连接池耗尽。如果max_connections设得太小，高并发时请求会排队等连接，表现出来也是超时。这种情况要看连接池的监控指标，如果活跃连接数长期接近上限，就该调大了。

5.3 缓存数据不一致怎么处理

缓存和真实数据不一致，是缓存系统的固有难题。在 AI Agent 场景下，主要表现为：知识库更新了，但缓存里还是旧答案。

我的处理策略是版本化 + 主动失效。版本化前面提过，就是在 key 里带版本号，知识库更新时版本号加一，旧缓存自然不再命中。主动失效是指更新知识库后，主动删除相关缓存。

但主动失效有个难点：怎么知道哪些缓存和更新的知识相关？如果 key 里没有知识库的标识，就没法精确删除。所以我在 key 设计时会带上知识库的 ID 或者标签，这样更新时可以按标签批量清理。

def invalidate_by_kb(kb_id: str): pattern = f"agent:qa:*:kb:{kb_id}:*" cursor = 0 while True: cursor, keys = r.scan(cursor, match=pattern, count=100) if keys: r.delete(*keys) if cursor == 0: break

用SCAN而不是KEYS，是因为SCAN是渐进式的，不会阻塞 Redis。count=100表示每次扫描 100 个 key，可以根据实际情况调整。如果 key 数量很大，这个过程可能耗时较长，建议放在后台任务里执行。

5.4 大 key 问题怎么发现和解决

大 key 是指 value 特别大的 key，比如几百 KB 甚至几 MB。在 AI Agent 场景下，如果缓存的是长文本回答，很容易出现大 key。大 key 的危害是：读取时占用带宽多、删除时阻塞 Redis、内存分布不均。

发现大 key 可以用redis-cli --bigkeys，它会扫描所有 key 并报告最大的几个。也可以用MEMORY USAGE key查看单个 key 的内存占用。

解决大 key 的思路是拆分。如果一个回答特别长，可以拆成多个 key 存储，或者只缓存回答的摘要，完整回答存到对象存储里。另一个思路是压缩，用 gzip 或者 zstd 压缩后再存，读取时解压。压缩率通常能到 3 到 5 倍，对文本内容效果很好。

import gzip def compress_and_set(key: str, value: bytes, ttl: int): compressed = gzip.compress(value) r.set(key, compressed, ex=ttl) def get_and_decompress(key: str) -> bytes: data = r.get(key) if data is None: return None return gzip.decompress(data)

压缩会增加 CPU 开销，但相比网络传输和内存节省，通常是划算的。我一般对超过 10KB 的 value 启用压缩。

5.5 缓存预热怎么做

服务刚上线或者 Redis 刚重启时，缓存是空的，所有请求都会穿透到后端。如果后端扛不住，就会出问题。解决办法是缓存预热：在服务正式对外之前，先把热点数据加载到缓存里。

预热的做法有几种。一种是从历史请求日志里提取高频问题，提前跑一遍 Agent 把结果缓存起来。另一种是在低峰期主动触发一些常见查询。还有一种是用定时任务，每天凌晨把核心知识库的常见问答刷新一遍。

def warmup_cache(hot_questions: list): for question in hot_questions: cache_key = build_cache_key("qa", question) if r.exists(cache_key): continue result = run_agent(question) ttl = get_ttl_with_jitter(7200) r.set(cache_key, serialize_result(result), ex=ttl) time.sleep(0.1)

time.sleep(0.1)是为了避免预热时把后端打满。预热是个慢工出细活的过程，不要追求速度，要追求稳定。

6. 监控与持续优化

6.1 必须关注的几个核心指标

缓存上线不是终点，持续监控才能保证它一直有效。我重点关注这几个指标：

命中率是最核心的。计算方式是命中次数 / (命中次数 + 未命中次数)。我一般按小时统计，如果某小时命中率突然下降，就要排查是不是有异常请求或者缓存被清了。

平均响应时间要分命中 and 未命中两组看。命中的响应时间应该在毫秒级，未命中的取决于 Agent 本身的耗时。如果命中组的响应时间也变长了，说明 Redis 有问题。

内存使用率要控制在 70% 以下。超过这个值，淘汰会变得频繁，命中率会下降。如果持续超过 80%，就该考虑扩容了。

淘汰数量（evicted_keys）如果持续增长，说明内存不够用，需要扩容或者优化 key 的存储。

连接数要关注活跃连接和空闲连接的比例。如果活跃连接长期接近上限，说明连接池太小。

6.2 用 Redis 自带命令做快速诊断

不用装额外的监控工具，Redis 自带的命令就能做很多诊断。

INFO stats能看到命中次数、未命中次数、淘汰数量等关键统计。INFO memory能看到内存使用情况。INFO clients能看到连接数。SLOWLOG GET能看到慢查询。

我习惯写一个简单的诊断脚本，定期跑一下，把关键指标打到日志里：

def log_redis_stats(): info = r.info() stats = { "hit_rate": info["keyspace_hits"] / max(info["keyspace_hits"] + info["keyspace_misses"], 1), "used_memory_mb": info["used_memory"] / 1024 / 1024, "connected_clients": info["connected_clients"], "evicted_keys": info["evicted_keys"], "expired_keys": info["expired_keys"] } logger.info(f"redis stats: {stats}")

这个脚本可以挂在定时任务里，每 5 分钟跑一次。时间长了就能看出趋势，提前发现问题。

6.3 缓存策略的迭代方向

缓存不是设好就不管了，随着业务变化要持续调整。

如果发现命中率上不去，可以考虑引入语义缓存。用 embedding 模型把问题向量化，在向量库里找相似问题。这能把“今天天气怎么样”和“今天天气如何”这类问题合并，命中率能提升 10 到 20 个百分点。但要注意误判风险，相似度阈值要调好，太高了命中率低，太低了返回错误答案。

如果发现缓存占用内存太多，可以考虑分级缓存。热点数据放 Redis，冷数据放本地磁盘或者对象存储。或者用 Redis 的EXPIRE命令给不同 key 设不同的过期时间，让冷数据自然淘汰。

如果发现缓存更新不及时，可以考虑主动更新。在知识库变更时，通过消息队列通知缓存服务，主动刷新相关 key。这比被动等 TTL 过期要实时得多。

提示：任何缓存策略的调整，都要先在测试环境验证，再灰度到生产。直接全量改配置，出了问题影响面太大。我一般会先放 10% 的流量到新策略，观察一天，没问题再逐步扩大。

6.4 我踩过的几个坑

最后分享几个实际踩过的坑，都是文档里不会写的。

第一个坑：用decode_responses=True导致二进制数据损坏。如果你的缓存内容包含压缩后的二进制数据，decode_responses=True会尝试用 UTF-8 解码，直接报错。解决办法是保持decode_responses=False，需要字符串时手动 decode。

第二个坑：TTL 设成 0 导致 key 永不过期。Redis 的SET命令，如果不传ex参数，key 就是永久的。我有一次忘了传，结果缓存越积越多，最后内存爆了。现在我的封装函数强制要求传 TTL，不传就报错。

第三个坑：在事务里做缓存操作导致死锁。Redis 的事务（MULTI/EXEC）里如果包含等待锁的操作，很容易死锁。我的建议是缓存操作尽量简单，不要在事务里做复杂的逻辑。

第四个坑：忽略序列化兼容性。如果你的 Agent 服务有多个版本在跑，新旧版本的序列化格式可能不兼容。新版本写入的数据，旧版本读出来可能报错。解决办法是在缓存 value 里带一个格式版本号，读取时先检查版本，不匹配就当作未命中处理。

这些坑看起来都是小问题，但在生产环境里，任何一个小问题都可能被放大成事故。缓存这东西，用好了是利器，用不好就是隐患。希望这些经验能帮你少走点弯路。