diff --git a/apps/api/src/repositories/embedding_repository.py b/apps/api/src/repositories/embedding_repository.py new file mode 100644 index 000000000..aba981b51 --- /dev/null +++ b/apps/api/src/repositories/embedding_repository.py @@ -0,0 +1,172 @@ +""" +Embedding Cache Repository - Redis 向量快取 +============================================ +2026-03-27 ogt: P1 違規修復 - 模組化改造 + +職責: Playbook 向量快取 CRUD (Redis) +設計: 依賴注入 Redis client，遵循 Repository Pattern + +版本: v1.0 +建立: 2026-03-27 (台北時區) +建立者: Claude Code (P1 模組化改造) +""" + +import json +from datetime import UTC, datetime + +import structlog +from redis.asyncio import Redis + +logger = structlog.get_logger(__name__) + +# Redis Keys +PLAYBOOK_EMBEDDING_PREFIX = "playbook:embedding:" +PLAYBOOK_EMBEDDING_INDEX = "playbook:embedding:index" + +# Cache TTL: 30 天 +EMBEDDING_TTL_SECONDS = 30 * 24 * 60 * 60 + + +class EmbeddingCacheRepository: + """ + Embedding Cache Repository 實作 + + 使用 Redis 儲存 Playbook 向量快取 + """ + + def __init__(self, redis: Redis): + """ + 初始化 Repository + + Args: + redis: Redis async client (DI 注入) + """ + self._redis = redis + + async def store( + self, + playbook_id: str, + embedding: list[float], + metadata: dict | None = None, + ) -> bool: + """ + 儲存 Playbook 向量到 Redis + + 存儲格式: + - playbook:embedding:{id} -> {vector: [...], metadata: {...}} + - playbook:embedding:index -> Set of playbook_ids + """ + try: + key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" + + data = { + "vector": embedding, + "metadata": metadata or {}, + "updated_at": datetime.now(UTC).isoformat(), + } + + await self._redis.set( + key, + json.dumps(data), + ex=EMBEDDING_TTL_SECONDS, + ) + + # 更新索引 + await self._redis.sadd(PLAYBOOK_EMBEDDING_INDEX, playbook_id) + + logger.debug( + "playbook_embedding_stored", + playbook_id=playbook_id, + vector_dim=len(embedding), + ) + return True + + except Exception as e: + logger.warning( + "playbook_embedding_store_failed", + playbook_id=playbook_id, + error=str(e), + ) + return False + + async def get(self, playbook_id: str) -> list[float] | None: + """取得 Playbook 向量""" + try: + key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" + + data = await self._redis.get(key) + if not data: + return None + + parsed = json.loads(data) + return parsed.get("vector") + + except Exception as e: + logger.warning( + "playbook_embedding_get_failed", + playbook_id=playbook_id, + error=str(e), + ) + return None + + async def get_all(self) -> dict[str, list[float]]: + """取得所有 Playbook 向量""" + try: + # 取得所有 playbook_id + playbook_ids = await self._redis.smembers(PLAYBOOK_EMBEDDING_INDEX) + if not playbook_ids: + return {} + + # 批次取得向量 + result = {} + for pid in playbook_ids: + pid_str = pid.decode() if isinstance(pid, bytes) else pid + vec = await self.get(pid_str) + if vec: + result[pid_str] = vec + + return result + + except Exception as e: + logger.warning( + "playbook_embeddings_get_all_failed", + error=str(e), + ) + return {} + + async def remove(self, playbook_id: str) -> bool: + """移除 Playbook 向量索引""" + try: + key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" + + await self._redis.delete(key) + await self._redis.srem(PLAYBOOK_EMBEDDING_INDEX, playbook_id) + + logger.info("playbook_embedding_removed", playbook_id=playbook_id) + return True + + except Exception as e: + logger.warning( + "playbook_embedding_remove_failed", + playbook_id=playbook_id, + error=str(e), + ) + return False + + +# ============================================================================= +# Factory Function (for DI) +# ============================================================================= + + +def create_embedding_cache_repository(redis: Redis) -> EmbeddingCacheRepository: + """ + 建立 EmbeddingCacheRepository 實例 + + Args: + redis: Redis async client + + Returns: + EmbeddingCacheRepository 實例 + """ + return EmbeddingCacheRepository(redis) diff --git a/apps/api/src/repositories/interfaces.py b/apps/api/src/repositories/interfaces.py index d7fbc32aa..fb7837ae5 100644 --- a/apps/api/src/repositories/interfaces.py +++ b/apps/api/src/repositories/interfaces.py @@ -243,3 +243,36 @@ class IPlaybookRepository(Protocol): ) -> bool: """更新執行統計""" ... + + +@runtime_checkable +class IEmbeddingCacheRepository(Protocol): + """ + Embedding Cache Repository Protocol + + 職責: Playbook 向量快取 (Redis) + 實作: EmbeddingCacheRepository + + 2026-03-27 ogt: 模組化改造 (P1 違規修復) + """ + + async def store( + self, + playbook_id: str, + embedding: list[float], + metadata: dict | None = None, + ) -> bool: + """儲存 Playbook 向量""" + ... + + async def get(self, playbook_id: str) -> list[float] | None: + """取得 Playbook 向量""" + ... + + async def get_all(self) -> dict[str, list[float]]: + """取得所有 Playbook 向量""" + ... + + async def remove(self, playbook_id: str) -> bool: + """移除 Playbook 向量""" + ... diff --git a/apps/api/src/services/playbook_rag.py b/apps/api/src/services/playbook_rag.py index a63f2c9c1..979ae7985 100644 --- a/apps/api/src/services/playbook_rag.py +++ b/apps/api/src/services/playbook_rag.py @@ -12,22 +12,26 @@ ADR-030: 智能自動修復系統 - Embedding 快取，避免重複計算 - 混合搜尋 (向量 + Jaccard) - Fallback: Embedding 失敗時用 Jaccard +- 2026-03-27 ogt: 模組化改造 (P1 違規修復) + - Repository Pattern for Redis + - DI httpx client from Lifespan -版本: v1.0 +版本: v1.1 建立: 2026-03-26 (台北時區) +修改: 2026-03-27 (模組化改造) """ -import json import math from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any +import httpx import structlog from src.core.config import settings -from src.core.redis_client import get_redis from src.models.playbook import Playbook, SymptomPattern +from src.repositories.interfaces import IEmbeddingCacheRepository logger = structlog.get_logger(__name__) @@ -40,13 +44,6 @@ logger = structlog.get_logger(__name__) EMBEDDING_MODEL = "nomic-embed-text" EMBEDDING_DIM = 768 # nomic-embed-text 向量維度 -# Redis Keys -PLAYBOOK_EMBEDDING_PREFIX = "playbook:embedding:" -PLAYBOOK_EMBEDDING_INDEX = "playbook:embedding:index" - -# Cache TTL: 30 天 (Embedding 不常變化) -EMBEDDING_TTL_SECONDS = 30 * 24 * 60 * 60 - # ============================================================================= # Data Models @@ -130,9 +127,25 @@ class PlaybookRAGService: 1. 將 Playbook 向量化並存入 Redis 2. 語意搜尋相似的 Playbook 3. 混合搜尋 (向量 + Jaccard) + + 2026-03-27 ogt: 模組化改造 + - 使用 DI 注入 http_client 和 embedding_cache """ - def __init__(self): + def __init__( + self, + http_client: httpx.AsyncClient, + embedding_cache: IEmbeddingCacheRepository, + ): + """ + 初始化 RAG Service + + Args: + http_client: httpx AsyncClient (DI 注入，來自 Lifespan) + embedding_cache: Embedding Cache Repository (DI 注入) + """ + self._http_client = http_client + self._embedding_cache = embedding_cache self.ollama_url = settings.OLLAMA_URL self.embedding_model = EMBEDDING_MODEL @@ -144,6 +157,8 @@ class PlaybookRAGService: """ 使用 Ollama 生成文字 embedding + 2026-03-27 ogt: 改用 DI 注入的 http_client (P1 違規修復) + Args: text: 要向量化的文字 @@ -151,33 +166,32 @@ class PlaybookRAGService: 向量 (768 維) 或 None (失敗時) """ try: - import httpx + # 使用 DI 注入的 http_client，設置單次請求 timeout + response = await self._http_client.post( + f"{self.ollama_url}/api/embeddings", + json={ + "model": self.embedding_model, + "prompt": text, + }, + timeout=30.0, # 單次請求 timeout + ) - async with httpx.AsyncClient(timeout=30.0) as client: - response = await client.post( - f"{self.ollama_url}/api/embeddings", - json={ - "model": self.embedding_model, - "prompt": text, - }, + if response.status_code != 200: + logger.warning( + "ollama_embedding_failed", + status_code=response.status_code, + text_preview=text[:50], ) + return None - if response.status_code != 200: - logger.warning( - "ollama_embedding_failed", - status_code=response.status_code, - text_preview=text[:50], - ) - return None + result = response.json() + embedding = result.get("embedding", []) - result = response.json() - embedding = result.get("embedding", []) + if not embedding: + logger.warning("ollama_embedding_empty", text_preview=text[:50]) + return None - if not embedding: - logger.warning("ollama_embedding_empty", text_preview=text[:50]) - return None - - return normalize_vector(embedding) + return normalize_vector(embedding) except Exception as e: logger.warning( @@ -249,7 +263,8 @@ class PlaybookRAGService: return await self.embed_text(text) # ========================================================================= - # Storage Operations + # Storage Operations (委派給 Repository) + # 2026-03-27 ogt: 改用 DI 注入的 embedding_cache (P1 違規修復) # ========================================================================= async def store_playbook_embedding( @@ -258,97 +273,19 @@ class PlaybookRAGService: embedding: list[float], metadata: dict | None = None, ) -> bool: - """ - 儲存 Playbook 向量到 Redis - - 存儲格式: - - playbook:embedding:{id} -> {vector: [...], metadata: {...}} - - playbook:embedding:index -> Set of playbook_ids - """ - try: - redis = get_redis() - key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" - - data = { - "vector": embedding, - "metadata": metadata or {}, - "updated_at": datetime.now(UTC).isoformat(), - } - - await redis.set( - key, - json.dumps(data), - ex=EMBEDDING_TTL_SECONDS, - ) - - # 更新索引 - await redis.sadd(PLAYBOOK_EMBEDDING_INDEX, playbook_id) - - logger.debug( - "playbook_embedding_stored", - playbook_id=playbook_id, - vector_dim=len(embedding), - ) - return True - - except Exception as e: - logger.warning( - "playbook_embedding_store_failed", - playbook_id=playbook_id, - error=str(e), - ) - return False + """儲存 Playbook 向量到 Redis (委派給 Repository)""" + return await self._embedding_cache.store(playbook_id, embedding, metadata) async def get_playbook_embedding( self, playbook_id: str, ) -> list[float] | None: - """取得 Playbook 向量""" - try: - redis = get_redis() - key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" - - data = await redis.get(key) - if not data: - return None - - parsed = json.loads(data) - return parsed.get("vector") - - except Exception as e: - logger.warning( - "playbook_embedding_get_failed", - playbook_id=playbook_id, - error=str(e), - ) - return None + """取得 Playbook 向量 (委派給 Repository)""" + return await self._embedding_cache.get(playbook_id) async def get_all_playbook_embeddings(self) -> dict[str, list[float]]: - """取得所有 Playbook 向量""" - try: - redis = get_redis() - - # 取得所有 playbook_id - playbook_ids = await redis.smembers(PLAYBOOK_EMBEDDING_INDEX) - if not playbook_ids: - return {} - - # 批次取得向量 - result = {} - for pid in playbook_ids: - pid_str = pid.decode() if isinstance(pid, bytes) else pid - vec = await self.get_playbook_embedding(pid_str) - if vec: - result[pid_str] = vec - - return result - - except Exception as e: - logger.warning( - "playbook_embeddings_get_all_failed", - error=str(e), - ) - return {} + """取得所有 Playbook 向量 (委派給 Repository)""" + return await self._embedding_cache.get_all() # ========================================================================= # Search Operations @@ -561,24 +498,8 @@ class PlaybookRAGService: ) async def remove_playbook_index(self, playbook_id: str) -> bool: - """移除 Playbook 向量索引""" - try: - redis = get_redis() - key = f"{PLAYBOOK_EMBEDDING_PREFIX}{playbook_id}" - - await redis.delete(key) - await redis.srem(PLAYBOOK_EMBEDDING_INDEX, playbook_id) - - logger.info("playbook_index_removed", playbook_id=playbook_id) - return True - - except Exception as e: - logger.warning( - "playbook_index_remove_failed", - playbook_id=playbook_id, - error=str(e), - ) - return False + """移除 Playbook 向量索引 (委派給 Repository)""" + return await self._embedding_cache.remove(playbook_id) async def reindex_all_playbooks( self, @@ -610,15 +531,54 @@ class PlaybookRAGService: # ============================================================================= -# Singleton +# Factory (DI-aware) +# 2026-03-27 ogt: 模組化改造 - 支援 DI 注入 # ============================================================================= _rag_service: PlaybookRAGService | None = None -def get_playbook_rag_service() -> PlaybookRAGService: - """取得 Playbook RAG 服務 singleton""" +async def get_playbook_rag_service() -> PlaybookRAGService: + """ + 取得 Playbook RAG 服務 singleton (lazy initialization) + + 2026-03-27 ogt: 改用 DI 注入，從 Lifespan 取得 http_client 和 Redis + """ global _rag_service if _rag_service is None: - _rag_service = PlaybookRAGService() + # 延遲導入避免循環依賴 + from src.core.http_client import get_general_client + from src.core.redis_client import get_redis + from src.repositories.embedding_repository import EmbeddingCacheRepository + + http_client = await get_general_client() + redis = get_redis() + embedding_cache = EmbeddingCacheRepository(redis) + + _rag_service = PlaybookRAGService( + http_client=http_client, + embedding_cache=embedding_cache, + ) return _rag_service + + +def create_playbook_rag_service( + http_client: httpx.AsyncClient, + embedding_cache: IEmbeddingCacheRepository, +) -> PlaybookRAGService: + """ + 建立 PlaybookRAGService 實例 (工廠函數) + + 用於測試或需要自訂依賴的場景 + + Args: + http_client: httpx AsyncClient + embedding_cache: Embedding Cache Repository + + Returns: + PlaybookRAGService 實例 + """ + return PlaybookRAGService( + http_client=http_client, + embedding_cache=embedding_cache, + ) diff --git a/apps/api/src/services/playbook_service.py b/apps/api/src/services/playbook_service.py index 8a9e3b0dc..a6b4c67c2 100644 --- a/apps/api/src/services/playbook_service.py +++ b/apps/api/src/services/playbook_service.py @@ -87,7 +87,14 @@ class PlaybookService: def __init__(self, repository: IPlaybookRepository | None = None): self._repository = repository or get_playbook_repository() - self._rag_service = get_playbook_rag_service() + # 2026-03-27 ogt: RAG Service 改為 lazy initialization (async factory) + self._rag_service = None + + async def _get_rag_service(self): + """Lazy initialization for RAG service (2026-03-27 async factory)""" + if self._rag_service is None: + self._rag_service = await get_playbook_rag_service() + return self._rag_service # === Core Operations === @@ -182,7 +189,8 @@ class PlaybookService: async def _index_playbook_async(self, playbook: Playbook) -> None: """非同步建立 Playbook 向量索引 (ADR-030 Phase 3)""" try: - success = await self._rag_service.index_playbook(playbook) + rag_service = await self._get_rag_service() + success = await rag_service.index_playbook(playbook) if success: logger.debug( "playbook_indexed", @@ -223,7 +231,8 @@ class PlaybookService: # Step 2: RAG 混合搜尋 (如果啟用) if use_rag and symptoms.alert_names: try: - hybrid_matches = await self._rag_service.hybrid_search( + rag_service = await self._get_rag_service() + hybrid_matches = await rag_service.hybrid_search( symptoms=symptoms, jaccard_results=jaccard_results, top_k=top_k * 2,