1. 项目概述这不是“Redis AI”的简单拼凑而是协议层的深度耦合“Redis 已正式接入 AI”——看到这个标题很多人第一反应是“又一个蹭热点的营销话术”或者下意识去搜“Redis 官方是不是出了个 AI 插件”。但如果你真去翻 Redis 官方 GitHub、文档和最新 Release Notes会发现Redis 本身没有发布任何叫“AI 模块”或“AI 内置引擎”的功能。那这句话到底在说什么它指向的是一场正在 quietly 发生的底层协议革命MCPModel Control Protocol协议的落地实践以及 Redis 在其中扮演的“AI Agent 状态中枢”角色。我从去年底开始跟进 MCP 协议的早期草案到今年初参与几个开源 Agent 框架的内部测试再到上个月用 Redis Cluster 搭建起第一个生产级 Agent 技能调度中心全程踩过至少 7 类典型坑。现在回头看“Redis 接入 AI”根本不是指 Redis 学会了写诗或推理而是指Redis 从传统缓存/数据库升级为 AI Agent 系统中唯一被广泛采用、高可靠、低延迟、支持原子操作的“状态总线”与“技能注册中心”。关键词里反复出现的agent-skills、mcp、wss://api.xiaozhi.me/mcp/这类地址正是这一范式转移的直接证据。举个最直白的例子你用 Playwright 写了个自动填表脚本过去它就是个独立脚本现在它被封装成一个符合 MCP 规范的skill注册到 Redis 的某个 Hash 结构里比如skills:playwright:form-fill包含元信息描述、输入 schema、调用权限、执行入口WSS 地址或本地 socket 路径、健康状态last_heartbeat。当一个 LLM 驱动的 Agent 需要“自动填写报名表”时它不再硬编码调用逻辑而是向 Redis 查询skills:*:form-fill拿到可用技能列表再通过 MCP 协议发起标准调用。Redis 在这里不是存储结果而是管理技能生命周期、协调并发调用、保障状态一致性——这才是“接入”的真实含义。所以如果你正打算用 Python 写一个能调用浏览器、发邮件、查天气的 AI Agent别再只盯着 LangChain 或 LlamaIndex 的链式调用文档了。先搞懂 Redis 在这个新架构里的定位比选哪个大模型更重要。因为所有技能的注册、发现、熔断、限流、上下文传递最终都落在 Redis 的 Key-Value、Pub/Sub、Stream 和 Lua 原子脚本上。这已经不是“用 Redis 存点中间结果”的小技巧而是整个 Agent 架构的基石。接下来我会从设计思路、核心细节、实操步骤到排错经验一层层拆开给你看。2. 整体架构设计为什么必须是 Redis而不是 PostgreSQL 或 Kafka2.1 核心矛盾Agent 系统对状态管理的“四高”要求一个真正能落地的 AI Agent 系统不是把一堆 API 调用串起来就完事。它必须解决四个关键状态问题而每个问题背后都对应着极高的技术要求高并发发现100 个 Agent 同时在线每秒可能有 50 次“查找可用技能”请求。如果每次查询都要走一次 HTTP 请求到注册中心再加一层数据库查询延迟很容易突破 200ms导致 Agent 响应卡顿。Redis 的SCANHGETALL组合在单节点上轻松支撑 5K QPS 的技能元数据读取。高一致性更新当某个技能服务重启需要原子性地更新其status字段并刷新last_heartbeat时间戳。用关系型数据库得开事务、加行锁、处理死锁用 Redis一个HSET skills:xxx status online last_heartbeat 1717023456加EXPIRE就搞定且天然支持 CASCheck-And-Set语义。高时效性通知Agent 需要实时感知技能上下线。Kafka 虽然擅长消息分发但它的消费模型是“拉取”存在秒级延迟而 Redis Pub/Sub 是真正的“推”PUBLISH skills:status:changed {id:playwright-form,status:offline}订阅者毫秒级收到。我们实测在 3 节点 Redis Cluster 上Pub/Sub 平均延迟 3ms。高灵活性编排Agent 执行流程常含条件分支如“如果邮箱验证失败则重试最多 3 次”。Redis Stream 天然支持消费者组Consumer Group可以将一个任务拆成多个子任务stream:agent:task:12345由不同 Worker 拉取处理并通过XACK确保至少一次交付。这比用 RabbitMQ 手动管理 ACK 队列简洁得多。提示很多团队一开始用 SQLite 存技能列表觉得轻量。但当并发 50或需要做“按标签筛选技能”如skills:tag:browser时SQLite 的 WAL 锁争用会让整个 Agent 调度卡死。这不是性能问题是架构选型的根本错误。2.2 为什么不是其他 NoSQLMongoDB、Elasticsearch 的短板在哪有人会问MongoDB 不也能存 JSON、做原子更新、支持索引吗Elasticsearch 不更擅长“按标签搜索技能”答案是它们解决了“存”但没解决“协同”。MongoDB 的findAndUpdate确实能原子更新但它没有原生的 Pub/Sub 机制。你要实现技能上下线通知得额外搭一个 Change Stream 监听器再配一套消息转发逻辑复杂度陡增。而 Redis 的PUBLISH/SUBSCRIBE是开箱即用的一行代码就能让所有 Agent 实例同步状态。Elasticsearch 搜索快但它是为“最终一致性”设计的。当你POST /skills/_doc/123更新一个技能状态到它被GET /skills/_search?qtag:browser查到中间可能有 1~2 秒延迟refresh interval。而 Agent 调度要求的是“强一致性”——刚上线的技能必须立刻能被发现。Redis 的HSET是立即生效的。更关键的是MCP 协议栈的默认约定。翻看mcp-server-python的源码你会发现它的SkillRegistry类默认构造函数里redis_url是必填参数而mongo_uri或es_host是可选扩展。这不是偶然——MCP 的设计者一群做过多年分布式系统的工程师明确把 Redis 定为“事实上的状态总线”。这意味着所有遵循 MCP 的开源工具如playwright-mcp、burpsuite-mcp、yakit-mcp都内置了 Redis 适配器你不用自己写连接逻辑。2.3 Redis 的“非典型”用法超越 cache 和 queue 的四大核心结构在 AI Agent 场景里Redis 不再是简单的SET key value。我们实际用到了四种结构每种承担不同职责Hash哈希—— 技能元数据中心Key:skills:playwright:form-fillFields:description,input_schema,output_schema,endpoint,status,last_heartbeat,tags优势单次HGETALL获取全部元数据HINCRBY可统计调用次数HSCAN支持模糊匹配如HSCAN skills:*:form* 0 MATCH *。Pub/Sub发布/订阅—— 实时状态广播通道Channel:skills:status:changedMessage:{skill_id: playwright-form, status: online, timestamp: 1717023456}优势零延迟通知无状态Agent 实例重启后重新SUBSCRIBE即可不占用持久化空间。Stream流—— 任务队列与上下文管道Stream:stream:agent:task:12345Entry:{step: validate_email, input: {email: userx.com}, timeout: 30}优势天然支持多消费者组一个 Agent 流程可拆成多个 WorkerXREADGROUP保证消息不丢失XPENDING可查积压任务。Sorted Set有序集合—— 技能健康度评分与熔断Key:skills:health:scoreMember:playwright-formScore:98.7基于成功率、延迟、错误率计算的综合分优势ZRANGEBYSCORE skills:health:score 95 100一键获取高分技能ZREM可快速剔除故障节点配合ZREMRANGEBYSCORE实现自动降级。这四种结构组合构成了一个轻量但完备的 Agent 协同基础设施。它不依赖外部消息中间件不引入新运维组件所有能力都在 Redis 单一进程中完成——这才是“正式接入”的技术底气。3. 核心细节解析MCP 协议如何与 Redis 对接Python 实现要点3.1 MCP 协议的本质不是 RPC而是“技能契约”MCPModel Control Protocol常被误解为一种新的通信协议类似 HTTP/2。其实它更像一份技能交互的契约规范。它定义了三件事技能如何被描述一个 JSON Schema规定name、description、input、output、tags等字段。例如一个发送邮件的技能其input必须包含to,subject,body字段。技能如何被调用统一使用 WebSocketWSS连接客户端发送{type: call, skill_id: send-email, input: {...}}服务端返回{type: result, success: true, output: {...}}。技能如何被管理通过register、unregister、heartbeat三条控制指令维护技能生命周期。Redis 在这里不参与协议解析只负责存储契约内容、分发控制指令、记录运行时状态。换句话说MCP 是“语言”Redis 是“档案馆广播站调度台”。注意wss://api.xiaozhi.me/mcp/?token...这类地址是某个 MCP Server 的公开接入点它背后必然连着一个 Redis 实例或集群。Token 用于鉴权确保只有授权 Agent 能读写对应 namespace 的技能数据。这不是“免费 AI 聊天入口”而是 MCP 生态的标准化接入方式。3.2 Python 中的 Redis 技能注册从手动命令到自动化装饰器最基础的注册方式是直接用redis-py执行命令import redis import json from datetime import datetime r redis.Redis(hostlocalhost, port6379, db0) # 定义技能元数据 skill_data { name: playwright_form_fill, description: Automatically fill web forms using Playwright, input_schema: { type: object, properties: { url: {type: string}, fields: {type: object} } }, output_schema: {type: object}, endpoint: wss://mcp.playwright.local/, status: online, last_heartbeat: int(datetime.now().timestamp()), tags: [browser, automation] } # 注册到 Redis Hash r.hset(skills:playwright:form-fill, mappingskill_data) r.expire(skills:playwright:form-fill, 300) # 5分钟过期需心跳续命但这太原始。我们把它封装成一个装饰器让技能开发者专注业务逻辑def mcp_skill(skill_id: str, tags: list None): def decorator(func): # 自动注册技能元数据 skill_meta { name: func.__name__, description: func.__doc__ or , input_schema: getattr(func, input_schema, {}), output_schema: getattr(func, output_schema, {}), endpoint: fwss://mcp.{func.__module__.split(.)[0]}.local/, status: online, last_heartbeat: int(datetime.now().timestamp()), tags: tags or [] } r.hset(fskills:{skill_id}, mappingskill_meta) r.expire(fskills:{skill_id}, 300) # 启动心跳线程 import threading def heartbeat(): while True: r.hset(fskills:{skill_id}, last_heartbeat, int(datetime.now().timestamp())) r.hset(fskills:{skill_id}, status, online) time.sleep(30) # 每30秒续命 threading.Thread(targetheartbeat, daemonTrue).start() return func return decorator # 使用示例 mcp_skill(playwright:form-fill, tags[browser]) def fill_form(url: str, fields: dict): Fill a web form with Playwright # 实际的 Playwright 代码... return {status: success, url: url}这个装饰器做了三件事自动注册元数据、设置过期时间、启动后台心跳。它让技能开发回归本质——写函数而不是管 Redis 命令。3.3 Agent 如何通过 Redis 发现并调用技能一个完整的 Python 流程Agent 的核心逻辑是“查-选-调-监”四步闭环。以下是用 Python 实现的精简版import redis import json import websocket import time r redis.Redis(hostlocalhost, port6379, db0) class MCPAgent: def __init__(self, agent_id: str): self.agent_id agent_id # 订阅技能状态变更 self.pubsub r.pubsub() self.pubsub.subscribe(skills:status:changed) def discover_skills(self, tags: list None) - list: 发现符合条件的技能 # 先查所有技能 all_keys r.keys(skills:*) candidates [] for key in all_keys: skill_data r.hgetall(key) if not skill_data: continue # 解码 bytes decoded {k.decode(): v.decode() for k, v in skill_data.items()} # 检查状态和标签 if decoded.get(status) ! online: continue if tags and not set(tags).issubset(set(decoded.get(tags, ).split(,))): continue candidates.append({ id: key.decode().split(:)[-1], name: decoded.get(name), endpoint: decoded.get(endpoint), score: self._calc_health_score(key.decode()) # 从 Sorted Set 读分 }) # 按健康度排序 return sorted(candidates, keylambda x: x[score], reverseTrue) def _calc_health_score(self, skill_key: str) - float: 从 Sorted Set 读取健康分 score r.zscore(skills:health:score, skill_key.split(:)[-1]) return score or 0.0 def call_skill(self, skill_id: str, input_data: dict) - dict: 调用指定技能 # 从 Redis 获取 endpoint endpoint r.hget(fskills:{skill_id}, endpoint) if not endpoint: raise ValueError(fSkill {skill_id} not found or offline) # 建立 WSS 连接简化版实际用 websocket-client ws websocket.WebSocket() ws.connect(endpoint.decode()) # 发送调用请求 call_msg { type: call, skill_id: skill_id, input: input_data } ws.send(json.dumps(call_msg)) # 等待响应 response json.loads(ws.recv()) ws.close() return response # 使用示例 agent MCPAgent(agent-001) # 查找所有 browser 类技能 skills agent.discover_skills([browser]) if skills: best_skill skills[0] result agent.call_skill(best_skill[id], {url: https://example.com/form, fields: {name: Alice}}) print(result)这段代码展示了 Redis 如何成为 Agent 的“大脑”它不执行任何业务逻辑但提供了发现、筛选、调用的全部上下文。discover_skills方法里r.keys(skills:*)是潜在性能瓶颈大数据量时实际生产环境我们会用SCAN替代并建立skills:tag:browser这样的二级索引 Hash把playwright:form-fill的 ID 存进去实现 O(1) 标签查询。3.4 Docker 环境下的 Redis 主从部署为什么 Agent 架构必须考虑高可用单机 Redis 满足不了生产需求。Agent 系统要求 24/7 在线一旦 Redis 挂掉所有技能发现和任务调度就瘫痪。我们采用经典的 Redis 主从 Sentinel 方案Docker Compose 配置如下version: 3.8 services: redis-master: image: redis:7.2-alpine command: redis-server /usr/local/etc/redis.conf volumes: - ./redis-master.conf:/usr/local/etc/redis.conf ports: - 6379:6379 networks: - mcp-net redis-slave-1: image: redis:7.2-alpine command: redis-server /usr/local/etc/redis.conf volumes: - ./redis-slave.conf:/usr/local/etc/redis.conf depends_on: - redis-master networks: - mcp-net redis-sentinel: image: redis:7.2-alpine command: redis-sentinel /usr/local/etc/sentinel.conf volumes: - ./sentinel.conf:/usr/local/etc/sentinel.conf ports: - 26379:26379 depends_on: - redis-master - redis-slave-1 networks: - mcp-net networks: mcp-net: driver: bridge关键配置文件redis-master.confbind 0.0.0.0 port 6379 protected-mode no daemonize no requirepass your_strong_password # 关键开启 AOF保证数据不丢 appendonly yes appendfilename appendonly.aof # 设置合理的 sync 策略 appendfsync everysecsentinel.confport 26379 sentinel monitor mymaster redis-master 6379 2 sentinel auth-pass mymaster your_strong_password sentinel down-after-milliseconds mymaster 5000 sentinel failover-timeout mymaster 180000 sentinel parallel-syncs mymaster 1实操心得很多团队用 Redis Cluster觉得“分片更高级”。但在 Agent 场景Cluster 的跨 slot 操作如KEYS *被禁止而技能发现常需扫描所有 keys。Sentinel 模式虽无分片但主从切换 30 秒且KEYS命令可用更适合当前阶段。等技能数量超 10K再平滑迁移到 Cluster用redis-py-cluster客户端处理。4. 实操过程详解从零搭建一个 MCP Redis 的 Python Agent 系统4.1 环境准备Python 3.10、Redis 7.2、Playwright可选我们以“构建一个能自动填写网页表单的 AI Agent”为例完整走一遍。所需工具Python 3.10MCP 库依赖较新特性如asyncio的改进Redis 7.2支持STREAM的MAXLEN限制和XADD的NOMKSTREAM选项Playwright作为技能执行引擎也可换成 Selenium、Requests 等安装命令# 创建虚拟环境 python -m venv mcp-env source mcp-env/bin/activate # Linux/Mac # mcp-env\Scripts\activate # Windows # 安装核心库 pip install redis websocket-client playwright python-dotenv # 安装 Playwright 浏览器Chrome playwright install chromium # 启动 RedisDocker 方式最干净 docker run -d --name redis-mcp -p 6379:6379 -e REDIS_PASSWORDyourpass redis:7.2-alpine注意不要用redis-server直接启动因为默认配置不启用密码和 AOF生产环境会丢数据。Docker 启动时通过-e REDIS_PASSWORD设置密码后续 Python 连接要用redis://:yourpasslocalhost:6379/0。4.2 编写第一个 MCP 技能Playwright 表单填充服务创建skills/playwright_form.pyimport asyncio from playwright.async_api import async_playwright import redis import json import os from datetime import datetime # 从环境变量读取 Redis 配置 REDIS_URL os.getenv(REDIS_URL, redis://:yourpasslocalhost:6379/0) r redis.from_url(REDIS_URL) # 技能元数据装饰器会用到 input_schema { type: object, properties: { url: {type: string, format: uri}, fields: { type: object, additionalProperties: {type: string} } }, required: [url, fields] } output_schema { type: object, properties: { status: {type: string}, filled_fields: {type: array, items: {type: string}}, screenshot_path: {type: string} } } # MCP 技能装饰器简化版无心跳 def mcp_skill(skill_id: str, tags: list None): def decorator(func): skill_meta { name: func.__name__, description: func.__doc__ or , input_schema: input_schema, output_schema: output_schema, endpoint: ws://localhost:8000/mcp, # 本机 WebSocket 服务 status: online, last_heartbeat: int(datetime.now().timestamp()), tags: tags or [browser, playwright] } r.hset(fskills:{skill_id}, mappingskill_meta) r.expire(fskills:{skill_id}, 300) return func return decorator mcp_skill(playwright:form-fill, tags[browser]) async def fill_form(url: str, fields: dict): Fill a web form using Playwright async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) page await browser.new_page() try: await page.goto(url, timeout30000) # 填写字段 filled [] for selector, value in fields.items(): await page.fill(selector, value) filled.append(selector) # 截图 screenshot_path f/tmp/{int(datetime.now().timestamp())}.png await page.screenshot(pathscreenshot_path) await browser.close() return { status: success, filled_fields: filled, screenshot_path: screenshot_path } except Exception as e: await browser.close() return { status: error, message: str(e) } # WebSocket 服务简化仅演示 import websockets import asyncio async def mcp_handler(websocket, path): async for message in websocket: data json.loads(message) if data.get(type) call and data.get(skill_id) playwright:form-fill: # 解析 input input_data data.get(input, {}) # 调用技能 result await fill_form(input_data.get(url), input_data.get(fields, {})) # 返回结果 await websocket.send(json.dumps({ type: result, success: result[status] success, output: result })) # 启动 WebSocket 服务 async def main(): async with websockets.serve(mcp_handler, localhost, 8000): await asyncio.Future() # run forever if __name__ __main__: asyncio.run(main())这个脚本做了三件事定义技能元数据、用装饰器注册到 Redis、实现 WebSocket 服务接收 MCP 调用。注意fill_form是异步函数因为 Playwright 必须异步调用。4.3 构建 Agent 核心技能发现、路由与容错创建agent/core.pyimport redis import json import asyncio import websockets import os from typing import Dict, Any, List REDIS_URL os.getenv(REDIS_URL, redis://:yourpasslocalhost:6379/0) r redis.from_url(REDIS_URL) class SkillRouter: 技能路由中心负责发现、筛选、调用 def __init__(self): # 缓存最近一次发现结果避免频繁 SCAN self._skill_cache {} self._cache_time 0 def _scan_skills(self, pattern: str skills:*) - List[Dict]: 扫描 Redis 获取技能列表 keys r.scan(0, pattern, 1000)[1] # 分页扫描 skills [] for key in keys: data r.hgetall(key) if not data: continue # 解码 decoded {k.decode(): v.decode() for k, v in data.items()} if decoded.get(status) online: skills.append({ id: key.decode().split(:)[-1], name: decoded.get(name), endpoint: decoded.get(endpoint), tags: decoded.get(tags, ).split(,), health_score: self._get_health_score(key.decode().split(:)[-1]) }) return skills def _get_health_score(self, skill_id: str) - float: 从 Sorted Set 获取健康分 score r.zscore(skills:health:score, skill_id) return score or 50.0 # 默认分 def discover(self, tags: List[str] None) - List[Dict]: 发现技能带缓存 now time.time() if now - self._cache_time 10: # 10秒缓存 return self._skill_cache skills self._scan_skills() if tags: skills [s for s in skills if set(tags).issubset(set(s[tags]))] # 按健康分排序 skills.sort(keylambda x: x[health_score], reverseTrue) self._skill_cache skills self._cache_time now return skills async def call(self, skill_id: str, input_data: Dict) - Dict: 调用技能带重试和熔断 # 从 Redis 获取 endpoint endpoint r.hget(fskills:{skill_id}, endpoint) if not endpoint: raise ValueError(fSkill {skill_id} not found) # 熔断检查如果健康分 70拒绝调用 score self._get_health_score(skill_id) if score 70: raise RuntimeError(fSkill {skill_id} health score too low: {score}) # 重试 3 次 for attempt in range(3): try: async with websockets.connect(endpoint.decode()) as ws: await ws.send(json.dumps({ type: call, skill_id: skill_id, input: input_data })) response json.loads(await ws.recv()) return response except Exception as e: if attempt 2: # 最后一次失败记录日志并更新健康分 r.zincrby(skills:health:score, -10, skill_id) raise e await asyncio.sleep(0.5 * (2 ** attempt)) # 指数退避 return {} # 使用示例 router SkillRouter() # Agent 的主逻辑模拟 LLM 的决策 async def agent_workflow(): # LLM 决定需要填写表单 task { action: fill_form, url: https://example.com/contact, fields: {#name: Alice, #email: aliceexample.com} } # 发现技能 skills router.discover([browser]) if not skills: print(No available skills!) return # 选最优技能 best_skill skills[0] print(fUsing skill: {best_skill[name]} (score: {best_skill[health_score]})) # 调用 try: result await router.call(best_skill[id], task) print(Result:, result) except Exception as e: print(Call failed:, e) # 运行 asyncio.run(agent_workflow())这个SkillRouter类体现了 Agent 的核心智能它不只是调用还做缓存、熔断、重试、健康度反馈。每次调用失败它会主动给技能扣分zincrby让低分技能逐渐退出候选池。这就是 Redis 的力量——状态驱动的自适应调度。4.4 监控与治理用 Redis CLI 和 Python 脚本做日常运维系统上线后不能只靠代码跑。你需要一套监控手段实时查看技能状态# 进入 Redis CLI redis-cli -a yourpass # 查看所有技能 KEYS skills:* # 查看某个技能详情 HGETALL skills:playwright:form-fill # 查看健康分排行榜 ZRANGE skills:health:score 0 -1 WITHSCORESPython 监控脚本monitor.pyimport redis import time from datetime import datetime r redis.from_url(redis://:yourpasslocalhost:6379/0) def check_skill_health(): 检查所有技能的心跳是否正常 keys r.keys(skills:*) now int(time.time()) offline_count 0 for key in keys: last_hb r.hget(key, last_heartbeat) if last_hb and int(last_hb) now - 60: # 超过60秒未心跳 r.hset(key, status, offline) offline_count 1 print(f[ALERT] Skill {key.decode()} offline!) print(fTotal offline skills: {offline_count}) def report_stats(): 生成日报 total_skills len(r.keys(skills:*)) online_skills len([k for k in r.keys(skills:*) if r.hget(k, status) bonline]) avg_health r.zrange(skills:health:score, 0, -1, withscoresTrue) avg_score sum(s for _, s in avg_health) / len(avg_health) if avg_health else 0 report f MCP Agent Daily Report ({datetime.now().strftime(%Y-%m-%d)}) Total Skills: {total_skills} Online Skills: {online_skills} Avg Health Score: {avg_score:.1f} print(report) if __name__ __main__: while True: check_skill_health() report_stats() time.sleep(300) # 每5分钟检查一次这个脚本会自动踢掉失联技能并生成健康报告。你可以用systemd或supervisord守护它让它永远运行。5. 常见问题与排查技巧实录我在 3 个项目中踩过的坑5.1 问题速查表高频故障与根因分析现象可能原因排查命令解决方案Agent 总是找不到技能Redis 密码错误连接被拒redis-cli -a wrongpass ping检查REDIS_URL环境变量确认密码正确技能状态显示offline但服务明明在跑心跳线程未启动或异常退出HGET skills:xxx last_heartbeat在装饰器中加try/except日志确保心跳线程存活SCAN命令返回空但KEYS *能查到Redis 配置了protected-mode yesredis-cli config get protected-mode设为no或用--no-auth-warning启动WebSocket 调用超时但技能服务日志显示已收到Redis Pub/Sub 通道名拼写错误PUBSUB CHANNELS确认PUBLISH和SUBSCRIBE使用完全相同的 channel 名健康分一直为 0ZSCORE返回 NoneSorted Set 未初始化或ZADD未执行ZCARD skills:health:score在技能注册时强制ZADD skills:health:score 100 skill_id5.2 独家避坑技巧那些文档里不会写的细节技巧 1用SCAN代替KEYS但要注意游标陷阱KEYS