183a57e7f1
- 将全部路由文件（12个）的 jsonify 响应迁移至 success_response/error_response 统一格式 - 修复 sync_routes.py error_response 参数错误（P0） - 新增异步任务系统：task_registry + task_routes - 新增状态码：TASK_NOT_FOUND(4014)、TASK_CONFLICT(4015)、REINDEX_ERROR(5040) - 修正 task_routes/exam_pkg 中语义不匹配的状态码 - 更新 curl 测试手册、后端对接规范文档 - 添加缓存性能报告和 Redis 迁移计划
343 lines
13 KiB
Markdown
343 lines
13 KiB
Markdown
## RAG 缓存 Redis 迁移方案
|
||
|
||
### 一、迁移目标
|
||
|
||
将当前四层进程内缓存迁移到 Redis，实现跨进程/跨实例共享、重启不丢失、多 worker 缓存一致。
|
||
|
||
### 二、迁移优先级
|
||
|
||
| 优先级 | 缓存层 | 复杂度 | 理由 |
|
||
|--------|--------|--------|------|
|
||
| P0 | Rerank Cache | 极低 | Redis Hash 天然匹配，无版本失效，0.75MB |
|
||
| P1 | Query Cache | 低 | 价值最大（跳过整个检索管线），接口简单 |
|
||
| P2 | Embedding Cache | 低~中 | 需注意 float 数组序列化效率和 MGET 批量优化 |
|
||
| P3 | Semantic Cache | 高 | FAISS 向量索引无法直接替换，建议混合方案 |
|
||
|
||
### 三、总体架构设计
|
||
|
||
```
|
||
┌──────────────────────────────────────────────┐
|
||
│ Gunicorn Worker 1..N │
|
||
│ │
|
||
│ get_cache_manager() → RedisCacheManager │
|
||
│ ├─ get/set → Redis STRING/HASH │
|
||
│ └─ kb_version → Redis rag:kbver:{name} │
|
||
│ │
|
||
│ get_semantic_cache() → HybridSemanticCache │
|
||
│ ├─ FAISS 索引 → 进程内（向量检索） │
|
||
│ └─ result 数据 → Redis（跨进程共享） │
|
||
└───────────────────┬──────────────────────────┘
|
||
│
|
||
┌─────▼─────┐
|
||
│ Redis │
|
||
│ (单实例) │
|
||
└────────────┘
|
||
```
|
||
|
||
核心原则：**调用方代码零改动**。`get_cache_manager()` 和 `get_semantic_cache()` 函数签名不变，内部实现从 LRUCache 切换为 Redis。engine.py、chat_routes.py、sync.py 等所有调用方不需要任何修改。
|
||
|
||
### 四、Redis Key 设计
|
||
|
||
#### 4.1 版本号机制
|
||
|
||
将 `kb_version` 存入 Redis，而非编入 key。原因：编入 key 会导致版本号变化后旧 key 残留在 Redis 中直到 TTL 过期，浪费内存。
|
||
|
||
```
|
||
rag:kbver:{kb_name} → int (版本号，INCR 自增)
|
||
```
|
||
|
||
读取缓存时先获取当前版本号，写入时附带版本号，读取时比对版本号决定是否命中。
|
||
|
||
#### 4.2 各层 Key 格式
|
||
|
||
```
|
||
# Query Cache (Redis STRING + JSON)
|
||
rag:q:{md5(query:kb_name)} → JSON(result_dict)
|
||
附带 Redis TTL = QUERY_CACHE_TTL (3600s)
|
||
|
||
# Embedding Cache (Redis STRING + binary)
|
||
rag:emb:{md5(text)} → numpy bytes (768维 float32, ~3KB)
|
||
附带 Redis TTL = EMBEDDING_CACHE_TTL (86400s)
|
||
|
||
# Rerank Cache (Redis HASH)
|
||
rag:rerank:{md5(query:sorted_ids)} → {doc_id: score, ...}
|
||
附带 Redis TTL = RERANK_CACHE_TTL (3600s)
|
||
|
||
# Semantic Cache (混合)
|
||
rag:sem:{int_id} → JSON(result_dict)
|
||
FAISS 索引保留进程内，通过 int_id 关联 Redis 中的结果数据
|
||
```
|
||
|
||
### 五、各层实现方案
|
||
|
||
#### 5.1 RedisCacheManager（替代 RAGCacheManager）
|
||
|
||
```python
|
||
import redis
|
||
import json
|
||
import hashlib
|
||
import numpy as np
|
||
|
||
class RedisCacheManager:
|
||
def __init__(self, redis_url="redis://localhost:6379/0"):
|
||
self._pool = redis.ConnectionPool.from_url(
|
||
redis_url, decode_responses=False, max_connections=10
|
||
)
|
||
self._r = redis.Redis(connection_pool=self._pool)
|
||
self._stats = {...} # 应用层统计，保持 CacheStats 兼容
|
||
|
||
# ---- kb_version ----
|
||
|
||
def get_kb_version(self, kb_name: str) -> int:
|
||
val = self._r.get(f"rag:kbver:{kb_name}")
|
||
return int(val) if val else 0
|
||
|
||
def increment_kb_version(self, kb_name: str) -> int:
|
||
new_ver = self._r.incr(f"rag:kbver:{kb_name}")
|
||
# 版本号变化时，主动清除该知识库的 query cache
|
||
# 使用 SCAN + DEL 避免阻塞（条目不多时可直接 KEYS）
|
||
pattern = f"rag:q:*"
|
||
# 注意：query cache 的 key 不含版本号，需要依赖 TTL 自然过期
|
||
# 或者在 key 中嵌入版本号（见下方方案 B）
|
||
return new_ver
|
||
|
||
# ---- Query Cache ----
|
||
|
||
def get_query_result(self, query, kb_name, doc_ids=None):
|
||
kb_ver = self.get_kb_version(kb_name)
|
||
key = self._query_key(query, kb_name, kb_ver)
|
||
data = self._r.get(key)
|
||
if data is None:
|
||
self._stats['query'].misses += 1
|
||
return None
|
||
self._stats['query'].hits += 1
|
||
return json.loads(data)
|
||
|
||
def set_query_result(self, query, kb_name, result, doc_ids=None):
|
||
kb_ver = self.get_kb_version(kb_name)
|
||
key = self._query_key(query, kb_name, kb_ver)
|
||
self._r.set(key, json.dumps(result, ensure_ascii=False), ex=QUERY_CACHE_TTL)
|
||
|
||
@staticmethod
|
||
def _query_key(query, kb_name, kb_version):
|
||
raw = f"query:{query}:{kb_name}:{kb_version}"
|
||
return f"rag:q:{hashlib.md5(raw.encode()).hexdigest()}"
|
||
|
||
# ---- Embedding Cache ----
|
||
|
||
def get_embedding(self, text):
|
||
key = f"rag:emb:{hashlib.md5(f'emb:{text}'.encode()).hexdigest()}"
|
||
data = self._r.get(key)
|
||
if data is None:
|
||
self._stats['embedding'].misses += 1
|
||
return None
|
||
self._stats['embedding'].hits += 1
|
||
return np.frombuffer(data, dtype=np.float32).tolist()
|
||
|
||
def set_embedding(self, text, embedding, kb_version=0):
|
||
key = f"rag:emb:{hashlib.md5(f'emb:{text}'.encode()).hexdigest()}"
|
||
arr = np.array(embedding, dtype=np.float32)
|
||
self._r.set(key, arr.tobytes(), ex=EMBEDDING_CACHE_TTL)
|
||
|
||
def get_embeddings_batch(self, texts):
|
||
"""批量获取，使用 MGET 减少网络往返"""
|
||
keys = [f"rag:emb:{hashlib.md5(f'emb:{t}'.encode()).hexdigest()}" for t in texts]
|
||
results = self._r.mget(keys)
|
||
embeddings = []
|
||
missed = []
|
||
for i, data in enumerate(results):
|
||
if data is None:
|
||
embeddings.append(None)
|
||
missed.append(i)
|
||
else:
|
||
embeddings.append(np.frombuffer(data, dtype=np.float32).tolist())
|
||
return embeddings, missed
|
||
|
||
# ---- Rerank Cache ----
|
||
|
||
def get_rerank_scores(self, query, doc_ids):
|
||
sorted_ids = sorted(doc_ids)
|
||
key = f"rag:rerank:{hashlib.md5(f'rerank:{query}:{':'.join(sorted_ids)}'.encode()).hexdigest()}"
|
||
data = self._r.hgetall(key)
|
||
if not data:
|
||
self._stats['rerank'].misses += 1
|
||
return None
|
||
self._stats['rerank'].hits += 1
|
||
return {k.decode(): float(v) for k, v in data.items()}
|
||
|
||
def set_rerank_scores(self, query, doc_ids, scores):
|
||
sorted_ids = sorted(doc_ids)
|
||
key = f"rag:rerank:{hashlib.md5(f'rerank:{query}:{':'.join(sorted_ids)}'.encode()).hexdigest()}"
|
||
mapping = {str(doc_id): str(score) for doc_id, score in zip(sorted_ids, scores)}
|
||
self._r.hset(key, mapping=mapping)
|
||
self._r.expire(key, RERANK_CACHE_TTL)
|
||
|
||
# ---- 统计与清除 ----
|
||
|
||
def get_all_stats(self):
|
||
return self._stats # CacheStats 兼容
|
||
|
||
def clear_all(self):
|
||
# 只删除 rag: 前缀的 key
|
||
for pattern in ["rag:q:*", "rag:emb:*", "rag:rerank:*", "rag:sem:*"]:
|
||
cursor = 0
|
||
while True:
|
||
cursor, keys = self._r.scan(cursor, match=pattern, count=100)
|
||
if keys:
|
||
self._r.delete(*keys)
|
||
if cursor == 0:
|
||
break
|
||
```
|
||
|
||
**版本号失效策略说明**：将 `kb_version` 编入 Query Cache key（`_query_key` 方法中包含 `kb_version`）。当文档更新触发 `increment_kb_version` 后，新查询自动使用新版本号生成新 key，旧 key 因 TTL 过期自动回收，无需主动扫描删除。Embedding Cache 不做版本失效（向量本身不因文档增删而变化，只在新文档加入时自然 miss）。
|
||
|
||
#### 5.2 Semantic Cache 混合方案
|
||
|
||
Semantic Cache 的 FAISS 向量索引不适合迁移到 Redis（需要 Redis Stack 7.2+ 的向量搜索能力）。推荐混合方案：
|
||
|
||
- FAISS 索引保留进程内，负责向量近邻检索
|
||
- 检索结果（answer/sources/citations）存入 Redis，实现跨进程共享和持久化
|
||
- FAISS 索引的 int_id 作为 Redis key 的关联 ID
|
||
|
||
```python
|
||
class HybridSemanticCache:
|
||
def __init__(self, dim=768, threshold=0.92, max_size=10000, redis_client=None):
|
||
# FAISS 索引（进程内）
|
||
self._index = faiss.IndexFlatIP(dim)
|
||
self._vectors = [] # 用于 numpy 降级
|
||
self._dim = dim
|
||
self._threshold = threshold
|
||
self._max_size = max_size
|
||
self._redis = redis_client
|
||
self._local_results = {} # 降级用：FAISS id -> result (无 Redis 时)
|
||
self._next_id = 0
|
||
self._lock = threading.RLock()
|
||
self._hits = 0
|
||
self._misses = 0
|
||
|
||
def get(self, query_emb):
|
||
with self._lock:
|
||
# FAISS 检索
|
||
emb = query_emb.reshape(1, -1).astype(np.float32)
|
||
emb /= np.linalg.norm(emb)
|
||
D, I = self._index.search(emb, 1)
|
||
if D[0][0] <= self._threshold:
|
||
self._misses += 1
|
||
return None
|
||
|
||
faiss_id = int(I[0][0])
|
||
self._hits += 1
|
||
|
||
# 优先从 Redis 读取结果
|
||
if self._redis:
|
||
data = self._redis.get(f"rag:sem:{faiss_id}")
|
||
if data:
|
||
return json.loads(data)
|
||
|
||
# 降级：从本地 Dict 读取
|
||
return self._local_results.get(faiss_id)
|
||
|
||
def set(self, query_emb, result):
|
||
with self._lock:
|
||
# 容量检查
|
||
if self._index.ntotal >= self._max_size:
|
||
self._evict_half()
|
||
|
||
# 添加到 FAISS 索引
|
||
emb = query_emb.reshape(1, -1).astype(np.float32)
|
||
emb /= np.linalg.norm(emb)
|
||
self._index.add(emb)
|
||
faiss_id = self._next_id
|
||
self._next_id += 1
|
||
|
||
# 结果存入 Redis（如果有）
|
||
if self._redis:
|
||
self._redis.set(
|
||
f"rag:sem:{faiss_id}",
|
||
json.dumps(result, ensure_ascii=False),
|
||
ex=86400 # 24 小时 TTL
|
||
)
|
||
else:
|
||
self._local_results[faiss_id] = result
|
||
```
|
||
|
||
### 六、配置变更
|
||
|
||
在 `config.example.py` 中新增：
|
||
|
||
```python
|
||
# Redis 缓存（设置后自动启用 Redis 替代内存缓存）
|
||
REDIS_CACHE_URL = os.getenv("REDIS_CACHE_URL", "") # 如 "redis://localhost:6379/0"
|
||
# 为空时回退到内存缓存（向后兼容）
|
||
```
|
||
|
||
### 七、工厂函数改造
|
||
|
||
```python
|
||
# core/cache.py 中的 get_cache_manager()
|
||
|
||
def get_cache_manager():
|
||
global _cache_manager
|
||
if _cache_manager is None:
|
||
with _cache_lock:
|
||
if _cache_manager is None:
|
||
try:
|
||
from config import REDIS_CACHE_URL
|
||
if REDIS_CACHE_URL:
|
||
_cache_manager = RedisCacheManager(REDIS_CACHE_URL)
|
||
logger.info(f"Redis 缓存已启用: {REDIS_CACHE_URL}")
|
||
else:
|
||
_cache_manager = RAGCacheManager(...) # 原有内存缓存
|
||
logger.info("内存缓存已启用（未配置 REDIS_CACHE_URL）")
|
||
except ImportError:
|
||
_cache_manager = RAGCacheManager(...)
|
||
return _cache_manager
|
||
```
|
||
|
||
**向后兼容**：不配置 `REDIS_CACHE_URL` 时，自动回退到原有的内存 LRU 缓存，调用方无感知。
|
||
|
||
### 八、部署变更
|
||
|
||
docker-compose.prod.yml 新增 Redis 服务：
|
||
|
||
```yaml
|
||
services:
|
||
redis:
|
||
image: redis:7-alpine
|
||
command: redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru
|
||
ports:
|
||
- "6379:6379"
|
||
volumes:
|
||
- redis_data:/data
|
||
|
||
rag-service:
|
||
environment:
|
||
- REDIS_CACHE_URL=redis://redis:6379/0
|
||
- GUNICORN_WORKERS=2 # 现在可以安全地多 worker
|
||
depends_on:
|
||
- redis
|
||
```
|
||
|
||
Redis 配置 `maxmemory 128mb` + `allkeys-lru` 淘汰策略。按前面估算，四层缓存满载约 95MB，128MB 足够且留有余量。
|
||
|
||
### 九、迁移步骤建议
|
||
|
||
1. 新增 `core/redis_cache.py`，实现 `RedisCacheManager` 和 `HybridSemanticCache`
|
||
2. 改造 `core/cache.py` 的 `get_cache_manager()` 工厂函数，根据配置选择实现
|
||
3. 改造 `core/semantic_cache.py` 的 `get_semantic_cache()` 工厂函数
|
||
4. `config.example.py` 新增 `REDIS_CACHE_URL` 配置项
|
||
5. `docker-compose.prod.yml` 新增 Redis 服务
|
||
6. `deploy/gunicorn.conf.py` 将 `max_requests` 调高至 5000
|
||
7. 本地测试：配置 Redis 后运行缓存冷热对比测试，验证命中率
|
||
8. 服务器部署：添加 Redis 容器，配置环境变量
|
||
|
||
### 十、预期收益
|
||
|
||
| 指标 | 当前（内存缓存） | 迁移后（Redis 缓存） |
|
||
|------|------------------|---------------------|
|
||
| 多 worker 支持 | 不支持 | 支持 |
|
||
| 重启后缓存 | 丢失 | 保留（Redis 持久化） |
|
||
| max_requests 重启 | 缓存冷启动 | 无影响 |
|
||
| 内存占用 | ~95MB/worker | ~5MB/worker + 128MB Redis |
|
||
| 缓存一致性 | 多 worker 不一致 | 全局一致 |
|