feat(api): Phase 13.3 智能路由 (#85-87)

- IntentClassifier: 意圖分類 (告警/部署/查詢/維運/審查)
- ComplexityScorer: 複雜度評分 (1-5 分)
- AIRouter: 動態模型選擇 (整合 Intent + Complexity)
- 測試: 完整單元測試覆蓋

Phase 13.3 設計: project_phase13_3_smart_router.md

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-26 10:01:04 +08:00
parent 496c569d51
commit 58b4004a18
4 changed files with 693 additions and 0 deletions

View File

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