From c68af6f806a6b20bc5cca045a7b2071634c8ff1b Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Wed, 8 Jul 2026 17:39:04 +0800 Subject: [PATCH] Cache Feishu retrieval data in Redis --- .../app/services/feishu_async_service.py | 37 +++++++++++++++ .../app/services/redis_cache_service.py | 45 +++++++++++++++++++ ...026-07-08-concurrency-optimization-plan.md | 34 ++++++++++++++ 3 files changed, 116 insertions(+) create mode 100644 ai_knowledge_base_v2/apps/backend/app/services/redis_cache_service.py diff --git a/ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py b/ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py index ca4f2c3..cec5612 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py +++ b/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}" diff --git a/ai_knowledge_base_v2/apps/backend/app/services/redis_cache_service.py b/ai_knowledge_base_v2/apps/backend/app/services/redis_cache_service.py new file mode 100644 index 0000000..b313116 --- /dev/null +++ b/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) diff --git a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md b/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md index a29ab49..8c41f24 100644 --- a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md +++ b/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、缓存清理和单知识库刷新能力。