diff --git a/apps/api/src/services/playbook_rag.py b/apps/api/src/services/playbook_rag.py new file mode 100644 index 000000000..a63f2c9c1 --- /dev/null +++ b/apps/api/src/services/playbook_rag.py @@ -0,0 +1,624 @@ +""" +Playbook RAG Service - Phase 3 向量化語意搜尋 +============================================= +ADR-030: 智能自動修復系統 + +使用 Embedding 進行 Playbook 語意搜尋: +1. Ollama nomic-embed-text 生成向量 +2. Redis 儲存向量 (JSON 格式) +3. 餘弦相似度搜尋 + +設計原則: +- Embedding 快取,避免重複計算 +- 混合搜尋 (向量 + Jaccard) +- Fallback: Embedding 失敗時用 Jaccard + +版本: v1.0 +建立: 2026-03-26 (台北時區) +""" + +import json +import math +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +import structlog + +from src.core.config import settings +from src.core.redis_client import get_redis +from src.models.playbook import Playbook, SymptomPattern + +logger = structlog.get_logger(__name__) + + +# ============================================================================= +# Constants +# ============================================================================= + +# Embedding Model (Ollama 本地) +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 +# ============================================================================= + + +@dataclass +class PlaybookMatch: + """Playbook 匹配結果""" + + playbook_id: str + similarity_score: float # 0.0 ~ 1.0 + match_type: str # "vector", "jaccard", "hybrid" + matched_keywords: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "playbook_id": self.playbook_id, + "similarity_score": round(self.similarity_score, 4), + "match_type": self.match_type, + "matched_keywords": self.matched_keywords, + } + + +@dataclass +class EmbeddingResult: + """Embedding 結果""" + + text: str + vector: list[float] + model: str + created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + def to_dict(self) -> dict[str, Any]: + return { + "text_hash": hash(self.text) % 2**32, # 不存完整 text + "vector": self.vector, + "model": self.model, + "created_at": self.created_at.isoformat(), + } + + +# ============================================================================= +# Vector Utilities +# ============================================================================= + + +def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float: + """計算餘弦相似度""" + if len(vec_a) != len(vec_b): + return 0.0 + + dot_product = sum(a * b for a, b in zip(vec_a, vec_b)) + norm_a = math.sqrt(sum(a * a for a in vec_a)) + norm_b = math.sqrt(sum(b * b for b in vec_b)) + + if norm_a == 0 or norm_b == 0: + return 0.0 + + return dot_product / (norm_a * norm_b) + + +def normalize_vector(vec: list[float]) -> list[float]: + """正規化向量 (L2 norm)""" + norm = math.sqrt(sum(v * v for v in vec)) + if norm == 0: + return vec + return [v / norm for v in vec] + + +# ============================================================================= +# Playbook RAG Service +# ============================================================================= + + +class PlaybookRAGService: + """ + Playbook RAG 服務 + + 功能: + 1. 將 Playbook 向量化並存入 Redis + 2. 語意搜尋相似的 Playbook + 3. 混合搜尋 (向量 + Jaccard) + """ + + def __init__(self): + self.ollama_url = settings.OLLAMA_URL + self.embedding_model = EMBEDDING_MODEL + + # ========================================================================= + # Embedding Operations + # ========================================================================= + + async def embed_text(self, text: str) -> list[float] | None: + """ + 使用 Ollama 生成文字 embedding + + Args: + text: 要向量化的文字 + + Returns: + 向量 (768 維) 或 None (失敗時) + """ + try: + import httpx + + 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 + + result = response.json() + embedding = result.get("embedding", []) + + if not embedding: + logger.warning("ollama_embedding_empty", text_preview=text[:50]) + return None + + return normalize_vector(embedding) + + except Exception as e: + logger.warning( + "ollama_embedding_error", + error=str(e), + text_preview=text[:50], + ) + return None + + async def embed_playbook(self, playbook: Playbook) -> list[float] | None: + """ + 將 Playbook 向量化 + + 結合症狀模式和修復步驟生成向量 + """ + # 構建文字表示 + text_parts = [] + + # 症狀 + if playbook.symptom_pattern: + sp = playbook.symptom_pattern + if sp.alert_names: + text_parts.append(f"告警: {', '.join(sp.alert_names)}") + if sp.affected_services: + text_parts.append(f"服務: {', '.join(sp.affected_services)}") + if sp.keywords: + text_parts.append(f"關鍵字: {', '.join(sp.keywords)}") + + # 名稱和描述 + text_parts.append(f"名稱: {playbook.name}") + if playbook.description: + text_parts.append(f"描述: {playbook.description}") + + # 修復步驟 + if playbook.repair_steps: + steps_text = "; ".join( + f"{s.sequence}. {s.description}" + for s in playbook.repair_steps[:5] # 最多 5 步 + ) + text_parts.append(f"步驟: {steps_text}") + + text = "\n".join(text_parts) + return await self.embed_text(text) + + async def embed_incident_query( + self, + alert_names: list[str], + affected_services: list[str], + description: str | None = None, + ) -> list[float] | None: + """ + 為 Incident 查詢生成 embedding + + 用於搜尋相似 Playbook + """ + text_parts = [] + + if alert_names: + text_parts.append(f"告警: {', '.join(alert_names)}") + if affected_services: + text_parts.append(f"服務: {', '.join(affected_services)}") + if description: + text_parts.append(f"描述: {description}") + + if not text_parts: + return None + + text = "\n".join(text_parts) + return await self.embed_text(text) + + # ========================================================================= + # Storage Operations + # ========================================================================= + + async def store_playbook_embedding( + 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: + 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 + + 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 + + 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 {} + + # ========================================================================= + # Search Operations + # ========================================================================= + + async def search_similar( + self, + query_embedding: list[float], + top_k: int = 5, + min_similarity: float = 0.5, + ) -> list[PlaybookMatch]: + """ + 向量相似度搜尋 + + 使用餘弦相似度在所有 Playbook 向量中搜尋 + + Args: + query_embedding: 查詢向量 + top_k: 返回前 K 個結果 + min_similarity: 最小相似度閾值 + + Returns: + 排序後的 PlaybookMatch 列表 + """ + # 取得所有 Playbook 向量 + all_embeddings = await self.get_all_playbook_embeddings() + + if not all_embeddings: + logger.debug("search_similar_no_embeddings") + return [] + + # 計算相似度 + similarities: list[tuple[str, float]] = [] + for playbook_id, embedding in all_embeddings.items(): + sim = cosine_similarity(query_embedding, embedding) + if sim >= min_similarity: + similarities.append((playbook_id, sim)) + + # 排序 (降序) + similarities.sort(key=lambda x: x[1], reverse=True) + + # 返回 Top K + results = [ + PlaybookMatch( + playbook_id=pid, + similarity_score=sim, + match_type="vector", + ) + for pid, sim in similarities[:top_k] + ] + + logger.info( + "playbook_vector_search", + total_embeddings=len(all_embeddings), + matches_found=len(results), + top_score=results[0].similarity_score if results else 0, + ) + + return results + + async def search_by_incident( + self, + alert_names: list[str], + affected_services: list[str], + description: str | None = None, + top_k: int = 5, + min_similarity: float = 0.5, + ) -> list[PlaybookMatch]: + """ + 根據 Incident 資訊搜尋相似 Playbook + + Convenience 方法,結合 embed + search + """ + # 生成查詢向量 + query_embedding = await self.embed_incident_query( + alert_names=alert_names, + affected_services=affected_services, + description=description, + ) + + if not query_embedding: + logger.warning( + "search_by_incident_embedding_failed", + alert_names=alert_names, + ) + return [] + + return await self.search_similar( + query_embedding=query_embedding, + top_k=top_k, + min_similarity=min_similarity, + ) + + # ========================================================================= + # Hybrid Search (Vector + Jaccard) + # ========================================================================= + + async def hybrid_search( + self, + symptoms: SymptomPattern, + jaccard_results: list[tuple[str, float]], # (playbook_id, jaccard_score) + top_k: int = 5, + vector_weight: float = 0.6, + jaccard_weight: float = 0.4, + ) -> list[PlaybookMatch]: + """ + 混合搜尋 (向量 + Jaccard) + + 結合向量語意相似度和 Jaccard 精確匹配 + + Args: + symptoms: 症狀模式 + jaccard_results: Jaccard 匹配結果 + top_k: 返回前 K 個 + vector_weight: 向量分數權重 + jaccard_weight: Jaccard 分數權重 + + Returns: + 混合排序後的結果 + """ + # 1. 向量搜尋 + query_embedding = await self.embed_incident_query( + alert_names=symptoms.alert_names, + affected_services=symptoms.affected_services, + description=None, + ) + + vector_scores: dict[str, float] = {} + if query_embedding: + vector_matches = await self.search_similar( + query_embedding=query_embedding, + top_k=top_k * 2, + min_similarity=0.3, + ) + vector_scores = {m.playbook_id: m.similarity_score for m in vector_matches} + + # 2. Jaccard 分數 + jaccard_scores = {pid: score for pid, score in jaccard_results} + + # 3. 合併所有 playbook_id + all_ids = set(vector_scores.keys()) | set(jaccard_scores.keys()) + + # 4. 計算混合分數 + hybrid_results: list[tuple[str, float, str]] = [] + for pid in all_ids: + v_score = vector_scores.get(pid, 0.0) + j_score = jaccard_scores.get(pid, 0.0) + + hybrid_score = (v_score * vector_weight) + (j_score * jaccard_weight) + + # 決定主要匹配類型 + if v_score > 0 and j_score > 0: + match_type = "hybrid" + elif v_score > 0: + match_type = "vector" + else: + match_type = "jaccard" + + hybrid_results.append((pid, hybrid_score, match_type)) + + # 5. 排序並返回 Top K + hybrid_results.sort(key=lambda x: x[1], reverse=True) + + results = [ + PlaybookMatch( + playbook_id=pid, + similarity_score=score, + match_type=match_type, + ) + for pid, score, match_type in hybrid_results[:top_k] + ] + + logger.info( + "playbook_hybrid_search", + vector_count=len(vector_scores), + jaccard_count=len(jaccard_scores), + hybrid_count=len(results), + top_score=results[0].similarity_score if results else 0, + ) + + return results + + # ========================================================================= + # Index Management + # ========================================================================= + + async def index_playbook(self, playbook: Playbook) -> bool: + """ + 為 Playbook 建立向量索引 + + 呼叫時機: Playbook 建立或更新時 + """ + embedding = await self.embed_playbook(playbook) + + if not embedding: + logger.warning( + "playbook_index_embedding_failed", + playbook_id=playbook.playbook_id, + ) + return False + + return await self.store_playbook_embedding( + playbook_id=playbook.playbook_id, + embedding=embedding, + metadata={ + "name": playbook.name, + "status": playbook.status.value if playbook.status else None, + "tags": playbook.tags, + }, + ) + + 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 + + async def reindex_all_playbooks( + self, + playbooks: list[Playbook], + ) -> tuple[int, int]: + """ + 重建所有 Playbook 向量索引 + + Returns: + (成功數, 失敗數) + """ + success = 0 + failed = 0 + + for playbook in playbooks: + if await self.index_playbook(playbook): + success += 1 + else: + failed += 1 + + logger.info( + "playbook_reindex_complete", + success=success, + failed=failed, + total=len(playbooks), + ) + + return success, failed + + +# ============================================================================= +# Singleton +# ============================================================================= + +_rag_service: PlaybookRAGService | None = None + + +def get_playbook_rag_service() -> PlaybookRAGService: + """取得 Playbook RAG 服務 singleton""" + global _rag_service + if _rag_service is None: + _rag_service = PlaybookRAGService() + return _rag_service diff --git a/apps/api/src/services/playbook_service.py b/apps/api/src/services/playbook_service.py index e3fed15b4..8a9e3b0dc 100644 --- a/apps/api/src/services/playbook_service.py +++ b/apps/api/src/services/playbook_service.py @@ -4,6 +4,7 @@ Playbook Service - #7 Playbook 萃取 Playbook 業務邏輯層 Phase 7.3: Service 實作 +Phase 3 ADR-030: RAG 向量搜尋整合 建立時間: 2026-03-26 (台北時區) 建立者: Claude Code (#7 Playbook 萃取) @@ -30,6 +31,7 @@ from src.models.playbook import ( ) from src.repositories.interfaces import IPlaybookRepository from src.repositories.playbook_repository import get_playbook_repository +from src.services.playbook_rag import get_playbook_rag_service from src.utils.timezone import now_taipei logger = structlog.get_logger(__name__) @@ -78,12 +80,14 @@ class PlaybookService: 職責: - 從 Incident 萃取 Playbook - - 提供 Playbook 推薦 + - 提供 Playbook 推薦 (混合搜尋: Jaccard + RAG) - 管理 Playbook 生命週期 + - 維護向量索引 """ def __init__(self, repository: IPlaybookRepository | None = None): self._repository = repository or get_playbook_repository() + self._rag_service = get_playbook_rag_service() # === Core Operations === @@ -161,6 +165,10 @@ class PlaybookService: # 7. 儲存 playbook = await self._repository.create(playbook) + # 8. ADR-030 Phase 3: 建立向量索引 (非阻塞,失敗不影響主流程) + import asyncio + asyncio.create_task(self._index_playbook_async(playbook)) + logger.info( "playbook_extracted", playbook_id=playbook.playbook_id, @@ -171,33 +179,94 @@ class PlaybookService: return playbook + async def _index_playbook_async(self, playbook: Playbook) -> None: + """非同步建立 Playbook 向量索引 (ADR-030 Phase 3)""" + try: + success = await self._rag_service.index_playbook(playbook) + if success: + logger.debug( + "playbook_indexed", + playbook_id=playbook.playbook_id, + ) + except Exception as e: + logger.warning( + "playbook_index_failed", + playbook_id=playbook.playbook_id, + error=str(e), + ) + async def get_recommendations( self, symptoms: SymptomPattern, top_k: int = 3, + use_rag: bool = True, ) -> list[PlaybookRecommendation]: """ 取得 Playbook 推薦 - 策略: - 1. 從 Repository 找相似症狀的 Playbook - 2. 按 similarity_score * success_rate 排序 - 3. 返回 Top K 推薦 + ADR-030 Phase 3 策略: + 1. Jaccard 精確匹配 (Repository) + 2. RAG 向量語意搜尋 (可選) + 3. 混合排序 (Jaccard 40% + Vector 60%) + 4. 按 similarity_score * success_rate 排序 """ - # 查詢相似 Playbook + # Step 1: Jaccard 精確匹配 similar_playbooks = await self._repository.find_by_symptoms( symptoms=symptoms, top_k=top_k * 2, # 多取一些用於後續過濾 min_similarity=0.4, ) - if not similar_playbooks: + jaccard_results = [(pb.playbook_id, sim) for pb, sim in similar_playbooks] + playbook_map = {pb.playbook_id: pb for pb, _ in similar_playbooks} + + # Step 2: RAG 混合搜尋 (如果啟用) + if use_rag and symptoms.alert_names: + try: + hybrid_matches = await self._rag_service.hybrid_search( + symptoms=symptoms, + jaccard_results=jaccard_results, + top_k=top_k * 2, + vector_weight=0.6, + jaccard_weight=0.4, + ) + + # 補充 playbook_map (RAG 可能找到 Jaccard 沒找到的) + for match in hybrid_matches: + if match.playbook_id not in playbook_map: + pb = await self._repository.get_by_id(match.playbook_id) + if pb: + playbook_map[match.playbook_id] = pb + + # 使用混合結果 + final_results = [ + (playbook_map[m.playbook_id], m.similarity_score) + for m in hybrid_matches + if m.playbook_id in playbook_map + ] + + logger.info( + "playbook_recommendation_hybrid", + jaccard_count=len(jaccard_results), + hybrid_count=len(final_results), + ) + except Exception as e: + # RAG 失敗時 fallback 到純 Jaccard + logger.warning( + "playbook_rag_fallback", + error=str(e), + ) + final_results = similar_playbooks + else: + final_results = similar_playbooks + + if not final_results: return [] - # 建立推薦列表 + # Step 3: 建立推薦列表 recommendations: list[PlaybookRecommendation] = [] - for playbook, similarity in similar_playbooks: + for playbook, similarity in final_results: # 找出匹配的症狀 matched_symptoms = self._find_matched_symptoms(symptoms, playbook.symptom_pattern) @@ -217,7 +286,7 @@ class PlaybookService: ) ) - # 按綜合分數排序 + # Step 4: 按綜合分數排序 (similarity * success_rate) recommendations.sort( key=lambda r: r.similarity_score * (0.5 + 0.5 * r.playbook.success_rate), reverse=True,