""" 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