diff --git a/apps/api/src/services/ai_router.py b/apps/api/src/services/ai_router.py new file mode 100644 index 000000000..e5b9eadb7 --- /dev/null +++ b/apps/api/src/services/ai_router.py @@ -0,0 +1,185 @@ +""" +AI Router - Phase 13.3 #87 +========================== +動態模型選擇器,整合意圖分類和複雜度評分 + +目標: 根據請求特性自動選擇最適模型 +策略: Intent + Complexity → Model Selection + +Phase 13.3 (2026-03-26): 初始實作 +""" + +from dataclasses import dataclass + +import structlog + +from src.services.complexity_scorer import ( + ComplexityScore, + get_complexity_scorer, +) +from src.services.intent_classifier import ( + IntentType, + get_intent_classifier, +) + +logger = structlog.get_logger(__name__) + + +@dataclass +class RoutingDecision: + """路由決策結果""" + + model: str # 選擇的模型 + intent: IntentType # 意圖分類 + complexity: ComplexityScore # 複雜度評分 + reason: str # 選擇原因 + fallback_models: list[str] # 備援模型列表 + + +class AIRouter: + """ + AI 路由器 + + 整合 IntentClassifier 和 ComplexityScorer, + 動態選擇最適合的模型。 + + 路由策略: + 1. 意圖優先覆寫 (某些意圖強制使用特定模型) + 2. 複雜度導向選擇 + 3. 成本/延遲平衡 + """ + + # 意圖強制覆寫 + INTENT_OVERRIDES: dict[IntentType, str | None] = { + IntentType.CODE_REVIEW: "qwen2.5:7b-instruct", # 程式碼審查需要強模型 + IntentType.DEPLOYMENT: None, # 不覆寫,依複雜度 + IntentType.ALERT_TRIAGE: None, + IntentType.QUERY: "llama3.2:3b", # 查詢用快速模型 + IntentType.MAINTENANCE: None, + IntentType.UNKNOWN: None, + } + + # Fallback 順序 + FALLBACK_ORDER = [ + "qwen2.5:7b-instruct", # 本地主力 + "llama3.2:3b", # 本地備援 + "gemini", # 雲端備援 + "claude", # 最終備援 + ] + + def __init__(self): + self._intent_classifier = get_intent_classifier() + self._complexity_scorer = get_complexity_scorer() + + async def route( + self, + text: str, + context: dict | None = None, + ) -> RoutingDecision: + """ + 路由請求到最適模型 + + Args: + text: 用戶輸入或告警內容 + context: 額外上下文 (服務、指標等) + + Returns: + RoutingDecision: 路由決策 + """ + context = context or {} + + # Step 1: 意圖分類 + intent = await self._intent_classifier.classify(text) + + # Step 2: 複雜度評分 + complexity = self._complexity_scorer.score(context) + + # Step 3: 模型選擇 + model, reason = self._select_model(intent, complexity) + + # Step 4: 建立 Fallback 列表 + fallbacks = self._build_fallback_list(model) + + decision = RoutingDecision( + model=model, + intent=intent, + complexity=complexity, + reason=reason, + fallback_models=fallbacks, + ) + + logger.info( + "ai_routing_decision", + model=model, + intent=intent.value, + complexity_score=complexity.score, + reason=reason, + ) + + return decision + + def _select_model( + self, + intent: IntentType, + complexity: ComplexityScore, + ) -> tuple[str, str]: + """ + 選擇模型 + + Returns: + (model_name, reason) + """ + # 檢查意圖覆寫 + override = self.INTENT_OVERRIDES.get(intent) + if override: + return override, f"意圖 {intent.value} 強制使用 {override}" + + # 依複雜度選擇 + model = complexity.recommended_model + reason = f"複雜度 {complexity.score}/5 → {model}" + + # 特殊情況調整 + if intent == IntentType.ALERT_TRIAGE and complexity.score >= 4: + # 高複雜度告警優先用雲端 + model = "gemini" + reason = f"高複雜度告警 (score={complexity.score}) → 使用雲端模型" + + return model, reason + + def _build_fallback_list(self, selected_model: str) -> list[str]: + """建立 Fallback 列表 (排除已選模型)""" + fallbacks = [m for m in self.FALLBACK_ORDER if m != selected_model] + return fallbacks + + def route_sync( + self, + text: str, + context: dict | None = None, + ) -> RoutingDecision: + """同步版本 (僅關鍵字匹配)""" + context = context or {} + + intent = self._intent_classifier.classify_sync(text) + complexity = self._complexity_scorer.score(context) + model, reason = self._select_model(intent, complexity) + fallbacks = self._build_fallback_list(model) + + return RoutingDecision( + model=model, + intent=intent, + complexity=complexity, + reason=reason, + fallback_models=fallbacks, + ) + + +# 單例 +_router: AIRouter | None = None + + +def get_ai_router() -> AIRouter: + """取得 AIRouter 單例""" + global _router + if _router is None: + _router = AIRouter() + return _router diff --git a/apps/api/src/services/complexity_scorer.py b/apps/api/src/services/complexity_scorer.py new file mode 100644 index 000000000..4ca2012db --- /dev/null +++ b/apps/api/src/services/complexity_scorer.py @@ -0,0 +1,164 @@ +""" +Complexity Scorer - Phase 13.3 #86 +=================================== +複雜度評分,用於智能路由模型選擇 + +目標: < 10ms 延遲 (純規則引擎) +策略: 基於特徵提取的加權評分 + +Phase 13.3 (2026-03-26): 初始實作 +""" + +from dataclasses import dataclass, field + +import structlog + +logger = structlog.get_logger(__name__) + + +@dataclass +class ComplexityScore: + """複雜度評分結果""" + + score: int # 1-5 (1=簡單, 5=極複雜) + features: dict[str, int] = field(default_factory=dict) + recommended_model: str = "qwen2.5:7b-instruct" + reasoning: str = "" + + +# 模型映射 (依複雜度) +MODEL_BY_COMPLEXITY = { + 1: "llama3.2:3b", # 簡單任務,快速回應 + 2: "qwen2.5:7b-instruct", # 中等任務 + 3: "qwen2.5:7b-instruct", # 複雜任務 + 4: "gemini", # 需要雲端能力 + 5: "claude", # 極複雜,需要最強模型 +} + + +class ComplexityScorer: + """ + 複雜度評分器 + + 基於規則的複雜度評估,無 LLM 依賴,確保 < 10ms + + 評分維度: + 1. 服務數量 (affected_services) + 2. 指標數量 (metrics) + 3. 是否需要程式碼分析 (requires_code_analysis) + 4. 是否跨系統 (cross_system) + 5. 是否有歷史關聯 (has_history) + 6. 嚴重程度 (severity) + """ + + # 權重配置 + WEIGHTS = { + "service_count": 0.5, # 每增加一個服務 +0.5 + "metric_count": 0.3, # 每增加一個指標 +0.3 + "code_analysis": 1.5, # 需要代碼分析 +1.5 + "cross_system": 1.0, # 跨系統 +1.0 + "has_history": -0.5, # 有歷史案例 -0.5 (降低複雜度) + "critical_severity": 1.0, # CRITICAL 告警 +1.0 + } + + def score(self, context: dict) -> ComplexityScore: + """ + 計算複雜度分數 + + Args: + context: 上下文資訊,包含: + - affected_services: list[str] + - metrics: list[str] + - requires_code_analysis: bool + - cross_system: bool + - has_history: bool + - severity: str + + Returns: + ComplexityScore: 評分結果 + """ + raw_score = 1.0 # 基準分 + features: dict[str, int] = {} + reasons: list[str] = [] + + # 特徵 1: 服務數量 + services = context.get("affected_services", []) + service_count = len(services) + if service_count > 1: + delta = (service_count - 1) * self.WEIGHTS["service_count"] + raw_score += delta + features["service_count"] = service_count + reasons.append(f"涉及 {service_count} 個服務") + + # 特徵 2: 指標數量 + metrics = context.get("metrics", []) + metric_count = len(metrics) + if metric_count > 2: + delta = (metric_count - 2) * self.WEIGHTS["metric_count"] + raw_score += delta + features["metric_count"] = metric_count + reasons.append(f"涉及 {metric_count} 個指標") + + # 特徵 3: 是否需要程式碼分析 + if context.get("requires_code_analysis", False): + raw_score += self.WEIGHTS["code_analysis"] + features["code_analysis"] = 1 + reasons.append("需要程式碼分析") + + # 特徵 4: 是否跨系統 + if context.get("cross_system", False): + raw_score += self.WEIGHTS["cross_system"] + features["cross_system"] = 1 + reasons.append("跨系統問題") + + # 特徵 5: 是否有歷史關聯 + if context.get("has_history", False): + raw_score += self.WEIGHTS["has_history"] # 負數,降低複雜度 + features["has_history"] = 1 + reasons.append("有歷史案例參考") + + # 特徵 6: 嚴重程度 + severity = context.get("severity", "").upper() + if severity == "CRITICAL": + raw_score += self.WEIGHTS["critical_severity"] + features["severity"] = 4 + reasons.append("CRITICAL 嚴重程度") + elif severity == "HIGH": + raw_score += 0.5 + features["severity"] = 3 + + # 正規化到 1-5 + final_score = max(1, min(5, round(raw_score))) + + # 選擇推薦模型 + recommended_model = MODEL_BY_COMPLEXITY.get( + final_score, "qwen2.5:7b-instruct" + ) + + result = ComplexityScore( + score=final_score, + features=features, + recommended_model=recommended_model, + reasoning="; ".join(reasons) if reasons else "基本複雜度", + ) + + logger.debug( + "complexity_scored", + score=final_score, + features=features, + model=recommended_model, + ) + + return result + + +# 單例 +_scorer: ComplexityScorer | None = None + + +def get_complexity_scorer() -> ComplexityScorer: + """取得 ComplexityScorer 單例""" + global _scorer + if _scorer is None: + _scorer = ComplexityScorer() + return _scorer diff --git a/apps/api/src/services/intent_classifier.py b/apps/api/src/services/intent_classifier.py new file mode 100644 index 000000000..064646be5 --- /dev/null +++ b/apps/api/src/services/intent_classifier.py @@ -0,0 +1,149 @@ +""" +Intent Classifier - Phase 13.3 #85 +=================================== +快速意圖分類,用於智能路由 + +目標: < 100ms 延遲 +策略: 關鍵字優先 → 小模型備援 + +Phase 13.3 (2026-03-26): 初始實作 +""" + +import re +from enum import Enum + +import structlog + +from src.core.config import settings + +logger = structlog.get_logger(__name__) + + +class IntentType(Enum): + """意圖類型""" + + ALERT_TRIAGE = "alert_triage" # 告警分流/處理 + DEPLOYMENT = "deployment" # 部署操作 (kubectl, rollout) + QUERY = "query" # 資訊查詢 (狀態, 日誌) + MAINTENANCE = "maintenance" # 維運操作 (重啟, 擴容) + CODE_REVIEW = "code_review" # 程式碼審查 + UNKNOWN = "unknown" + + +# 關鍵字映射 (優先匹配,0ms) +INTENT_KEYWORDS: dict[IntentType, list[str]] = { + IntentType.ALERT_TRIAGE: [ + "alert", "告警", "警報", "異常", "error", "critical", "warning", + "高負載", "high cpu", "memory", "oom", "crash", "down", + ], + IntentType.DEPLOYMENT: [ + "deploy", "部署", "rollout", "kubectl apply", "helm", "release", + "版本", "upgrade", "更新", "上線", + ], + IntentType.QUERY: [ + "查詢", "狀態", "status", "describe", "get", "list", "日誌", "log", + "哪個", "什麼", "how many", "多少", + ], + IntentType.MAINTENANCE: [ + "restart", "重啟", "scale", "擴容", "縮容", "rollback", "回滾", + "維護", "maintenance", "patch", "修補", + ], + IntentType.CODE_REVIEW: [ + "review", "審查", "pr", "pull request", "commit", "diff", + "程式碼", "code", "merge", + ], +} + + +class IntentClassifier: + """ + 意圖分類器 + + 使用兩階段分類策略: + 1. 關鍵字快速匹配 (0ms) + 2. 小模型 LLM 分類 (< 100ms) - 備援 + """ + + # 小模型,低延遲 + MODEL = "qwen2.5:1b" + + def __init__(self): + self._keyword_cache: dict[str, IntentType] = {} + + async def classify(self, text: str) -> IntentType: + """ + 分類意圖 + + Args: + text: 用戶輸入或告警內容 + + Returns: + IntentType: 分類結果 + """ + text_lower = text.lower() + + # 階段 1: 關鍵字快速匹配 (0ms) + intent = self._keyword_match(text_lower) + if intent != IntentType.UNKNOWN: + logger.debug( + "intent_classified_by_keyword", + intent=intent.value, + text_preview=text[:50], + ) + return intent + + # 階段 2: LLM 分類 (< 100ms) + # 目前先用關鍵字,LLM 整合待 Qwen 1B 部署 + logger.debug( + "intent_fallback_to_unknown", + text_preview=text[:50], + ) + return IntentType.UNKNOWN + + def _keyword_match(self, text: str) -> IntentType: + """關鍵字匹配""" + # 檢查快取 + cache_key = text[:100] + if cache_key in self._keyword_cache: + return self._keyword_cache[cache_key] + + # 計算每個意圖的匹配分數 + scores: dict[IntentType, int] = {} + + for intent, keywords in INTENT_KEYWORDS.items(): + score = 0 + for keyword in keywords: + if keyword in text: + score += 1 + # 完整匹配加分 + if re.search(rf"\b{re.escape(keyword)}\b", text): + score += 1 + if score > 0: + scores[intent] = score + + if not scores: + return IntentType.UNKNOWN + + # 選擇最高分 + best_intent = max(scores, key=lambda k: scores[k]) + + # 快取結果 + self._keyword_cache[cache_key] = best_intent + + return best_intent + + def classify_sync(self, text: str) -> IntentType: + """同步版本 (僅關鍵字匹配)""" + return self._keyword_match(text.lower()) + + +# 單例 +_classifier: IntentClassifier | None = None + + +def get_intent_classifier() -> IntentClassifier: + """取得 IntentClassifier 單例""" + global _classifier + if _classifier is None: + _classifier = IntentClassifier() + return _classifier diff --git a/apps/api/tests/test_smart_router.py b/apps/api/tests/test_smart_router.py new file mode 100644 index 000000000..c57cebef8 --- /dev/null +++ b/apps/api/tests/test_smart_router.py @@ -0,0 +1,195 @@ +""" +Smart Router Tests - Phase 13.3 +=============================== +測試意圖分類、複雜度評分、AI 路由 +""" + +import pytest + +from src.services.intent_classifier import ( + IntentClassifier, + IntentType, + get_intent_classifier, +) +from src.services.complexity_scorer import ( + ComplexityScorer, + get_complexity_scorer, +) +from src.services.ai_router import ( + AIRouter, + get_ai_router, +) + + +class TestIntentClassifier: + """測試意圖分類器""" + + def test_alert_keywords(self): + """測試告警關鍵字匹配""" + classifier = IntentClassifier() + + # 中文告警 + assert classifier.classify_sync("高負載警報") == IntentType.ALERT_TRIAGE + assert classifier.classify_sync("CPU 異常告警") == IntentType.ALERT_TRIAGE + assert classifier.classify_sync("OOM error detected") == IntentType.ALERT_TRIAGE + + def test_deployment_keywords(self): + """測試部署關鍵字匹配""" + classifier = IntentClassifier() + + assert classifier.classify_sync("部署新版本") == IntentType.DEPLOYMENT + assert classifier.classify_sync("kubectl apply -f manifest.yaml") == IntentType.DEPLOYMENT + assert classifier.classify_sync("rollout deployment api") == IntentType.DEPLOYMENT + + def test_query_keywords(self): + """測試查詢關鍵字匹配""" + classifier = IntentClassifier() + + assert classifier.classify_sync("查詢 Pod 狀態") == IntentType.QUERY + assert classifier.classify_sync("kubectl get pods") == IntentType.QUERY + assert classifier.classify_sync("現在有多少 replicas") == IntentType.QUERY + + def test_maintenance_keywords(self): + """測試維運關鍵字匹配""" + classifier = IntentClassifier() + + assert classifier.classify_sync("重啟服務") == IntentType.MAINTENANCE + assert classifier.classify_sync("scale deployment to 5") == IntentType.MAINTENANCE + assert classifier.classify_sync("回滾到上一版") == IntentType.MAINTENANCE + + def test_code_review_keywords(self): + """測試程式碼審查關鍵字匹配""" + classifier = IntentClassifier() + + assert classifier.classify_sync("review this PR") == IntentType.CODE_REVIEW + assert classifier.classify_sync("審查這個 commit") == IntentType.CODE_REVIEW + + def test_unknown_intent(self): + """測試未知意圖""" + classifier = IntentClassifier() + + assert classifier.classify_sync("hello world") == IntentType.UNKNOWN + assert classifier.classify_sync("今天天氣如何") == IntentType.UNKNOWN + + +class TestComplexityScorer: + """測試複雜度評分器""" + + def test_simple_context(self): + """測試簡單上下文""" + scorer = ComplexityScorer() + + result = scorer.score({}) + assert result.score == 1 + assert result.recommended_model == "llama3.2:3b" + + def test_multi_service_context(self): + """測試多服務上下文""" + scorer = ComplexityScorer() + + result = scorer.score({ + "affected_services": ["api", "worker", "redis"], + }) + assert result.score >= 2 + assert "service_count" in result.features + + def test_code_analysis_context(self): + """測試需要程式碼分析""" + scorer = ComplexityScorer() + + result = scorer.score({ + "requires_code_analysis": True, + }) + assert result.score >= 2 + assert result.features.get("code_analysis") == 1 + + def test_critical_severity(self): + """測試 CRITICAL 嚴重程度""" + scorer = ComplexityScorer() + + result = scorer.score({ + "severity": "CRITICAL", + }) + assert result.score >= 2 + assert result.features.get("severity") == 4 + + def test_complex_context(self): + """測試複雜上下文""" + scorer = ComplexityScorer() + + result = scorer.score({ + "affected_services": ["api", "worker", "redis", "postgres"], + "metrics": ["cpu", "memory", "latency", "error_rate", "rps"], + "cross_system": True, + "severity": "CRITICAL", + }) + assert result.score >= 4 + # 複雜情況應該用雲端模型 + assert result.recommended_model in ["gemini", "claude"] + + +class TestAIRouter: + """測試 AI 路由器""" + + def test_query_routes_to_fast_model(self): + """測試查詢路由到快速模型""" + router = AIRouter() + + decision = router.route_sync("查詢 Pod 狀態", {}) + assert decision.model == "llama3.2:3b" + assert decision.intent == IntentType.QUERY + + def test_code_review_routes_to_strong_model(self): + """測試程式碼審查路由到強模型""" + router = AIRouter() + + decision = router.route_sync("review this PR", {}) + assert decision.model == "qwen2.5:7b-instruct" + assert decision.intent == IntentType.CODE_REVIEW + + def test_complex_alert_routes_to_cloud(self): + """測試複雜告警路由到雲端""" + router = AIRouter() + + decision = router.route_sync("高負載告警", { + "affected_services": ["api", "worker", "redis", "postgres"], + "metrics": ["cpu", "memory", "latency", "error_rate"], + "cross_system": True, + "severity": "CRITICAL", + }) + assert decision.intent == IntentType.ALERT_TRIAGE + assert decision.complexity.score >= 4 + # 高複雜度告警應該用雲端 + assert decision.model in ["gemini", "claude", "qwen2.5:7b-instruct"] + + def test_fallback_list(self): + """測試 Fallback 列表""" + router = AIRouter() + + decision = router.route_sync("查詢 Pod 狀態", {}) + # Fallback 不應包含已選模型 + assert decision.model not in decision.fallback_models + # 應該有備援 + assert len(decision.fallback_models) >= 2 + + +class TestSingletons: + """測試單例""" + + def test_intent_classifier_singleton(self): + """測試 IntentClassifier 單例""" + c1 = get_intent_classifier() + c2 = get_intent_classifier() + assert c1 is c2 + + def test_complexity_scorer_singleton(self): + """測試 ComplexityScorer 單例""" + s1 = get_complexity_scorer() + s2 = get_complexity_scorer() + assert s1 is s2 + + def test_ai_router_singleton(self): + """測試 AIRouter 單例""" + r1 = get_ai_router() + r2 = get_ai_router() + assert r1 is r2