cache manage

tangwang
1 parent 6f7840cf
Showing 5 changed files with 755 additions and 6 deletions Show diff stats
docs/DEVELOPER_GUIDE.md
docs/缓存与Redis使用说明.md
scripts/redis/redis_cache_health_check.py
scripts/redis/check_cache_stats.py -> scripts/redis/redis_cache_prefix_stats.py
scripts/redis/find_memory_usage.py -> scripts/redis/redis_memory_heavy_keys.py
@@ -423,6 +423,7 @@ services:
 | 向量模块与 clip-as-service | [embeddings/README.md](../embeddings/README.md) |
 | TEI 服务专项（安装/部署/GPU-CPU 模式） | [TEI_SERVICE说明文档.md](./TEI_SERVICE说明文档.md) |
 | CN-CLIP 服务专项（环境/运维/GPU） | [CNCLIP_SERVICE说明文档.md](./CNCLIP_SERVICE说明文档.md) |
+| 缓存 / Redis 使用与 key 设计 | [缓存与Redis使用说明.md](./缓存与Redis使用说明.md) |
  
 ### 10.2 仓库内入口
  
@@ -0,0 +1,347 @@
+## 缓存与 Redis 使用说明
+
+本仓库中 Redis 主要用于**性能型缓存**，当前落地的业务缓存包括：
+
+- **文本向量缓存**（embedding 缓存）
+- **翻译结果缓存**（Qwen-MT 等机器翻译）
+- **商品内容理解缓存**（锚文本 / 语义属性 / 标签）
+
+底层连接配置统一来自 `config/env_config.py` 的 `REDIS_CONFIG`：
+
+- **Host/Port**：`REDIS_HOST` / `REDIS_PORT`（默认 `localhost:6479`）
+- **Password**：`REDIS_PASSWORD`
+- **Socket & 超时**：`REDIS_SOCKET_TIMEOUT` / `REDIS_SOCKET_CONNECT_TIMEOUT` / `REDIS_RETRY_ON_TIMEOUT`
+- **通用缓存 TTL**：`REDIS_CACHE_EXPIRE_DAYS`（默认 `360*2` 天，代码注释为 “6 months”）
+- **翻译缓存 TTL & 前缀**：`REDIS_TRANSLATION_CACHE_EXPIRE_DAYS`、`REDIS_TRANSLATION_CACHE_PREFIX`
+
+---
+
+## 1. 缓存总览表
+
+| 模块 / 场景 | Key 模板 | Value 内容示例 | 过期策略 | 备注 |
+|------------|----------|----------------|----------|------|
+| 文本向量缓存（embedding） | `embedding:{language}:{norm_flag}:{query}` | `pickle.dumps(np.ndarray)`，如 1024 维 BGE 向量 | TTL=`REDIS_CONFIG["cache_expire_days"]` 天；访问时滑动过期 | 见 `embeddings/text_encoder.py` |
+| 翻译结果缓存（Qwen-MT 翻译） | `{cache_prefix}:{model}:{src}:{tgt}:{sha256(payload)}` | 机翻后的单条字符串 | TTL=`services.translation.cache.ttl_seconds` 秒；可配置滑动过期 | 见 `query/qwen_mt_translate.py` + `config/config.yaml` |
+| 商品内容理解缓存（anchors / 语义属性 / tags） | `{ANCHOR_CACHE_PREFIX}:{tenant_or_global}:{target_lang}:{md5(title)}` | `json.dumps(dict)`，包含 id/title/category/tags/anchor_text 等 | TTL=`ANCHOR_CACHE_EXPIRE_DAYS` 天 | 见 `indexer/product_enrich.py` |
+
+下面按模块详细说明。
+
+---
+
+## 2. 文本向量缓存（embeddings/text_encoder.py）
+
+- **代码位置**：`embeddings/text_encoder.py` 中 `TextEmbeddingEncoder`
+- **用途**：缓存调用向量服务（6005）的文本向量结果，避免重复计算。
+
+### 2.1 Key 设计
+
+- 函数：`_get_cache_key(query: str, language: str, normalize_embeddings: bool) -> str`
+- 模板：
+
+```text
+embedding:{language}:{norm_flag}:{query}
+```
+
+- 字段说明：
+  - `language`：当前实现中统一传入 `"generic"`；
+  - `norm_flag`：`"norm1"` 表示归一化向量，`"norm0"` 表示未归一化；
+  - `query`：原始文本（未做哈希），注意长度特别长的 query 会直接出现在 key 中。
+
+### 2.2 Value 与类型
+
+- 类型：`pickle.dumps(np.ndarray)`，在读取时通过 `pickle.loads` 还原为 `np.ndarray`。
+- 典型示例：BGE-M3 1024 维 `float32` 向量。
+
+### 2.3 过期策略
+
+- 初始化：
+  - `self.expire_time = timedelta(days=REDIS_CONFIG.get("cache_expire_days", 180))`
+  - `.env` 中可通过 `REDIS_CACHE_EXPIRE_DAYS` 配置，默认 `360*2`（代码注释标注为 6 个月）。
+- 写入：
+  - `redis.setex(cache_key, self.expire_time, serialized_data)`
+- 访问（滑动过期）：
+  - 命中缓存后，会调用 `redis.expire(cache_key, self.expire_time)` 延长 TTL。
+
+### 2.4 特殊处理
+
+- 若缓存中的向量 **为空 / shape 异常 / 含 NaN/Inf**，会：
+  - 直接丢弃该缓存（并尝试 `delete` key）；
+  - 回退为重新调用向量服务。
+
+---
+
+## 3. 翻译结果缓存（query/qwen_mt_translate.py）
+
+- **代码位置**：`query/qwen_mt_translate.py` 中 `Translator` 类
+- **用途**：缓存 Qwen-MT 翻译（及 translator service 复用的翻译）结果，减少云端请求，遵守限速。
+- **配置入口**：`config/config.yaml -> services.translation.cache`，统一由 `config/services_config.get_translation_cache_config()` 解析。
+
+### 3.1 Key 设计
+
+- 内部构造函数：`_build_cache_key(...)`
+- 模板：
+
+```text
+{cache_prefix}:{model}:{src}:{tgt}:{sha256(payload)}
+```
+
+其中：
+
+- `cache_prefix`：来自 `services.translation.cache.key_prefix`，默认 `trans:v2`；
+- `model`：如 `"qwen-mt"`；
+- `src`：源语言（如 `zh` / `en` / `auto`），是否包含在 key 中由 `key_include_source_lang` 控制；
+- `tgt`：目标语言，如 `en` / `zh`；
+- `sha256(payload)`：对以下内容整体做 SHA-256：
+  - `model`
+  - `src` / `tgt`
+  - `context`（受 `key_include_context` 控制）
+  - `prompt`（受 `key_include_prompt` 控制）
+  - 原始 `text`
+
+> 注意：所有 key 设计集中在 `_build_cache_key`，**不要在其他位置手动拼翻译缓存 key**。
+
+### 3.2 Value 与类型
+
+- 类型：**UTF-8 字符串**，即翻译后的文本结果。
+- 存取逻辑：
+  - 读取：`redis.get(key)` 返回 `str` 或 `None`；
+  - 写入：`redis.setex(key, expire_seconds, translation)`。
+
+### 3.3 过期策略
+
+- 配置来源：`config/config.yaml -> services.translation.cache`，经 `get_translation_cache_config()` 解析：
+
+```yaml
+services:
+  translation:
+    cache:
+      enabled: true
+      key_prefix: "trans:v2"
+      ttl_seconds: 62208000        # 默认约 720 天
+      sliding_expiration: true
+      key_include_context: true
+      key_include_prompt: true
+      key_include_source_lang: true
+```
+
+- 运行时行为：
+  - 创建 `Translator` 时，从 `cache_cfg` 读取：
+    - `self.cache_prefix`
+    - `self.expire_seconds`
+    - `self.cache_sliding_expiration`
+    - `self.cache_include_*` 一系列布尔开关；
+  - **读缓存**：
+    - 命中后，若 `sliding_expiration=True`，会调用 `redis.expire(key, expire_seconds)`；
+  - **写缓存**：
+    - 使用 `redis.setex(key, expire_seconds, translation)`。
+
+### 3.4 关联模块
+
+- `api/translator_app.py` 会通过 `query.qwen_mt_translate.Translator` 复用同一套缓存逻辑；
+- 文档说明：`docs/翻译模块说明.md` 中提到“推荐通过 Redis 翻译缓存复用结果”。
+
+---
+
+## 4. 商品内容理解缓存（indexer/product_enrich.py）
+
+- **代码位置**：`indexer/product_enrich.py`
+- **用途**：在生成商品锚文本（qanchors）、语义属性、标签等内容理解结果时复用缓存，避免对同一标题重复调用大模型。
+
+### 4.1 Key 设计
+
+- 配置项：
+  - `ANCHOR_CACHE_PREFIX = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")`
+  - `ANCHOR_CACHE_EXPIRE_DAYS = int(REDIS_CONFIG.get("anchor_cache_expire_days", 30))`
+- Key 构造函数：`_make_anchor_cache_key(title, target_lang, tenant_id)`
+- 模板：
+
+```text
+{ANCHOR_CACHE_PREFIX}:{tenant_or_global}:{target_lang}:{md5(title)}
+```
+
+- 字段说明：
+  - `ANCHOR_CACHE_PREFIX`：默认 `"product_anchors"`，可通过 `.env` 中的 `REDIS_ANCHOR_CACHE_PREFIX`（若存在）间接配置到 `REDIS_CONFIG`；
+  - `tenant_or_global`：`tenant_id` 去空白后的字符串，若为空则使用 `"global"`；
+  - `target_lang`：内容理解输出语言，例如 `zh`；
+  - `md5(title)`：对原始商品标题（UTF-8）做 MD5。
+
+### 4.2 Value 与类型
+
+- 类型：`json.dumps(dict, ensure_ascii=False)`。
+- 典型结构（简化）：
+
+```json
+{
+  "id": "123",
+  "lang": "zh",
+  "title_input": "原始标题",
+  "title": "归一化后的商品标题",
+  "category_path": "...",
+  "tags": "...",
+  "target_audience": "...",
+  "usage_scene": "...",
+  "anchor_text": "..., ..."
+}
+```
+
+- 读取时通过 `json.loads(raw)` 还原为 `Dict[str, Any]`。
+
+### 4.3 过期策略
+
+- TTL：`ttl = ANCHOR_CACHE_EXPIRE_DAYS * 24 * 3600` 秒（默认 30 天）；
+- 写入：`redis.setex(key, ttl, json.dumps(result, ensure_ascii=False))`；
+- 读取：仅做 `redis.get(key)`，**不做滑动过期**。
+
+### 4.4 调用流程中的位置
+
+- 单条调用（索引阶段常见）时，`analyze_products()` 会先尝试命中缓存：
+  - 若命中，直接返回缓存结果；
+  - 若 miss，调用 LLM，解析结果后再写入缓存。
+
+---
+
+## 5. Redis 运维脚本工具
+
+`scripts/redis/` 下提供三个脚本，用于查看缓存数量、内存占用与健康状态。连接配置均来自 `config/env_config.py` 的 `REDIS_CONFIG`，运行前需在项目根目录执行（或保证 `PYTHONPATH` 含项目根），以便加载配置。
+
+### 5.1 redis_cache_health_check.py（缓存健康巡检）
+
+**功能**：按**业务缓存类型**（embedding / translation / anchors）做健康巡检，不扫全库。
+
+- 对每类缓存：SCAN 匹配对应 key 前缀，统计**匹配 key 数量**（受 `--max-scan` 上限约束）；
+- **TTL 分布**：对采样 key 统计 `no-expire-or-expired` / `0-1h` / `1h-1d` / `1d-30d` / `>30d`；
+- **近期活跃 key**：从采样中选出 `OBJECT IDLETIME <= 600s` 的 key，用于判断是否有新写入；
+- **样本 key 与 value 预览**：对 embedding 显示 ndarray 信息，对 translation 显示译文片段，对 anchors 显示 JSON 摘要。
+
+**适用场景**：日常查看三类缓存是否在增长、TTL 是否合理、是否有近期写入；与「缓存总览表」中的 key 设计一一对应。
+
+**用法示例**：
+
+```bash
+# 默认：检查 embedding / translation / anchors 三类
+python scripts/redis/redis_cache_health_check.py
+
+# 只检查某一类或两类
+python scripts/redis/redis_cache_health_check.py --type embedding
+python scripts/redis/redis_cache_health_check.py --type translation anchors
+
+# 按自定义 pattern 检查（不按业务类型）
+python scripts/redis/redis_cache_health_check.py --pattern "mycache:*"
+
+# 调整采样与扫描规模
+python scripts/redis/redis_cache_health_check.py --sample-size 100 --max-scan 50000 --db 0
+```
+
+**常用参数**：
+
+| 参数 | 说明 | 默认 |
+|------|------|------|
+| `--type` | 缓存类型：`embedding` / `translation` / `anchors`，可多选 | 三类都检查 |
+| `--pattern` | 自定义 key pattern（如 `mycache:*`），指定后忽略 `--type` | - |
+| `--db` | Redis 数据库编号 | 0 |
+| `--sample-size` | 每类采样的 key 数量 | 50 |
+| `--max-scan` | 每类最多 SCAN 的 key 数量上限 | 20000 |
+
+---
+
+### 5.2 redis_cache_prefix_stats.py（按前缀统计条数与内存）
+
+**功能**：**全局视角**，扫描当前 DB 下所有 key，按 key 的**前缀**（第一个冒号前）分类，统计每类 key 的**条数**与**内存占用量**（含占比），并输出每类示例 key 与 Redis 总内存信息。
+
+- 使用 `SCAN` 扫全库，按前缀聚合；
+- 内存优先用 Redis `MEMORY USAGE`，不可用时用 key+value 长度估算；key 过多时按 `--sample-size` 采样后按均值推算总内存；
+- 输出：前缀、条数、内存及计算方式、占比、简要说明（如「翻译缓存」「向量化缓存」）；末尾附 Redis 总内存与 maxmemory（若配置）。
+
+**适用场景**：了解「哪一类前缀占了多少条、多少内存」，做容量规划或清理决策；支持多 DB（`--all-db`）或指定前缀（`--prefix`）缩小范围。
+
+**用法示例**：
+
+```bash
+# 默认 DB 0，全库按前缀统计
+python scripts/redis/redis_cache_prefix_stats.py
+
+# 统计所有有数据的 DB
+python scripts/redis/redis_cache_prefix_stats.py --all-db
+
+# 指定 DB
+python scripts/redis/redis_cache_prefix_stats.py --db 1
+
+# 只统计指定前缀（可多个）
+python scripts/redis/redis_cache_prefix_stats.py --prefix trans embedding product_anchors
+
+# 全 DB + 指定前缀
+python scripts/redis/redis_cache_prefix_stats.py --all-db --prefix trans embedding
+```
+
+**常用参数**：
+
+| 参数 | 说明 | 默认 |
+|------|------|------|
+| `--prefix` | 只统计这些前缀下的 key（如 `trans` `embedding`） | 全库按前缀统计 |
+| `--db` | 数据库编号（0–15） | 0 |
+| `--all-db` | 对所有有数据的 DB 分别执行 | 否 |
+| `--sample-size` | 单前缀 key 过多时，用于内存采样的数量 | 100 |
+| `--real` | 对单前缀 key 数 ≤10000 时计算全部 key 真实内存（较慢） | 否 |
+
+---
+
+### 5.3 redis_memory_heavy_keys.py（大 key / 内存占用排查）
+
+**功能**：找出当前 DB 中**占用内存最多的 key**，并分析「按 key 估算的总内存」与「Redis 实际使用内存」的差异原因。
+
+- 全库 `SCAN` 获取 key 列表，对 key 采样（默认最多 1000）调用 `MEMORY USAGE`（或估算）得到单 key 内存；
+- 按内存排序，输出**占用最高的 N 个 key**（`--top`，默认 50）；
+- 输出：前缀分布、采样统计、估算总内存 vs 实际内存、差异说明（碎片、内部结构等）；并检测超大 value（>1MB）、key 类型分布。
+
+**适用场景**：内存异常升高时定位大 key；理解为何「按 key 加总」与 `used_memory` 不一致。
+
+**用法示例**：
+
+```bash
+# 显示占用内存最多的 50 个 key（默认）
+python scripts/redis/redis_memory_heavy_keys.py
+
+# 显示前 100 个
+python scripts/redis/redis_memory_heavy_keys.py --top 100
+```
+
+**常用参数**：
+
+| 参数 | 说明 | 默认 |
+|------|------|------|
+| `--top` | 显示内存占用最高的 N 个 key | 50 |
+
+---
+
+### 5.4 三个脚本的选用建议
+
+| 需求 | 推荐脚本 |
+|------|----------|
+| 看三类业务缓存（embedding/translation/anchors）的数量、TTL、近期写入、样本 value | `redis_cache_health_check.py` |
+| 看全库或某前缀的 key 条数与内存占比 | `redis_cache_prefix_stats.py` |
+| 找占用内存最多的大 key、分析内存差异 | `redis_memory_heavy_keys.py` |
+
+---
+
+## 6. 其他 Redis 相关代码（测试与替身）
+
+除上述运维脚本外，以下代码使用 Redis 或其替身，但不定义新的业务缓存 key 协议：
+
+- **单元测试**：`tests/test_embedding_pipeline.py` 中的 `_FakeRedis` 用于 mock embedding 缓存；`tests/test_translator_failure_semantics.py` 中的 `_RecordingRedis` 用于验证翻译失败时不写缓存。
+
+---
+
+## 7. 添加新缓存时的建议
+
+新增 Redis 缓存时，建议遵循以下约定：
+
+- **配置集中**：
+  - key 前缀、TTL 优先放在 `config/env_config.py`（通用）或 `config/config.yaml -> services.<capability>.cache`；
+  - 避免在业务代码中硬编码 TTL。
+- **命名规范**：
+  - 统一使用 `prefix:维度1:维度2:...` 的扁平 key 结构；
+  - 对长文本/value 使用 `md5`/`sha256` 做哈希，避免过长 key。
+- **文档同步**：
+  - 新增缓存后，应在本文件中补充一行总览表 + 详细小节；
+  - 若缓存与外部系统/历史实现兼容（如 Java 侧翻译缓存），需在说明中显式标注。
+
@@ -0,0 +1,401 @@
+#!/usr/bin/env python3
+"""
+缓存状态巡检脚本
+
+按「缓存类型」维度（embedding / translation / anchors）查看：
+- 估算 key 数量
+- TTL 分布（采样）
+- 近期活跃 key（按 IDLETIME 近似）
+- 若干样本 key 及 value 概览
+
+使用示例：
+
+  # 默认：检查已知三类缓存，使用 env_config 中的 Redis 配置
+  python scripts/redis/redis_cache_health_check.py
+
+  # 只看某一类缓存
+  python scripts/redis/redis_cache_health_check.py --type embedding
+  python scripts/redis/redis_cache_health_check.py --type translation anchors
+
+  # 自定义前缀（pattern），不限定缓存类型
+  python scripts/redis/redis_cache_health_check.py --pattern "mycache:*"
+
+  # 调整采样规模
+  python scripts/redis/redis_cache_health_check.py --sample-size 100 --max-scan 50000
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import pickle
+import sys
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, Iterable, List, Optional, Tuple
+
+import redis
+import numpy as np
+
+# 让脚本可以直接使用 config/env_config 与 services_config
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from config.env_config import REDIS_CONFIG  # type: ignore
+from config.services_config import get_translation_cache_config  # type: ignore
+
+
+@dataclass
+class CacheTypeConfig:
+    name: str
+    pattern: str
+    description: str
+
+
+def _load_known_cache_types() -> Dict[str, CacheTypeConfig]:
+    """根据当前配置装配三种已知缓存类型及其前缀 pattern。"""
+    cache_types: Dict[str, CacheTypeConfig] = {}
+
+    # embedding 缓存：固定 embedding:* 前缀
+    cache_types["embedding"] = CacheTypeConfig(
+        name="embedding",
+        pattern="embedding:*",
+        description="文本向量缓存（embeddings/text_encoder.py）",
+    )
+
+    # translation 缓存：prefix 来自 services.translation.cache.key_prefix
+    cache_cfg = get_translation_cache_config()
+    trans_prefix = cache_cfg.get("key_prefix", "trans:v2")
+    cache_types["translation"] = CacheTypeConfig(
+        name="translation",
+        pattern=f"{trans_prefix}:*",
+        description="翻译结果缓存（query/qwen_mt_translate.Translator）",
+    )
+
+    # anchors 缓存：prefix 来自 REDIS_CONFIG['anchor_cache_prefix']（若存在），否则 product_anchors
+    anchor_prefix = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")
+    cache_types["anchors"] = CacheTypeConfig(
+        name="anchors",
+        pattern=f"{anchor_prefix}:*",
+        description="商品内容理解缓存（indexer/product_enrich.py，anchors/语义属性/tags）",
+    )
+
+    return cache_types
+
+
+def get_redis_client(db: int = 0) -> redis.Redis:
+    return redis.Redis(
+        host=REDIS_CONFIG.get("host", "localhost"),
+        port=REDIS_CONFIG.get("port", 6479),
+        password=REDIS_CONFIG.get("password"),
+        db=db,
+        decode_responses=False,  # 原始 bytes，方便区分 pickle / str
+        socket_timeout=10,
+        socket_connect_timeout=10,
+    )
+
+
+def scan_keys(
+    client: redis.Redis, pattern: str, max_scan: int, scan_count: int = 1000
+) -> List[bytes]:
+    """使用 SCAN 扫描匹配 pattern 的 key，最多扫描 max_scan 个结果。"""
+    keys: List[bytes] = []
+    cursor: int = 0
+    scanned = 0
+    while True:
+        cursor, batch = client.scan(cursor=cursor, match=pattern, count=scan_count)
+        for k in batch:
+            keys.append(k)
+            scanned += 1
+            if scanned >= max_scan:
+                return keys
+        if cursor == 0:
+            break
+    return keys
+
+
+def ttl_bucket(ttl: int) -> str:
+    """将 TTL（秒）归类到简短区间标签中。"""
+    if ttl < 0:
+        # -1: 永不过期；-2: 不存在
+        return "no-expire-or-expired"
+    if ttl <= 3600:
+        return "0-1h"
+    if ttl <= 86400:
+        return "1h-1d"
+    if ttl <= 30 * 86400:
+        return "1d-30d"
+    return ">30d"
+
+
+def format_seconds(sec: int) -> str:
+    if sec < 0:
+        return str(sec)
+    if sec < 60:
+        return f"{sec}s"
+    if sec < 3600:
+        return f"{sec // 60}m{sec % 60}s"
+    if sec < 86400:
+        h = sec // 3600
+        m = (sec % 3600) // 60
+        return f"{h}h{m}m"
+    d = sec // 86400
+    h = (sec % 86400) // 3600
+    return f"{d}d{h}h"
+
+
+def decode_value_preview(
+    cache_type: str, key: bytes, raw_value: Optional[bytes]
+) -> str:
+    """根据缓存类型生成简短的 value 概览字符串。"""
+    if raw_value is None:
+        return "<nil>"
+
+    # embedding: pickle 序列化的 numpy.ndarray
+    if cache_type == "embedding":
+        try:
+            arr = pickle.loads(raw_value)
+            if isinstance(arr, np.ndarray):
+                return f"ndarray shape={arr.shape} dtype={arr.dtype}"
+            return f"pickle object type={type(arr).__name__}"
+        except Exception:
+            return f"<binary {len(raw_value)} bytes>"
+
+    # anchors: JSON dict
+    if cache_type == "anchors":
+        try:
+            text = raw_value.decode("utf-8", errors="replace")
+            obj = json.loads(text)
+            if isinstance(obj, dict):
+                brief = {
+                    k: obj.get(k)
+                    for k in ["id", "lang", "title_input", "title", "category_path", "anchor_text"]
+                    if k in obj
+                }
+                return "json " + json.dumps(brief, ensure_ascii=False)[:200]
+            # 其他情况简单截断
+            return "json " + text[:200]
+        except Exception:
+            return raw_value.decode("utf-8", errors="replace")[:200]
+
+    # translation: 纯字符串
+    if cache_type == "translation":
+        try:
+            text = raw_value.decode("utf-8", errors="replace")
+            return text[:200]
+        except Exception:
+            return f"<binary {len(raw_value)} bytes>"
+
+    # 兜底：尝试解码为 UTF-8
+    try:
+        text = raw_value.decode("utf-8", errors="replace")
+        return text[:200]
+    except Exception:
+        return f"<binary {len(raw_value)} bytes>"
+
+
+def analyze_cache_type(
+    client: redis.Redis,
+    cache_type: str,
+    cfg: CacheTypeConfig,
+    sample_size: int,
+    max_scan: int,
+) -> None:
+    """对单个缓存类型做统计与样本展示。"""
+    print("=" * 80)
+    print(f"Cache type: {cache_type}  pattern={cfg.pattern}")
+    print(f"Description: {cfg.description}")
+    print("=" * 80)
+
+    keys = scan_keys(client, cfg.pattern, max_scan=max_scan)
+    total_scanned = len(keys)
+    print(f"Scanned up to {max_scan} keys, matched {total_scanned} keys.")
+    if total_scanned == 0:
+        print("No keys found for this cache type.\n")
+        return
+
+    # 采样
+    if total_scanned <= sample_size:
+        sampled_keys = keys
+    else:
+        # 简单取前 sample_size 个，scan 本身已是渐进遍历，足够近似随机
+        sampled_keys = keys[:sample_size]
+
+    ttl_hist: Dict[str, int] = defaultdict(int)
+    recent_keys: List[Tuple[bytes, int, int]] = []  # (key, ttl, idletime)
+    samples: List[Tuple[bytes, int, int, str]] = []  # (key, ttl, idletime, preview)
+
+    for k in sampled_keys:
+        try:
+            ttl = client.ttl(k)
+        except Exception:
+            ttl = -3  # 表示 TTL 查询失败
+
+        # TTL 分布
+        ttl_hist[ttl_bucket(ttl)] += 1
+
+        # 近期活跃判断（idletime 越小越“新”）
+        idletime = -1
+        try:
+            # OBJECT IDLETIME 返回秒数（整数）
+            idletime = client.object("idletime", k)  # type: ignore[arg-type]
+        except Exception:
+            pass
+
+        # 记录近期活跃样本
+        if idletime >= 0 and idletime <= 600:
+            recent_keys.append((k, ttl, idletime))
+
+        # 收集样本 value 预览（控制数量）
+        if len(samples) < 5:
+            raw_val = None
+            try:
+                raw_val = client.get(k)
+            except Exception:
+                pass
+            preview = decode_value_preview(cache_type, k, raw_val)
+            samples.append((k, ttl, idletime, preview))
+
+    # TTL 分布输出
+    print("\nTTL distribution (sampled):")
+    total_sampled = len(sampled_keys)
+    for bucket in ["no-expire-or-expired", "0-1h", "1h-1d", "1d-30d", ">30d"]:
+        cnt = ttl_hist.get(bucket, 0)
+        pct = (cnt / total_sampled * 100.0) if total_sampled else 0.0
+        print(f"  {bucket:<18}: {cnt:>4} ({pct:>5.1f}%)")
+
+    # 近期活跃 key
+    recent_keys = sorted(recent_keys, key=lambda x: x[2])[:5]
+    print("\nRecent active keys (idletime <= 600s, from sampled set):")
+    if not recent_keys:
+        print("  (none in sampled set)")
+    else:
+        for k, ttl, idle in recent_keys:
+            try:
+                k_str = k.decode("utf-8", errors="replace")
+            except Exception:
+                k_str = repr(k)
+            if len(k_str) > 80:
+                k_str = k_str[:77] + "..."
+            print(
+                f"  key={k_str}  ttl={ttl} ({format_seconds(ttl)})  "
+                f"idletime={idle} ({format_seconds(idle)})"
+            )
+
+    # 样本 value 概览
+    print("\nSample keys & value preview:")
+    if not samples:
+        print("  (no samples)")
+    else:
+        for k, ttl, idle, preview in samples:
+            try:
+                k_str = k.decode("utf-8", errors="replace")
+            except Exception:
+                k_str = repr(k)
+            if len(k_str) > 80:
+                k_str = k_str[:77] + "..."
+            print(f"  key={k_str}")
+            print(f"    ttl={ttl} ({format_seconds(ttl)})  idletime={idle} ({format_seconds(idle)})")
+            print(f"    value: {preview}")
+
+    print()  # 结尾空行
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Redis 缓存状态巡检（按缓存类型）")
+    parser.add_argument(
+        "--type",
+        dest="types",
+        nargs="+",
+        choices=["embedding", "translation", "anchors"],
+        help="指定要检查的缓存类型（默认：三种全部）",
+    )
+    parser.add_argument(
+        "--pattern",
+        type=str,
+        help="自定义 key pattern（如 'mycache:*'）。设置后将忽略 --type，仅按 pattern 进行一次检查。",
+    )
+    parser.add_argument(
+        "--db",
+        type=int,
+        default=0,
+        help="Redis 数据库编号（默认 0）",
+    )
+    parser.add_argument(
+        "--sample-size",
+        type=int,
+        default=50,
+        help="每种缓存类型采样的 key 数量（默认 50）",
+    )
+    parser.add_argument(
+        "--max-scan",
+        type=int,
+        default=20000,
+        help="每种缓存类型最多 SCAN 的 key 数量上限（默认 20000）",
+    )
+
+    args = parser.parse_args()
+
+    print("Redis 缓存状态巡检")
+    print("=" * 80)
+    print(f"Checked at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    print(f"Redis host={REDIS_CONFIG.get('host', 'localhost')} port={REDIS_CONFIG.get('port', 6479)} db={args.db}")
+    print()
+
+    try:
+        client = get_redis_client(db=args.db)
+        client.ping()
+        print("✅ Redis 连接成功\n")
+    except Exception as exc:
+        print(f"❌ Redis 连接失败: {exc}")
+        print(f"  Host: {REDIS_CONFIG.get('host', 'localhost')}")
+        print(f"  Port: {REDIS_CONFIG.get('port', 6479)}")
+        print(f"  Password: {'已配置' if REDIS_CONFIG.get('password') else '未配置'}")
+        return
+
+    # 如果指定了自定义 pattern，则只做一次“匿名类型”的巡检
+    if args.pattern:
+        anon_cfg = CacheTypeConfig(
+            name=f"pattern:{args.pattern}",
+            pattern=args.pattern,
+            description="自定义 pattern 检查",
+        )
+        analyze_cache_type(
+            client=client,
+            cache_type="custom",
+            cfg=anon_cfg,
+            sample_size=args.sample_size,
+            max_scan=args.max_scan,
+        )
+        print("巡检完成。")
+        return
+
+    # 否则根据已知缓存类型巡检
+    known_types = _load_known_cache_types()
+    types_to_check: Iterable[str]
+    if args.types:
+        types_to_check = args.types
+    else:
+        types_to_check = known_types.keys()
+
+    for t in types_to_check:
+        cfg = known_types.get(t)
+        if not cfg:
+            print(f"⚠️ 未知缓存类型: {t}，跳过")
+            continue
+        analyze_cache_type(
+            client=client,
+            cache_type=t,
+            cfg=cfg,
+            sample_size=args.sample_size,
+            max_scan=args.max_scan,
+        )
+
+    print("巡检完成。")
+
+
+if __name__ == "__main__":
+    main()
+
@@ -7,19 +7,19 @@
 使用方法：
  
 直接使用（默认数据库 0）：
-python scripts/redis/check_cache_stats.py 
+python scripts/redis/redis_cache_prefix_stats.py
  
 统计所有数据库：
-python scripts/redis/check_cache_stats.py --all-db
+python scripts/redis/redis_cache_prefix_stats.py --all-db
  
 统计指定数据库：
-python scripts/redis/check_cache_stats.py --db 1
+python scripts/redis/redis_cache_prefix_stats.py --db 1
  
 只统计以下三种前缀：
-python scripts/redis/check_cache_stats.py --prefix trans embedding product
+python scripts/redis/redis_cache_prefix_stats.py --prefix trans embedding product
  
 统计所有数据库的指定前缀：
-python scripts/redis/check_cache_stats.py --all-db --prefix trans embedding
+python scripts/redis/redis_cache_prefix_stats.py --all-db --prefix trans embedding
  
  
  
@@ -487,7 +487,7 @@ def analyze_all_databases(args):
         print(f"总 key 数量: {total_keys_all_db:,}")
         print(f"\n提示: 要查看详细的内存统计，请分别运行每个数据库:")
         for db_num in databases:
-            print(f"  python scripts/redis/check_cache_stats.py --db {db_num}")
+            print(f"  python scripts/redis/redis_cache_prefix_stats.py --db {db_num}")
  
 def main():
     """主函数"""