最近被问得最多的问题，就是“Redis 怎么接 AI”。很多人看到“Redis 已正式接入 AI”这种说法，第一反应是觉得 Redis 成了一个能跑大模型的新平台，第二反应是到处找 Redis 官方出的 AI 插件包。我在几个实际项目里折腾了一圈，想认真把这件事说清楚：Redis 在 AI 应用里的角色，从来不是“跑模型”，而是给 AI 应用当数据底座。文章会从成本、延迟、会话状态、向量检索、多 Agent 协同这几个维度逐一拆开，再给出可以直接抄走的部署方案和排错记录，适合正在给 AI 应用做后端服务、或者考虑用 Redis 省大模型调用费用的团队参考。
1. AI 应用为什么突然都需要 Redis
先聊一个现象。去年我们团队做一个知识库问答产品，最开始所有逻辑都在 Python 进程里：调用一次大模型，把返回结果原样丢给前端，多轮对话历史塞在内存列表里，用户请求一多，服务直接被打满。那时候我才意识到，AI 应用的后端，本质上还是在处理三件非常“传统”的事：缓存、状态、检索。
1.1 大模型 API 的成本和延迟逼出第一层需求
大模型接口是按 Token 计费的，一次请求动辄几百毫秒到几秒。用户经常会在会话里反复问同一个问题，只是换了种说法；如果每次都去调用模型，钱包和响应时间都受不了。最朴素的做法就是把模型回答缓存起来，而 Redis 的字符串、哈希结构、TTL 过期机制，恰好都是为这种场景准备的。你可以把它理解为“给大模型回答加了一块免重复计算的前置挡板”。
1.2 多轮对话状态不再只属于单个进程
传统 Web 应用做 session 管理时，Redis 就已经是标配。AI 应用也一样：服务拆成多个无状态实例后，某一轮对话落到实例 A，下一轮落到实例 B，如果上下文只存在进程内存里，用户就会觉得这个机器人“失忆”了。把多轮消息、临时变量、任务上下文放进 Redis，让所有实例共享同一份状态，是 AI 应用后端绕不开的架构选择。
1.3 向量检索成为标配，Redis 顺势补齐
RAG（检索增强生成）成为主流之后，知识库问答、私有文档分析这类场景都要求先把文本切块、做 Embedding，再根据用户问题做相似度检索。过去这套东西要么用专业的向量数据库，要么用 Elasticsearch 加插件，部署成本都不低。Redis 从模块时代就开始支持向量检索，到了 Redis Stack 版本已经做得比较成熟，可以让团队少维护一个组件，直接用 Redis 同时承载缓存、状态和向量检索三件事。
Redis 传统用法和 AI 场景用法的对比如下：
| 职责 | 传统应用用法 | AI 应用用法 |
|---|---|---|
| 缓存 | 缓存商品详情、接口响应 | 缓存大模型回答、Embedding 结果 |
| 状态 | 管理用户登录 Session | 管理多轮对话上下文、Agent 运行状态 |
| 数据结构 | List 做消息队列，Set 做标签 | Stream 做事件流，ZSet 做排序和 TopK |
| 检索 | 不涉及 | 向量索引、相似度检索、语义缓存 |
| 锁与协同 | 秒杀防超卖 | 防止多个 Agent 重复处理同一任务 |
2. 第一项接入：语义缓存挡掉重复的大模型调用
2.1 从“精确相同问题”开始做缓存
最简单的一版缓存，就是把用户的问题原文直接当 key。用户问“什么是 Redis 的数据类型”，第二次再问完全一样的句子，直接从 Redis 返回上次的答案，完全不调用大模型。实现只需要几十行代码：
import redis import json r = redis.Redis(host="localhost", port=6379, decode_responses=True) def call_llm(prompt: str): # 替换成你实际使用的大模型接口 return {"text": "模拟模型返回的答案", "prompt": prompt} def chat_with_cache(user_query: str): cache_key = f"llm:resp:{user_query}" cached = r.get(cache_key) if cached: return json.loads(cached) answer = call_llm(user_query) r.setex(cache_key, 3600, json.dumps(answer)) return answer这个版本上线之后，我们立刻发现一个问题：一模一样的问题重复率并没有想象中高。用户经常把问题换个说法再问，比如“帮我总结一下合同里的付款条款”和“请概括一下付款方式”，意思高度接近，但字符串 key 完全不同，缓存根本命中不了。于是我们需要从“字符匹配”升级到“语义匹配”。
2.2 语义命中为什么能拦截更多重复调用
语义匹配的思路是：先把用户问题转成向量，然后在已缓存的向量集合里找最相似的一条；如果相似度足够高，就认为新旧问题语义相同，直接用上一次的答案。这就是语义缓存。它的核心不是“存回答”，而是“判断这次调用与历史上哪次调用最像”。
要实现它，需要三个部件：文本转向量模型、Redis 的向量索引、一个相似度阈值。向量模型用的是 SentenceTransformer 或云厂商的 Embedding 接口，Redis 负责存储和查询向量，阈值决定命中的松紧程度。
2.3 语义缓存的完整代码实现
下面这段代码基于 Redis Stack 的向量搜索能力。索引怎么建、参数什么意思，我会在第 3 节详细解释，先看整体流程：
import redis import uuid import numpy as np from sentence_transformers import SentenceTransformer r = redis.Redis(host="localhost", port=6379, decode_responses=False) model = SentenceTransformer("BAAI/bge-small-zh-v1.5") def embed(text: str): vec = model.encode(text, normalize_embeddings=True) return np.array(vec, dtype=np.float32).tobytes() def call_llm(question: str): return f"模拟回答：{question}" def chat_with_semantic_cache(question: str, threshold: float = 0.88): q_vec = embed(question) # 在向量索引中找与当前问题最相似的一条缓存 results = r.execute_command( "FT.SEARCH", "idx_llm_cache", "*=>[KNN 1 @vec $Q AS score]", "PARAMS", "2", "Q", q_vec, "SORTBY", "score", "ASC", "RETURN", "3", "answer", "question", "score", "DIALECT", "4" ) if results and len(results) > 1: fields = results[2] score = float(fields[fields.index("score") + 1]) cosine_sim = 1 - score # COSINE 距离转相似度 if cosine_sim >= threshold: return fields[fields.index("answer") + 1] answer = call_llm(question) r.hset( f"llm:semantic:{uuid.uuid4().hex}", mapping={ "question": question, "answer": answer, "vec": q_vec, }, ) return answer注意几个细节。第一，decode_responses=False很重要，因为向量是二进制数据，如果用解码模式，向量会被尝试转成字符串，直接破坏数据。第二，score是余弦距离，越小表示越相似，所以程序里要1 - score才是相似度。第三，RETURN里要手动带上score，否则查询结果里拿不到用于判断的值。
2.4 阈值和过期策略的调优经验
阈值直接影响两个指标：命中率和错误命中率。阈值设得太低，比如 0.75，很多“意思沾边但实际想问不同事”的问题会被错误复用答案；阈值设得太高，比如 0.98，语义缓存退化成精确缓存，省不了多少调用。我们一般从 0.85 开始调，观察一周的命中率和用户反馈，再上下微调。
TTL 策略也有讲究。知识库类问题，答案稳定性高，建议缓存 24 小时以上；促销政策、实时价格这类信息一天变好几轮，缓存时间缩到 30 分钟。另外我建议用“滑动过期”而不是“固定过期”：每次命中缓存时都重新执行一次EXPIRE，等于把这条缓存的寿命续上，热门问题会被长期保留，冷门问题自然淘汰，内存利用率更高。
3. 第二项接入：用 Redis 搭建轻量级向量检索
语义缓存只是向量检索的一个应用场景。只要你在做 RAG 或者知识库问答，就一定需要“把用户问题向量化，然后在向量库里找最相似内容”的能力。这里讲讲怎么用 Redis 搭一个能上线的轻量级向量检索。
3.1 什么场景该选 Redis，而不是专用向量库
选型前先看清边界。如果你们的向量规模在上百万条以内，Redis 完全能扛住；如果团队已经在用 Redis，再省一个组件，运维成本很低。但如果你要做十亿级向量检索、需要复杂的多租户隔离、或者要求毫秒级高并发查询，还是得用 Milvus、Weaviate 这类专业向量数据库。Redis 在这里的优势是“一体多用”：缓存、状态、队列、向量检索都在一个系统里，数据不用在多个存储之间搬来搬去。在我们那个知识库项目里，文本块和向量存在同一个 Redis key 下，查询时一次拿到相似内容和元数据，省了业务层的二次回表。
3.2 创建向量索引的参数含义
Redis 的向量检索能力来自 RedisSearch 模块。建索引的命令如下：
FT.CREATE idx_chunks ON HASH PREFIX 1 chunk: SCHEMA \ title AS TEXT \ content AS TEXT \ vec AS VECTOR HNSW 6 \ TYPE FLOAT32 \ DIM 768 \ DISTANCE_METRIC COSINE \ M 40 \ EF_CONSTRUCTION 200这里重点说三个参数：
TYPE FLOAT32表示向量的每个维度用 4 字节浮点数存储，768 维的向量一条就是约 3KB。DIM 768必须和实际 Embedding 模型输出的维度一致。用 bge-large 就是 1024，用 bge-small-zh 是 512，写错会直接报错或者查询结果全是垃圾。M 40和EF_CONSTRUCTION 200是 HNSW 图算法的两个参数。M 是每个节点的最大连接数，越大检索越准，但内存和构建时间也越高；EF_CONSTRUCTION 是构建时候选集的规模，越大索引质量越高。初次尝试可以用 M=16 到 40、EF_CONSTRUCTION=200，后续根据召回率慢慢调。
3.3 写入向量和执行 KNN 查询
写入向量不用专门的命令，把向量二进制塞进普通哈希就行：
HSET chunk:1001 title "Redis 的数据类型" content "Redis 支持五种基本数据类型：字符串、列表、哈希、集合、有序集合" vec "0.001,0.002,...,0.768维度"实际生产代码里，向量字段应该是 Embedding 模型输出的二进制字节流，而不是逗号分隔的文本。Python 侧这么写：
import numpy as np from redis.commands.search.query import Query def add_chunk(doc_id, title, content, vector): r.hset( f"chunk:{doc_id}", mapping={ "title": title, "content": content, "vec": np.array(vector, dtype=np.float32).tobytes(), }, ) def search_chunks(query_vec: bytes, top_k: int = 5): q = ( Query("*=>[KNN $K @vec $QUERY AS score]") .sort_by("score") .return_fields("title", "content", "score") .dialect(4) ) params = {"K": top_k, "QUERY": query_vec} result = r.ft("idx_chunks").search(q, query_params=params) return result.docs这里面KNN $K @vec $QUERY是 RedisSearch 的向量查询语法，结果会按距离升序排列，score越小越相似。注意到我用了$占位符，实际查询时必须通过query_params传入，不能直接拼字符串，否则遇到二进制数据会出各种编码问题。
3.4 FLAT 与 HNSW 的选择
RedisSearch 支持两种索引算法：FLAT（暴力扫描）和 HNSW（近似最近邻）。FLAT 会把所有向量逐一比对，结果绝对准确，但数据量上去以后查询耗时线性增长；HNSW 是近似算法，先建图再跳转搜索，速度极快，但索引构建耗时和内存占用更高，而且召回率会受参数影响。
实际项目里我是这样选的：
| 场景 | 推荐算法 | 理由 |
|---|---|---|
| 数据量小于 10 万条 | FLAT | 全量遍历毫秒级，结果精确，省内存 |
| 数据量大、对召回率要求极高 | FLAT | 不接受任何近似误差 |
| 数据量大、QPS 高 | HNSW | 查询速度快，配合 M 和 EF 参数控制准确度 |
| 语义缓存这类重复查询 | HNSW | 响应时间敏感，容忍极小概率偏离 |
从 FLAT 换到 HNSW，不要只改索引算法。HNSW 的构建过程会占用 CPU，大批量写入时要错开业务高峰，否则会把 Redis 的 CPU 占满，其他缓存操作跟着遭殃。
4. 第三项接入：多 Agent 的公共状态层
除了缓存和检索，AI 应用还有一个经常被忽视的需求：多个 Agent、多个服务实例之间怎么协作。我们做 Agent 编排的时候，把 Redis 当成了所有人的“公共大脑”，主要解决三类问题。
4.1 会话状态在多实例间共享
无状态服务 + Redis 存状态，是所有水平扩展方案的基础。对话消息我用 Redis List 存储，按时间顺序RPUSH，读取时取最后 N 条拼到 prompt 里：
def save_message(session_id, role, content): key = f"ai:session:{session_id}:messages" r.rpush(key, json.dumps({"role": role, "content": content})) r.expire(key, 1800) # 30 分钟无操作自动清理 r.ltrim(key, -20, -1) # 最多保留最近 20 条，防止 prompt 过长这里有两个细节值得说。一是ltrim一定要做，LLM 的输入长度有限，塞几百条历史消息进去只会白白烧 Token。二是过期时间设 30 分钟而不是永久，绝大多数对话场景不会超过这个窗口，状态不清理的话 Redis 会被无用的 Session 撑爆。
4.2 用分布式锁防止 Agent 重复调度
多个 Agent 同时监听任务时，很容易出现“一个任务被两个 Worker 同时处理”。Redis 分布式锁是标准解法：
import time import uuid def acquire_lock(task_id: str, expire_seconds: int = 60): lock_key = f"ai:lock:task:{task_id}" lock_value = uuid.uuid4().hex ok = r.set(lock_key, lock_value, nx=True, ex=expire_seconds) if ok: return lock_value return None def release_lock(task_id: str, lock_value: str): lock_key = f"ai:lock:task:{task_id}" # 用 Lua 脚本保证“判断持有者+删除”的原子性 script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ r.eval(script, 1, lock_key, lock_value)注意锁一定要设置过期时间，否则某个 Worker 处理到一半挂了，锁永远不释放，后续任务全部卡死。我也见过很多人不加锁值判断，直接DEL，结果误删了别人的锁，导致并发失控。上面的 Lua 脚本就是为了解决“删除超时旧锁”和“误删新锁”的经典问题。
4.3 任务队列与流水线编排
Agent 执行一个复杂任务时，往往要拆成多个步骤：先检索，再总结，再生成图表。每个步骤可以被不同 Worker 消费。用 Redis List 就能实现一个轻量队列：
# 生产者 r.lpush("ai:agent:summary", json.dumps({"doc_id": "1001", "session_id": "s01"})) # 消费者 def worker(): while True: task_data = r.brpop("ai:agent:summary", timeout=30) if task_data is None: continue task = json.loads(task_data[1]) print(f"处理任务：{task}")brpop是阻塞式读取，没有任务时 Worker 会休眠等待，而不是空转打满 CPU。消息量再大一些可以换 Redis Stream，支持消费者组、消息确认、重试这些队列正经需求；但中小项目用 List 已经足够，少引一层 Kafka。
4.4 Key 命名、内存上限和数据隔离
多 Agent 系统里 Key 极其容易乱。我建议统一用三段式命名：ai:{项目名}:{模块}:{具体ID}。比如ai:legal:session:u100:messages和ai:legal:lock:task:xxx。这样有个直接好处：可以通过SCAN ai:legal:*分批清理某个项目的数据，排查问题时也能一眼看出数据归属。
内存治理上，区分三类数据：缓存类给 TTL，会话状态给短 TTL，业务配置不给 TTL。不要让所有数据都永久保留，否则一次异常流量就可能让 Redis 内存涨满。另外巡检内存时用INFO命令看used_memory和maxmemory的比值，超过 80% 就要准备扩容或者清理。
5. 部署形态与生产参数调优
前面讲的都是功能，但 Redis 接 AI 能不能稳定跑在生产环境，关键还要看部署和配置。
5.1 版本与模块选择
我的建议非常直接：新项目直接用 Redis Stack，官方把 RedisSearch、JSON、Bloom Filter 这些模块打包在一起了，一条命令就能把向量检索用起来。如果你所在公司不允许使用附加模块，只能用原生 Redis，那向量检索就得另想办法，Redis 就老老实实只负责缓存、队列和状态，避免自己编译模块踩坑。
版本上选 7.x 或者更高的稳定版，不建议用老旧的 5.x、6.x，向量检索和 Stream 的很多重要改进都没有。
5.2 Docker 部署的关键参数
本地开发和测试可以直接用官方镜像：
docker run -d \ --name redis-stack \ -p 6379:6379 \ -v ./redis-data:/data \ -e REDIS_ARGS="--appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru" \ redis/redis-stack-server:latest--appendonly yes必须加，RDB 快照有丢数据窗口，AOF 能在 Redis 重启后恢复到最新状态。--maxmemory 2gb是给容器设硬上限，避免向量数据无限增长拖垮整台机器。--maxmemory-policy allkeys-lru是内存满了之后的淘汰策略，下面单独说。
5.3 向量数据的内存估算公式
向量检索最占内存的就是向量本身。一条 768 维 FLOAT32 向量的裸数据是768 * 4 = 3072字节，100 万条就是约 3GB。这还只是裸数据，HNSW 索引需要额外的图结构，通常会让内存占用翻到 1.5 到 2 倍，也就是说 100 万条 768 维向量，粗略按向量字节数 * 1.8估，约 5.4GB。
这个估算很重要。我们第一次上线就是没算内存，1000 万文本块灌进去，Redis 直接打满了 8GB。建议你先拿一小批数据实测每条平均开销，再估算全量需求，然后乘以 1.5 的余量去申请内存。
5.4 淘汰策略和持久化策略
allkeys-lru适合缓存场景，AI 项目里的模型回答缓存、Embedding 缓存都适合这个策略。但要注意：会话状态和分布式锁也被当作普通 key 淘汰掉了。会话丢了用户顶多重开一次对话，问题不大；但如果锁 key 被淘汰，可能出现并发重复执行。所以对重要的任务锁，可以单独用一个 Redis 实例，或者把锁的过期时间设置得足够短，让它被淘汰后能快速恢复。
持久化方面，生产环境我建议同时开启 RDB 和 AOF。RDB 负责快速恢复启动，AOF 负责减少数据丢失窗口。向量索引属于“可重建数据”，即使全丢了也能重新灌一遍，所以不要为了它过度追求持久化，反而应该关注重启后向量索引的恢复时间。
6. 五个实战踩坑记录
最后这部分，基本是我每次分享都会被追问的环节。这些坑都不是大问题，但每个都能让人折腾半天。
6.1 decode_responses 与二进制向量冲突
第一次把向量写进 Redis 时，客户端配置了decode_responses=True，结果二进制字节流被解码成字符串再写入，查询时怎么比对都不对。排查很久才发现是客户端连接配置的问题。凡是存向量的连接，必须用decode_responses=False，读取后手动处理编码。
6.2 模型版本切换导致向量维度不匹配
团队里有人升级了 Embedding 模型，旧数据是 768 维，新写入的向量是 1024 维，查询时 Redis 直接返回维度不匹配的异常，而且异常信息非常隐晦。解决方案是建立模型版本校验：每条记录的 key 里带上模型版本号，比如chunk:model_bge_small:v1:1001，升级模型时从新版本号开始写，老数据逐步迁移，不混用。
6.3 大 Key 扫描阻塞 Redis
有一次测试环境 Redis 卡顿严重，排查发现一个 Agent 的会话消息列表没有做ltrim，几十万条消息堆在一个 List Key 里；而监控脚本又用了KEYS *扫描全库，遇到这个超大 Key 时直接阻塞了 Redis。后来监控改用SCAN，并且把所有 Key 长度过大的风险点都设了阈值告警。这里要提醒一句：生产环境千万不能用KEYS *，数据量大一点就是事故。
6.4 异步框架下的连接池耗尽
FastAPI 对接 Redis，路由函数里每次请求都新建一个连接，并发一高，连接数直接撑破 Redis 的maxclients。后来改成全局复用连接池：
pool = redis.ConnectionPool( host="localhost", port=6379, decode_responses=False, max_connections=50, ) r = redis.Redis(connection_pool=pool)连接池大小要根据服务的 Worker 数和单请求的 Redis 操作数来估。另外异步场景直接用redis.asyncio的客户端，不要在线程池里硬塞同步客户端，省得连接上下文频繁切换。
6.5 重启后向量索引丢失
Redis 重启后，FT._LIST查不到之前的向量索引，但普通数据还在。原因是向量索引属于模块维护的元数据，只做 RDB 持久化时，模块加载顺序或者索引元数据恢复都可能出问题。解决方法是：启动命令里显式加载模块，并把 AOF 持久化打开。真实生产环境要写一个“启动后检查索引是否存在，不存在则自动重建”的初始化任务，这样不管什么原因丢了索引，服务都能自愈。
做了一圈下来，我的个人体会是：Redis 和 AI 的结合点，其实比想象中要朴素。它靠的不是什么黑魔法，而是把大模型调用链路里那些重复、共享、协作的问题，用一款成熟的基础组件接住。你不需要一开始就把架构搞得很复杂，先把语义缓存跑起来，再把状态和队列接上，最后根据业务量决定要不要让 Redis 承担向量检索，这条路走下来最稳。真要在生产环境长期跑，记得从第 5 节的部署参数开始核对，第 6 节的坑提前规避，能省下不少半夜拉群的力气。
