Cache Feishu retrieval data in Redis

ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py

@@ -21,6 +21,7 @@ from app.services.feishu_service import (
     _token_cache_key,
 )
 from app.services.knowledge_service import KnowledgeScope
+from app.services.redis_cache_service import redis_get_json, redis_get_text, redis_set_json, redis_set_text
 
 if TYPE_CHECKING:
     from app.services.rag_service import RetrievedChunk
@@ -199,43 +200,61 @@ class AsyncFeishuKnowledgeService:
     @classmethod
     async def _get_node_info_with_cache(cls, node_token: str, config: FeishuRetrievalConfig) -> dict:
         cache_key = f"{config.app_id}:node_info:{node_token}"
+        redis_key = _redis_cache_key(config, "node", node_token)
         now = time.time()
 
         if cache_key in FeishuKnowledgeService._content_cache:
             cached_data, ts = FeishuKnowledgeService._content_cache[cache_key]
             if now - ts < FeishuKnowledgeService._CACHE_TTL:
                 return eval(cached_data)  # noqa: S307
+        cached_info = await redis_get_json(redis_key)
+        if isinstance(cached_info, dict):
+            FeishuKnowledgeService._content_cache[cache_key] = (str(cached_info), now)
+            return cached_info
 
         info = await _get_wiki_node_info_async(node_token, config)
         FeishuKnowledgeService._content_cache[cache_key] = (str(info), now)
+        await redis_set_json(redis_key, info, FeishuKnowledgeService._CACHE_TTL)
         return info
 
     @classmethod
     async def _get_document_content_with_cache(cls, doc_token: str, config: FeishuRetrievalConfig) -> str:
         cache_key = f"{config.app_id}:doc:{doc_token}"
+        redis_key = _redis_cache_key(config, "doc", doc_token)
         now = time.time()
 
         if cache_key in FeishuKnowledgeService._content_cache:
             content, ts = FeishuKnowledgeService._content_cache[cache_key]
             if now - ts < FeishuKnowledgeService._CACHE_TTL:
                 return content
+        cached_content = await redis_get_text(redis_key)
+        if cached_content:
+            FeishuKnowledgeService._content_cache[cache_key] = (cached_content, now)
+            return cached_content
 
         content = await _get_docx_content_async(doc_token, config)
         FeishuKnowledgeService._content_cache[cache_key] = (content, now)
+        await redis_set_text(redis_key, content, FeishuKnowledgeService._CACHE_TTL)
         return content
 
     @classmethod
     async def _get_file_content_with_cache(cls, file_token: str, config: FeishuRetrievalConfig) -> str:
         cache_key = f"{config.app_id}:file:{file_token}"
+        redis_key = _redis_cache_key(config, "file", file_token)
         now = time.time()
 
         if cache_key in FeishuKnowledgeService._content_cache:
             content, ts = FeishuKnowledgeService._content_cache[cache_key]
             if now - ts < FeishuKnowledgeService._CACHE_TTL:
                 return content
+        cached_content = await redis_get_text(redis_key)
+        if cached_content:
+            FeishuKnowledgeService._content_cache[cache_key] = (cached_content, now)
+            return cached_content
 
         content = await _get_file_content_async(file_token, config)
         FeishuKnowledgeService._content_cache[cache_key] = (content, now)
+        await redis_set_text(redis_key, content, FeishuKnowledgeService._CACHE_TTL)
         return content
 
     @classmethod
@@ -246,25 +265,36 @@ class AsyncFeishuKnowledgeService:
         config: FeishuRetrievalConfig,
     ) -> list[dict]:
         cache_key = f"{config.app_id}:children:{space_id}:{node_token}"
+        redis_key = _redis_cache_key(config, "children", space_id, node_token)
         now = time.time()
 
         if cache_key in FeishuKnowledgeService._content_cache:
             cached_data, ts = FeishuKnowledgeService._content_cache[cache_key]
             if now - ts < FeishuKnowledgeService._CACHE_TTL:
                 return eval(cached_data)  # noqa: S307
+        cached_children = await redis_get_json(redis_key)
+        if isinstance(cached_children, list):
+            FeishuKnowledgeService._content_cache[cache_key] = (str(cached_children), now)
+            return cached_children
 
         children = await _get_wiki_children_async(space_id, node_token, config)
         FeishuKnowledgeService._content_cache[cache_key] = (str(children), now)
+        await redis_set_json(redis_key, children, FeishuKnowledgeService._CACHE_TTL)
         return children
 
 
 async def _get_tenant_token_async(config: FeishuRetrievalConfig) -> str:
     now = time.time()
     cache_key = _token_cache_key(config.app_id, config.app_secret)
+    redis_key = _redis_cache_key(config, "tenant_token")
     cached = _token_cache.get(cache_key)
 
     if cached and cached["token"] and now < cached["expires_at"]:
         return cached["token"]
+    cached_token = await redis_get_text(redis_key)
+    if cached_token:
+        _token_cache[cache_key] = {"token": cached_token, "expires_at": now + 300}
+        return cached_token
 
     url = f"{FEISHU_OPEN_API}/open-apis/auth/v3/tenant_access_token/internal"
     payload = {"app_id": config.app_id, "app_secret": config.app_secret}
@@ -283,6 +313,7 @@ async def _get_tenant_token_async(config: FeishuRetrievalConfig) -> str:
         token = data["tenant_access_token"]
         expire = data.get("expire", 7200)
         _token_cache[cache_key] = {"token": token, "expires_at": now + expire - 300}
+        await redis_set_text(redis_key, token, max(60, int(expire) - 300))
         return token
     except httpx.HTTPError as exc:
         raise ExternalServiceError(f"获取飞书Token网络错误：{exc}", provider="feishu") from exc
@@ -386,3 +417,9 @@ async def _get_wiki_children_async(
             break
 
     return all_children
+
+
+def _redis_cache_key(config: FeishuRetrievalConfig, *parts: str) -> str:
+    namespace = _token_cache_key(config.app_id, config.app_secret)
+    suffix = ":".join(str(part) for part in parts)
+    return f"ai_kb:feishu:{namespace}:{suffix}"

ai_knowledge_base_v2/apps/backend/app/services/redis_cache_service.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from app.services.redis_client import get_redis_client
+
+
+async def redis_get_text(key: str) -> str | None:
+    client = get_redis_client()
+    if client is None:
+        return None
+    try:
+        value = await client.get(key)
+    except Exception:
+        return None
+    return value if isinstance(value, str) else None
+
+
+async def redis_set_text(key: str, value: str, ttl_seconds: int) -> None:
+    client = get_redis_client()
+    if client is None:
+        return
+    try:
+        await client.set(key, value, ex=ttl_seconds)
+    except Exception:
+        return
+
+
+async def redis_get_json(key: str) -> Any | None:
+    value = await redis_get_text(key)
+    if not value:
+        return None
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return None
+
+
+async def redis_set_json(key: str, value: Any, ttl_seconds: int) -> None:
+    try:
+        payload = json.dumps(value, ensure_ascii=False)
+    except (TypeError, ValueError):
+        return
+    await redis_set_text(key, payload, ttl_seconds)

ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md

@@ -418,3 +418,37 @@
 1. 当前 Redis 队列使用轮询晋升等待请求，下一步可以结合 Pub/Sub 或 blocking pop 降低等待侧轮询。
 2. Redis 飞书共享缓存还未接入，将在阶段 4B 完成。
 3. 生产部署需要确认 Redis 持久化、内存上限、淘汰策略和监控告警。
+
+## 阶段 4B 实施记录
+
+2026-07-08 完成 Redis 飞书共享缓存第一版。
+
+已完成：
+
+1. 新增 `redis_cache_service`，提供 Redis 文本和 JSON 缓存读写 helper。
+2. 用户端 async 飞书/RAG 链路已接入 Redis 缓存：
+   - tenant token。
+   - Wiki node info。
+   - Wiki children。
+   - Docx 文本内容。
+   - Drive file 文本内容。
+3. 缓存 key 按飞书应用凭证 hash 和资源 token 隔离。
+4. Redis 不可用或缓存读写失败时自动降级，不阻断问答主流程。
+5. 保留进程内缓存作为本进程内的一级缓存，Redis 作为跨 worker / 跨实例共享缓存。
+
+验证记录：
+
+1. 本机 `compileall` 检查通过。
+2. `git diff --check` 通过。
+3. 容器内 `compileall` 检查通过。
+4. 容器内 Redis 文本和 JSON 缓存读写验证通过。
+5. 后端健康检查通过。
+6. 真实 `/chat/completions` 验证通过：首个事件为 `generating`，随后返回 `content`，最后返回 `[DONE]`。
+7. 真实会话历史验证通过：流式完成后用户消息和助手消息均已落库。
+8. 真实问答后 Redis 中已生成飞书缓存 key。
+
+阶段限制：
+
+1. 当前 Redis 缓存 TTL 仍复用进程内缓存的 5 分钟默认值。
+2. 同步旧飞书链路暂未接入 Redis 缓存，用户端 async 问答链路已接入。
+3. 后续可以在系统设置开放飞书缓存 TTL、缓存清理和单知识库刷新能力。