224 lines
7.4 KiB
Python
224 lines
7.4 KiB
Python
"""
|
||
Drift Interpreter - Phase 25 P2 Config Drift Detection
|
||
=======================================================
|
||
職責：Nemotron 意圖分析（不生成修復指令）
|
||
只回答「這是人為操作？Hotfix？系統自動變更？」
|
||
|
||
設計邊界（核心原則）:
|
||
- 只輸出意圖分析，不生成 kubectl 或 git 指令
|
||
- 確定性修復由 DriftRemediator 負責
|
||
- Nemotron 超時 → UNKNOWN，不阻塞主流程
|
||
|
||
版本: v1.0
|
||
建立: 2026-04-04 (台北時區)
|
||
建立者: ogt (首席架構師設計) + Claude Code (實作)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from typing import TYPE_CHECKING
|
||
|
||
import structlog
|
||
|
||
from src.models.drift import DriftIntent, DriftInterpretation
|
||
|
||
if TYPE_CHECKING:
|
||
from src.models.drift import DriftReport
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
_INTENT_PROMPT_TEMPLATE = """你是 AWOOOI GitOps 守門員，請分析以下 K8s 配置漂移的意圖。
|
||
|
||
## 漂移詳情
|
||
{diff_summary}
|
||
|
||
## 任務
|
||
判斷這次漂移最可能的原因：
|
||
- emergency_hotfix: 繞過 CI 的緊急修補（image tag 改變但無對應 Git commit）
|
||
- human_error: 誤操作（非預期的隨機欄位改變）
|
||
- automated_change: 系統自動變更（HPA replicas, 系統注入的 annotation 等）
|
||
- unknown: 無法判斷
|
||
|
||
請以 JSON 回應：
|
||
{{
|
||
"intent": "emergency_hotfix|human_error|automated_change|unknown",
|
||
"explanation": "用繁體中文解釋你的判斷理由（一句話）",
|
||
"risk": "HIGH|MEDIUM|LOW",
|
||
"confidence": 0.0到1.0之間的數字
|
||
}}
|
||
|
||
只輸出 JSON，不要任何額外說明。
|
||
"""
|
||
|
||
|
||
def _strip_think_blocks(text: str) -> str:
|
||
"""移除 qwen/deepseek 類模型常見的 <think> 推理段。"""
|
||
return re.sub(r"<think>[\s\S]*?</think>", "", text, flags=re.IGNORECASE).strip()
|
||
|
||
|
||
def _extract_first_json_object(text: str) -> dict | None:
|
||
"""
|
||
從 LLM 回應中擷取第一個 JSON object。
|
||
|
||
Ollama qwen3/deepseek 常會在 JSON 前後加 `<think>` 或短句；這些文字不應
|
||
讓 drift intent 直接降級成 UNKNOWN。
|
||
"""
|
||
cleaned = _strip_think_blocks(text)
|
||
|
||
candidates = [cleaned]
|
||
candidates.extend(match.group(1).strip() for match in re.finditer(r"```(?:json)?\s*([\s\S]+?)```", cleaned))
|
||
|
||
start = cleaned.find("{")
|
||
if start >= 0:
|
||
in_string = False
|
||
escaped = False
|
||
depth = 0
|
||
for idx, ch in enumerate(cleaned[start:], start=start):
|
||
if escaped:
|
||
escaped = False
|
||
continue
|
||
if ch == "\\":
|
||
escaped = True
|
||
continue
|
||
if ch == '"':
|
||
in_string = not in_string
|
||
continue
|
||
if in_string:
|
||
continue
|
||
if ch == "{":
|
||
depth += 1
|
||
elif ch == "}":
|
||
depth -= 1
|
||
if depth == 0:
|
||
candidates.append(cleaned[start : idx + 1])
|
||
break
|
||
|
||
for candidate in candidates:
|
||
try:
|
||
data = json.loads(candidate)
|
||
except Exception:
|
||
continue
|
||
if isinstance(data, dict):
|
||
return data
|
||
return None
|
||
|
||
|
||
class NemotronDriftInterpreter:
|
||
"""
|
||
使用 Nemotron 分析漂移意圖
|
||
|
||
職責邊界：
|
||
✅ 輸出意圖分析
|
||
❌ 不生成修復指令
|
||
❌ 不直接呼叫 kubectl 或 git
|
||
"""
|
||
|
||
async def analyze(self, report: DriftReport) -> DriftInterpretation:
|
||
"""
|
||
分析漂移意圖
|
||
|
||
Args:
|
||
report: 已分類的 DriftReport
|
||
|
||
Returns:
|
||
DriftInterpretation（超時或失敗時回傳 UNKNOWN）
|
||
"""
|
||
if not report.items or (report.high_count == 0 and report.medium_count == 0):
|
||
return DriftInterpretation(
|
||
intent=DriftIntent.UNKNOWN,
|
||
explanation="無顯著漂移，不需要意圖分析",
|
||
confidence=1.0,
|
||
)
|
||
|
||
diff_text = self._format_diff_for_prompt(report)
|
||
prompt = _INTENT_PROMPT_TEMPLATE.format(diff_summary=diff_text)
|
||
|
||
result = await self._call_nemotron(prompt)
|
||
return result
|
||
|
||
def _format_diff_for_prompt(self, report: DriftReport) -> str:
|
||
"""格式化 diff 給 Nemotron 分析用"""
|
||
lines = []
|
||
for item in report.items[:10]: # 最多 10 項避免 token 過多
|
||
if item.is_allowlisted:
|
||
continue
|
||
lines.append(
|
||
f"- {item.resource_kind}/{item.resource_name}: "
|
||
f"{item.field_path} "
|
||
f"Git={str(item.git_value)[:40]} → "
|
||
f"K8s={str(item.actual_value)[:40]}"
|
||
)
|
||
return "\n".join(lines) if lines else "（均為白名單欄位）"
|
||
|
||
async def _call_nemotron(self, prompt: str) -> DriftInterpretation:
|
||
"""
|
||
呼叫 OpenClaw AI Router 進行意圖分析。
|
||
|
||
2026-04-17 ogt + Claude Sonnet 4.6: 改用 OpenClaw 取代直接 Ollama httpx 呼叫
|
||
根因：直接呼叫 Ollama 繞過 AI Router，無 fallback → "All connection attempts failed"
|
||
修復：統一走 openclaw.call()，自動享有 Provider 降級與 fallback 機制
|
||
廢棄：BUG-001 的 httpx 直連繞過法（nvidia_provider 介面已於 7eb8375 穩定）
|
||
"""
|
||
try:
|
||
from src.services.openclaw import get_openclaw
|
||
openclaw = get_openclaw()
|
||
response_text, _provider, success = await openclaw.call(
|
||
prompt,
|
||
alert_context={
|
||
"intent_hint": "config",
|
||
"task_type": "diagnose",
|
||
"enforce_ollama_first": True,
|
||
"allow_gcp_heavy_model": True,
|
||
"target_resource": "config-drift",
|
||
"alert_type": "ConfigDriftInternalScan",
|
||
},
|
||
)
|
||
|
||
if not success or not response_text:
|
||
logger.warning("drift_interpreter_openclaw_failed", provider=_provider)
|
||
return self._unknown_result("AI Router 回傳失敗或空值")
|
||
|
||
return self._parse_response(response_text)
|
||
|
||
except Exception as e:
|
||
logger.warning("drift_interpreter_error", error=str(e))
|
||
return self._unknown_result(str(e))
|
||
|
||
def _parse_response(self, text: str) -> DriftInterpretation:
|
||
"""解析 Nemotron JSON 回應"""
|
||
data = _extract_first_json_object(text)
|
||
if data is None:
|
||
return self._unknown_result("無法解析 JSON")
|
||
|
||
try:
|
||
intent_str = data.get("intent", "unknown")
|
||
intent = DriftIntent(intent_str) if intent_str in DriftIntent._value2member_map_ else DriftIntent.UNKNOWN
|
||
return DriftInterpretation(
|
||
intent=intent,
|
||
explanation=data.get("explanation", ""),
|
||
risk=data.get("risk", "MEDIUM"),
|
||
confidence=float(data.get("confidence", 0.0)),
|
||
)
|
||
except Exception as e:
|
||
return self._unknown_result(f"模型解析失敗: {e}")
|
||
|
||
def _unknown_result(self, reason: str) -> DriftInterpretation:
|
||
return DriftInterpretation(
|
||
intent=DriftIntent.UNKNOWN,
|
||
explanation=f"意圖分析失敗：{reason}",
|
||
risk="MEDIUM",
|
||
confidence=0.0,
|
||
)
|
||
|
||
|
||
_interpreter: NemotronDriftInterpreter | None = None
|
||
|
||
|
||
def get_drift_interpreter() -> NemotronDriftInterpreter:
|
||
global _interpreter
|
||
if _interpreter is None:
|
||
_interpreter = NemotronDriftInterpreter()
|
||
return _interpreter
|