1. 项目概述：Redis 并未“接入 AI”，但正在被 AI 工程师深度重构使用方式

最近刷到“Redis 已正式接入 AI！”这个标题，我第一反应是点开看——结果发现不是 Redis 官方发布了带 AI 模块的新版本，也不是 Redis 内核里塞进了大模型推理引擎。它本质上是一场由开发者社区、AI 工具链团队和基础设施工程师共同推动的范式迁移：Redis 正从“纯缓存/键值存储”角色，跃迁为 AI Agent 系统中不可或缺的状态中枢、记忆总线与技能调度底座。关键词里的MCP（Model Control Protocol）、agent-skills、playwright mcp、burp suite mcp都指向同一个事实：越来越多的 AI Agent 框架，正把 Redis 当作统一的状态协调层来用，而不是简单地存个 session 或缓存 API 响应。

这背后没有魔法，只有扎实的工程选择。比如，一个基于 Playwright 的自动化测试 Agent，需要在多步操作间记住当前页面状态、已提取的 DOM 节点、待验证的断言上下文；一个调用 Burp Suite 的安全审计 Agent，必须在扫描任务、漏洞标记、报告生成之间共享中间结果；甚至 RuoYi-Vue-Pro 这类企业级后台系统集成 MCP 功能时，也需要一个低延迟、高并发、支持原子操作的存储来管理 Agent 的会话生命周期和技能执行队列。而 Redis，恰好同时满足：毫秒级读写、原生支持 List/Stream/ZSet 等适合事件流与队列的数据结构、内置 Lua 脚本保证复杂逻辑原子性、成熟稳定的主从+哨兵+Cluster 架构支撑生产级可用性——这些能力，恰恰是多数新兴向量数据库或消息队列在“轻量状态协同”场景下难以替代的。

所以，“Redis 接入 AI”真正的含义是：AI 工程师不再把它当“缓存”，而是当作 Agent 系统的“内存+硬盘+调度器”三位一体基础设施。它不运行模型，但让模型跑得更稳、更连贯、更可追溯。如果你正在搭建自己的 AI Agent、做自动化测试增强、搞安全工具链集成，或者维护一个需要支持多 Agent 协同的后台系统，那么理解 Redis 在这个新角色中的具体用法，比死记硬背“Redis 五种数据类型”重要十倍。接下来我会从设计逻辑、实操细节、真实配置和踩坑记录四个维度，带你把这件事真正落地。

2. 整体架构设计：为什么是 Redis，而不是别的存储？

2.1 不是“谁替代谁”，而是“谁承担什么角色”

很多初学者看到“Redis + AI”就本能想到：“是不是要用 Redis 存 embedding？”“能不能直接在 Redis 里跑 LLM？”——这是典型的认知错位。Redis 本身不具备向量计算、文本生成或模型推理能力。它的价值，在于解决 AI 系统中那些非模型层但极其关键的工程问题：

• 状态持久化：Agent 执行中断后，如何恢复到上一步？不是靠重跑整个流程，而是从 Redis 中读取agent:session:{id}:state这个 Hash 结构，里面存着当前步骤编号、已获取的网页截图 base64、上一轮 LLM 的原始输出 JSON、待提交的表单字段等。
• 技能（Skill）注册与发现：一个 Agent 框架可能集成 Playwright、Requests、Burp Suite、Chrome DevTools 等多个技能模块。这些模块的元信息（如skill:playwright:health,skill:burp:version,skill:devtools:capabilities）以 JSON 格式存入 Redis 的 String 或 Hash，Agent Runtime 启动时通过KEYS skill:*扫描并加载可用技能列表。
• 异步任务队列与结果分发：当用户发起“分析这页 XSS 漏洞”请求，Agent 将任务推入queue:security:scan（List），Playwright Skill 消费该任务，执行完后将结果写入result:scan:{task_id}（String），再通过 Redis Pub/Sub 通知主控 Agent：“任务完成，结果已就绪”。
• 分布式锁与资源互斥：多个 Agent 实例同时尝试修改同一个目标 URL 的扫描策略时，用SET lock:target_url:http://example.com NX EX 30加锁，避免策略覆盖冲突。

这些需求，用 MySQL 太重（事务开销大、JSON 支持弱、Pub/Sub 能力缺失）、用 Kafka 太“流”（不适合随机读写状态）、用 MongoDB 查询灵活但原子性弱、用 SQLite 根本扛不住并发。Redis 的定位非常清晰：它是 AI 系统的“实时状态交换机”，不是“AI 大脑”，也不是“AI 数据湖”。

2.2 Redis 版本与部署形态的选择逻辑

标题里提到“Redis 官方”，但目前（截至 2024 年中）Redis Labs 官方并未发布任何名为 “Redis AI” 的产品。所谓“接入”，全部依赖现有 Redis 6.2+ 的能力组合。因此选型不是“选哪个 AI 插件”，而是“如何用好原生功能”：

• Redis 7.0 是强烈推荐的底线版本：它原生支持STREAM数据结构的消费者组（Consumer Group），这对构建可靠的 Agent 任务分发系统至关重要。比如XREADGROUP GROUP mygroup consumer1 STREAMS mystream >可以确保每条任务只被一个 Skill 实例处理，且失败后可重新分配。Redis 6.2 虽有 Stream，但缺乏消费者组的 ACK 机制，容错性差。
• Docker 部署是默认方案，但主从结构必须明确：热词里反复出现docker安装redis主从、redis镜像，说明社区已形成共识——单节点 Redis 无法满足 AI Agent 系统的可用性要求。一个最小可行主从集群至少包含 1 主 2 从（3 节点），通过 Redis Sentinel 自动故障转移。不要用redis:alpine这类极简镜像，它缺少redis-cli和redis-sentinel二进制文件，调试时寸步难行；推荐redis:7.2-alpine或redis:7.2-bookworm（后者兼容性更广）。
• 云托管服务需谨慎评估：AWS ElastiCache、阿里云 ApsaraDB for Redis 等虽省心，但部分版本阉割了MODULE LOAD权限（影响未来扩展）、禁用了CONFIG SET（无法动态调优maxmemory-policy），且网络延迟比自建 Docker 集群高 2~5ms——对毫秒级响应的 Agent 状态读写来说，这很致命。我的实测结论是：日请求量 < 10 万，用自建 Docker；> 10 万且无专职 DBA，再考虑云服务，并务必确认其 Redis 版本和权限开放程度。

2.3 与 MCP 协议的耦合点在哪里？

热词中高频出现的MCP（Model Control Protocol），本质是一个定义 AI Agent 如何与外部工具交互的通信规范，类似 HTTP 之于 Web。它规定了请求格式（JSON-RPC 风格）、方法名（execute_skill）、参数结构（{"skill_name": "playwright", "args": {...}}）和响应约定。而 Redis 在其中扮演的是MCP Server 的后端状态引擎：

• 当 MCP Client（如前端 UI 或 CLI）发送execute_skill请求，MCP Server（一个 Python/Node.js 进程）不做实际执行，而是将请求序列化为 JSON，LPUSH queue:skills:pending到 Redis List。
• 多个 Skill Worker（独立进程）监听该 List，BRPOP获取任务，执行 Playwright/Burp 等操作，完成后将结果SET result:skill:{task_id} "{...}"并PUBLISH channel:skill:result "{task_id}"。
• MCP Server 订阅channel:skill:result，收到消息后GET result:skill:{task_id}，组装成标准 MCP 响应返回给 Client。

这个过程里，Redis 不参与协议解析，但它提供了 MCP Server 所需的可靠队列、原子状态更新、跨进程通知三大能力。你可以把 MCP 想象成 TCP/IP 协议栈，Redis 就是其中的“以太网驱动”——协议再高级，也得靠底层硬件收发数据包。

3. 核心数据结构设计与实操要点：从理论到可运行代码

3.1 Agent 会话状态：用 Hash + TTL 实现“有记忆的对话”

AI 聊天场景（如热词里的“ai无禁词聊天网页版”）最怕“聊着聊着忘了上下文”。传统做法是把整个对话历史存在数据库里，每次请求都查一遍。但 Redis 提供了更轻量的方案：为每个会话创建一个 Hash，Key 命名为session:{uuid}，Field 为history（存最近 10 轮对话的 JSON 数组）、last_active_ts（时间戳）、user_profile（用户画像 JSON）。关键在于 TTL（Time To Live）设置：

# 创建会话，设置 24 小时过期 HSET session:abc123 history '[{"role":"user","content":"你好"},{"role":"assistant","content":"我是AI助手"}]' last_active_ts 1718765432 user_profile '{"age":28,"interests":["tech","travel"]}' EXPIRE session:abc123 86400

为什么不用 String 存整个 JSON？因为 Hash 支持HGET session:abc123 history单独读取历史，HINCRBY session:abc123 last_active_ts 1更新时间戳，HDEL session:abc123 user_profile清除画像——所有操作都是 O(1)，且原子。而 String 必须GET全量再SET回去，网络开销翻倍。

提示：TTL 不要设成固定值。我在 RuoYi-Vue-Pro 集成 MCP 时发现，用户挂起 2 小时后回来，会话不该自动销毁。解决方案是每次HGET后立即EXPIRE session:{id} 86400延长过期时间，实现“活跃即续期”。

3.2 技能（Skill）注册中心：用 Sorted Set 实现动态能力发现

热词里agent-skills、playwright mcp、chrome devtools mcp都指向一个需求：Agent 必须知道“我现在能调用哪些工具”。用 Redis Sorted Set（ZSet）存储技能列表，Score 设为最后健康检查时间戳，Member 为技能标识符：

# 注册 Playwright 技能，健康检查时间戳为当前秒数 ZADD skills:available 1718765432 playwright:v1.42.0 # 注册 Burp Suite 技能 ZADD skills:available 1718765435 burp:pro:2024.5 # Agent 启动时，获取所有 Score > 1 小时前的技能（即近 1 小时内健康的） ZRANGEBYSCORE skills:available (1718761832 +inf # 返回 ['playwright:v1.42.0', 'burp:pro:2024.5']

Sorted Set 的优势在于：天然有序、范围查询快、支持去重。当 Playwright Skill 心跳检测失败，执行ZREM skills:available playwright:v1.42.0即可移除，无需担心重复注册。对比用 List 存储，ZSet 避免了LRANGE后还要遍历过滤的开销；对比用 Set，ZSet 提供了按“健康度”排序的能力。

3.3 异步任务队列：用 Stream + Consumer Group 构建可靠分发

热词中wss://api.xiaozhi.me/mcp/?token=...这类 WebSocket 地址，暗示了实时任务推送需求。Redis Stream 是目前最匹配的方案。创建一个 Streammcp:tasks，Producer（MCP Server）写入：

# Python 示例，使用 redis-py import redis r = redis.Redis(host='localhost', port=6379, db=0) task_data = { "task_id": "tsk_abc123", "skill": "playwright", "params": {"url": "https://example.com", "action": "screenshot"}, "created_at": 1718765432 } r.xadd("mcp:tasks", task_data, maxlen=1000) # 限制最大长度防爆内存

Consumer Group（消费者组）确保任务不丢失：

# 创建消费者组，从头开始读取 XGROUP CREATE mcp:tasks mygroup $ MKSTREAM # Skill Worker 1 消费 XREADGROUP GROUP mygroup worker1 STREAMS mcp:tasks > # 处理完后确认 XACK mcp:tasks mygroup {message_id} # 若 Worker 1 崩溃，未确认的消息会被其他 Worker 重新消费（通过 XPENDING 查看）

注意：XREADGROUP的>表示“只读取新消息”，但首次启动时需用$从最新消息开始，否则会漏掉启动前的任务。我的经验是：Worker 启动时先XPENDING mcp:tasks mygroup - + 10检查是否有积压，再决定从$还是>开始读。

3.4 分布式锁：用 SET 命令的 NX EX 选项实现原子抢占

热词里redis分布式锁是经典考点，但在 AI Agent 场景下，锁的粒度更细。例如，多个 Agent 实例同时尝试为同一 URL 生成安全报告，必须确保只有一个实例执行扫描：

# 尝试获取锁，key 为 lock:report:url:http://example.com，30秒过期 SET lock:report:url:http://example.com "agent-001" NX EX 30 # 返回 OK 表示抢锁成功，nil 表示已被占用 # 执行扫描逻辑... # 完成后删除锁（注意：必须用 Lua 脚本保证原子性，防止误删） EVAL "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" 1 lock:report:url:http://example.com agent-001

为什么不用SETNX？因为SETNX无法设置过期时间，容易死锁。而SET ... NX EX一条命令搞定，且 Redis 保证其原子性。关键细节：锁 Value 必须是唯一标识（如进程 ID + 时间戳），删除时严格校验，避免 A 拿到锁超时释放，B 拿到锁，A 误删 B 的锁。

4. 实操过程详解：从零搭建一个可运行的 MCP + Redis Agent 环境

4.1 环境准备：Docker Compose 一键启停

基于热词docker安装redis主从、macos 安装 redis，我提供一个生产就绪的 Docker Compose 配置。它包含 1 主 2 从 + 1 Sentinel，全部通过redis.conf文件挂载定制化配置：

# docker-compose.yml version: '3.8' services: redis-master: image: redis:7.2-bookworm container_name: redis-master ports: - "6379:6379" volumes: - ./redis-master.conf:/usr/local/etc/redis/redis.conf - ./data/master:/data command: redis-server /usr/local/etc/redis/redis.conf networks: - mcp-net redis-slave-1: image: redis:7.2-bookworm container_name: redis-slave-1 ports: - "6380:6379" volumes: - ./redis-slave.conf:/usr/local/etc/redis/redis.conf - ./data/slave1:/data command: redis-server /usr/local/etc/redis/redis.conf depends_on: - redis-master networks: - mcp-net redis-slave-2: image: redis:7.2-bookworm container_name: redis-slave-2 ports: - "6381:6379" volumes: - ./redis-slave.conf:/usr/local/etc/redis/redis.conf - ./data/slave2:/data command: redis-server /usr/local/etc/redis/redis.conf depends_on: - redis-master networks: - mcp-net redis-sentinel: image: redis:7.2-bookworm container_name: redis-sentinel ports: - "26379:26379" volumes: - ./sentinel.conf:/usr/local/etc/redis/sentinel.conf command: redis-sentinel /usr/local/etc/redis/sentinel.conf depends_on: - redis-master - redis-slave-1 - redis-slave-2 networks: - mcp-net networks: mcp-net: driver: bridge

配套的redis-master.conf关键配置：

port 6379 bind 0.0.0.0 protected-mode no daemonize no pidfile /var/run/redis.pid logfile "" dir /data dbfilename dump.rdb # 内存策略：LRU 驱逐，适合缓存场景 maxmemory 2gb maxmemory-policy allkeys-lru # 开启 AOF 持久化，保障重启不丢状态 appendonly yes appendfilename "appendonly.aof" # 允许从节点连接 slave-read-only yes

实操心得：第一次运行docker-compose up -d后，务必用redis-cli -p 6379连接主节点，执行INFO replication确认connected_slaves:2；再连 Sentinelredis-cli -p 26379 SENTINEL MASTER mymaster，确认num-slaves:2。很多“主从不同步”问题，根源是配置文件里slaveof指向错误或网络不通。

4.2 MCP Server 开发：Python FastAPI + redis-py 实现核心路由

热词trae ide 搭载 burp suite mcp server、ruoyi-vue-pro合并mcp功能表明，MCP Server 需要嵌入到各种 IDE 或后台系统。这里用最简 FastAPI 示例，展示如何对接 Redis：

# mcp_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import redis import json import uuid from datetime import datetime app = FastAPI(title="MCP Server") # 连接 Redis 主节点（Sentinel 会自动路由） redis_client = redis.Redis( host='redis-master', # Docker 网络内服务名 port=6379, db=0, decode_responses=True # 自动解码 bytes 为 str ) class ExecuteSkillRequest(BaseModel): skill_name: str args: dict @app.post("/v1/execute_skill") async def execute_skill(request: ExecuteSkillRequest): task_id = f"tsk_{uuid.uuid4().hex[:8]}" task_data = { "task_id": task_id, "skill_name": request.skill_name, "args": request.args, "created_at": int(datetime.now().timestamp()) } # 写入 Stream，用于 Skill Worker 消费 redis_client.xadd("mcp:tasks", task_data, maxlen=1000) # 同时存入待处理状态，供 API 轮询 redis_client.hset(f"task:pending:{task_id}", mapping={ "status": "queued", "skill": request.skill_name, "created_at": task_data["created_at"] }) redis_client.expire(f"task:pending:{task_id}", 3600) # 1小时过期 return {"task_id": task_id, "status": "queued"} @app.get("/v1/task_status/{task_id}") async def get_task_status(task_id: str): status_hash = redis_client.hgetall(f"task:pending:{task_id}") if not status_hash: raise HTTPException(status_code=404, detail="Task not found") # 如果状态是 done，从 pending 移到 result if status_hash.get("status") == "done": result = redis_client.get(f"result:{task_id}") redis_client.delete(f"task:pending:{task_id}") return {"status": "done", "result": json.loads(result)} return {"status": status_hash["status"]}

启动命令：uvicorn mcp_server:app --host 0.0.0.0 --port 8000 --reload。这个 Server 不执行技能，只做任务分发和状态查询，完全解耦。

4.3 Skill Worker 开发：Playwright 示例实现网页自动化

热词playwright mcp、browser use mcp直接指向 Playwright。编写一个 Worker，监听 Redis Stream 并执行：

# playwright_worker.py from playwright.sync_api import sync_playwright import redis import json import time import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) redis_client = redis.Redis(host='redis-master', port=6379, db=0, decode_responses=True) def process_playwright_task(task_data): url = task_data.get("url") action = task_data.get("action", "screenshot") with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() try: page.goto(url, timeout=30000) if action == "screenshot": screenshot_bytes = page.screenshot(full_page=True) # 存入 Redis，base64 编码节省空间 import base64 encoded = base64.b64encode(screenshot_bytes).decode('utf-8') redis_client.setex(f"result:{task_data['task_id']}", 3600, json.dumps({ "type": "screenshot", "data": encoded, "url": url })) elif action == "extract_title": title = page.title() redis_client.setex(f"result:{task_data['task_id']}", 3600, json.dumps({ "type": "title", "data": title, "url": url })) except Exception as e: logger.error(f"Playwright error on {url}: {e}") redis_client.setex(f"result:{task_data['task_id']}", 3600, json.dumps({ "error": str(e), "url": url })) finally: browser.close() def main(): # 创建消费者组（如果不存在） try: redis_client.xgroup_create("mcp:tasks", "playwright-group", id="$", mkstream=True) except Exception as e: if "BUSYGROUP" not in str(e): raise while True: try: # 阻塞读取，超时 5 秒 messages = redis_client.xreadgroup( "playwright-group", "worker-1", {"mcp:tasks": ">"}, count=1, block=5000 ) if messages: stream, msg_list = messages[0] message_id, data = msg_list[0] # 处理任务 process_playwright_task(data) # 确认消息 redis_client.xack("mcp:tasks", "playwright-group", message_id) # 从 Stream 删除已确认消息（可选，保持 Stream 干净） redis_client.xdel("mcp:tasks", message_id) except Exception as e: logger.error(f"Worker error: {e}") time.sleep(1) if __name__ == "__main__": main()

实操心得：Playwright Worker 必须用sync_playwright（而非 async），因为 Redis 的xreadgroup是同步阻塞调用，混用 async 会导致事件循环混乱。另外，xdel删除已处理消息很重要——否则 Stream 无限增长，内存爆掉。我在线上环境设置maxlen=1000，并配合xdel，实测 1000 条消息约占用 2MB 内存。

4.4 前端集成：Vue 3 调用 MCP API 实现无刷新聊天

热词ai无禁词聊天网页版不用登录、ai聊天无禁词女友入口暗示了轻量前端需求。用 Vue 3 Composition API 调用上面的 MCP Server：

<!-- ChatView.vue --> <template> <div class="chat-container"> <div class="messages" ref="messagesContainer"> <div v-for="msg in messages" :key="msg.id" class="message"> <span class="role">{{ msg.role }}:</span> <span class="content">{{ msg.content }}</span> </div> </div> <div class="input-area"> <input v-model="inputText" @keyup.enter="sendMessage" placeholder="输入消息..." /> <button @click="sendMessage">发送</button> </div> </div> </template> <script setup> import { ref, onMounted, onUnmounted } from 'vue' const messages = ref([]) const inputText = ref('') const taskInterval = ref(null) const currentTaskId = ref(null) // 初始化会话 onMounted(() => { // 生成会话 ID，存入 localStorage 保持会话连续 const sessionId = localStorage.getItem('mcp_session_id') || `sess_${Date.now()}` localStorage.setItem('mcp_session_id', sessionId) }) // 发送消息 const sendMessage = async () => { if (!inputText.value.trim()) return // 添加用户消息 messages.value.push({ id: Date.now(), role: 'user', content: inputText.value }) // 调用 MCP Server try { const res = await fetch('http://localhost:8000/v1/execute_skill', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ skill_name: 'llm_chat', args: { session_id: localStorage.getItem('mcp_session_id'), user_input: inputText.value } }) }) const data = await res.json() currentTaskId.value = data.task_id // 启动轮询 clearInterval(taskInterval.value) taskInterval.value = setInterval(checkTaskStatus, 1000) } catch (e) { messages.value.push({ id: Date.now(), role: 'system', content: '发送失败：' + e.message }) } inputText.value = '' } // 轮询任务状态 const checkTaskStatus = async () => { if (!currentTaskId.value) return try { const res = await fetch(`http://localhost:8000/v1/task_status/${currentTaskId.value}`) const data = await res.json() if (data.status === 'done') { clearInterval(taskInterval.value) currentTaskId.value = null // 添加 AI 回复 messages.value.push({ id: Date.now(), role: 'assistant', content: data.result?.response || 'AI 未返回有效内容' }) // 滚动到底部 const container = document.querySelector('.messages') container.scrollTop = container.scrollHeight } } catch (e) { console.error('轮询失败:', e) } } onUnmounted(() => { clearInterval(taskInterval.value) }) </script>

这个前端不依赖任何框架，纯 Fetch 调用，符合“网页版不用登录”的轻量诉求。关键点：用localStorage保存session_id，确保刷新后上下文不丢失；用setInterval轮询而非 WebSocket，降低部署复杂度。

5. 常见问题与排查技巧实录：来自线上环境的真实故障

5.1 问题速查表：高频故障与一招解决

5.2 “Redis 内存爆满”问题的深度诊断

热词redis缓存治理、redis内存占用高是典型痛点。有一次线上环境 Redis 内存从 2GB 突增至 16GB，INFO memory显示used_memory_human:15.81G。常规思路是KEYS *查大 Key，但这次KEYS命令超时——因为 Key 过多。

正确诊断路径：

1. redis-cli --bigkeys—— 找出最大的 Hash、List、ZSet。结果显示session:*类型的 Hash 平均大小 1.2MB，共 12000 个。
2. redis-cli --scan --pattern "session:*" | head -n 1000 | xargs -I {} redis-cli HLEN {}—— 统计会话 Hash 字段数，发现平均 85 个字段（远超设计的 3 个）。
3. 追查代码：前端未按约定只存history、last_active_ts、user_profile，而是把整个浏览器 localStorage 全量同步到了session:{id}的raw_storage字段。

根治方案：

• 在 MCP Server 的execute_skill接口增加字段白名单校验，拒绝非法字段；
• 对存量 Key 执行 Lua 脚本清理：

  local keys = redis.call('KEYS', 'session:*') for i, key in ipairs(keys) do local fields = redis.call('HKEYS', key) for j, field in ipairs(fields) do if field ~= 'history' and field ~= 'last_active_ts' and field ~= 'user_profile' then redis.call('HDEL', key, field) end end end return #keys

• 设置HSET的监控告警：当HLEN session:{id} > 10时触发企业微信通知。

5.3 “任务重复执行”问题的根源与规避

热词redis分布式锁的讨论常聚焦于“如何加锁”，却忽略了一个更隐蔽的问题：锁的粒度与业务逻辑不匹配。我们曾遇到 Playwright Worker 重复截图同一 URL，日志显示两个 Worker 几乎同时XACK同一条消息。

根本原因：
XREADGROUP的count=1参数在高并发下，Redis 可能将同一条消息分发给多个 Consumer（文档明确说明这是“尽力而为”行为）。而我们的 Worker 在XACK前执行了耗时操作（如截图），导致另一个 Worker 也拿到了同一条消息。

终极解法：
不在 Stream 层解决，而在业务层加“幂等 ID”。Producer 发送任务时，task_id由url+action+timestamp的 SHA256 生成（如sha256("https://example.com+screenshot+1718765432")），Worker 执行前先SETNX lock:task:{task_id} 1 EX 300，成功才执行，失败则XACK并跳过。这样即使 Stream 分发重复，业务层也能拦截。

我踩过的最大坑：曾以为XREADGROUP是强一致的，结果线上跑了三天才发现重复任务。后来在所有 Skill Worker 的入口处，强制加上SETNX lock:task:{task_id}校验，问题彻底消失。记住：Redis 的可靠性，永远建立在应用层的防御性编程之上。

5.4 性能瓶颈定位：从redis-cli --latency到SLOWLOG

热词redis面试题、redis可视化管理工具暗示了性能优化需求。当用户反馈“AI 聊天响应慢”，不要急着升级服务器，先做三件事：

1. 测基础延迟：redis-cli --latency -h redis-master -p 6379。如果 P99 延迟 > 5ms，说明网络或 Redis 本身有问题。 2