440 lines
19 KiB
Python
440 lines
19 KiB
Python
"""
|
||
AWOOOI AIOps Phase 2 — Solver Agent（軍師）
|
||
===========================================
|
||
職責：對每個根因假設產修復方案
|
||
|
||
輸入：DiagnosisReport（來自 Diagnostician）
|
||
輸出：ActionPlan（候選動作，含 blast_radius + rollback_cost + confidence）
|
||
|
||
設計原則：
|
||
1. 每個 Hypothesis 至少產 1 個 CandidateAction
|
||
2. blast_radius 評分影響 Reviewer 的審查嚴格度
|
||
3. blast_radius > 50 → Reviewer 必須 request_revision
|
||
4. 熔斷降級：LLM 失敗 → rule-based mock（基於 category 推 RESTART 為兜底動作）
|
||
5. Solver 不直接觸碰執行層（Coordinator 的工作）
|
||
|
||
ADR-082: Phase 2 多 Agent 協作
|
||
2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 2 初始建立
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import time
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
from src.agents.base import BaseAgent
|
||
from src.agents.protocol import (
|
||
ActionPlan,
|
||
AgentRole,
|
||
AgentVote,
|
||
CandidateAction,
|
||
DiagnosisReport,
|
||
)
|
||
from src.services.sanitization_service import sanitize
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# Phase 2 單步 LLM timeout（保留 Critic/Coordinator 的全局預算）
|
||
PHASE2_STEP_TIMEOUT_SEC = 20.0
|
||
|
||
|
||
class SolverAgent(BaseAgent):
|
||
"""
|
||
Solver Agent — 修復方案軍師
|
||
|
||
Usage:
|
||
agent = SolverAgent()
|
||
plan = await agent.run(diagnosis_report)
|
||
"""
|
||
|
||
AGENT_NAME = AgentRole.SOLVER.value
|
||
AGENT_DESCRIPTION = "Remediation plan specialist. Produces candidate actions with blast radius scoring."
|
||
|
||
async def run(
|
||
self,
|
||
diagnosis: DiagnosisReport,
|
||
timeout_sec: float = 0.0, # noqa: ARG002 — 已廢棄，保留簽名相容性
|
||
) -> ActionPlan:
|
||
"""
|
||
根據診斷報告產出修復計畫。
|
||
|
||
Args:
|
||
diagnosis: Diagnostician 輸出
|
||
timeout_sec: 已廢棄 (2026-04-16 ogt) — LLM 等完整回應，真實異常才降級
|
||
|
||
Returns:
|
||
ActionPlan（真實異常時 degraded=True）
|
||
"""
|
||
start_ms = int(time.monotonic() * 1000)
|
||
|
||
# 若 Diagnostician 已棄權，Solver 也應棄權（無論降級假設是否存在）
|
||
# Gate 2: 原條件 `and not diagnosis.hypotheses` 誤放行降級的 confidence=0.2 假設
|
||
if diagnosis.vote == AgentVote.ABSTAIN:
|
||
return ActionPlan(
|
||
candidates=[],
|
||
diagnosis_report=diagnosis,
|
||
latency_ms=0,
|
||
vote=AgentVote.ABSTAIN,
|
||
degraded=diagnosis.degraded,
|
||
)
|
||
|
||
try:
|
||
plan = await self._solve(diagnosis)
|
||
plan.latency_ms = int(time.monotonic() * 1000) - start_ms
|
||
logger.info(
|
||
"solver_done",
|
||
candidates=len(plan.candidates),
|
||
vote=plan.vote,
|
||
latency_ms=plan.latency_ms,
|
||
)
|
||
return plan
|
||
|
||
except Exception:
|
||
latency = int(time.monotonic() * 1000) - start_ms
|
||
logger.exception("solver_error")
|
||
return self._degraded_plan(diagnosis, latency, "error")
|
||
|
||
async def _solve(self, diagnosis: DiagnosisReport) -> ActionPlan:
|
||
"""核心 LLM 推理邏輯。"""
|
||
top = diagnosis.top_hypothesis
|
||
if not top:
|
||
return ActionPlan(
|
||
candidates=[],
|
||
diagnosis_report=diagnosis,
|
||
latency_ms=0,
|
||
vote=AgentVote.ABSTAIN,
|
||
)
|
||
|
||
# 2026-04-17 ogt + Claude Sonnet 4.6 (Checkpoint-2 環境感知):
|
||
# 根因：LLM 在無叢集上下文時「盲猜」資源名稱 → awooiii-api（三個 i）→ K8s not found
|
||
# 修復：生成指令前先拉取實際 Deployment 清單，注入 prompt 讓 LLM 對齊真實名稱
|
||
# 失敗無害：kubectl 超時或拒絕 → _k8s_inventory 為空 → prompt 仍正常但無鎖定效果
|
||
_k8s_inventory = await _fetch_k8s_inventory(namespace="awoooi-prod")
|
||
|
||
prompt = self._build_prompt({
|
||
"hypothesis": top.description,
|
||
"category": top.category,
|
||
"confidence": top.confidence,
|
||
"k8s_inventory": _k8s_inventory,
|
||
})
|
||
|
||
# 2026-04-16 ogt + Claude Sonnet 4.6: 傳遞 hypothesis 結構化資料給 OPENCLAW_NEMO
|
||
# 根因：原本 call(prompt) 不傳 context → nemo fallback 把 prompt[:500]（系統說明）
|
||
# 當 signal description → LLM 回傳「設計修復方案的軍師 Agent」垃圾
|
||
# 修復：把 top hypothesis description 放進 alert_context.signals 讓 nemo 看到真實診斷
|
||
_hypothesis_text = (top.description or "（待診斷）")[:800]
|
||
alert_context = {
|
||
"incident_id": diagnosis.evidence_snapshot_id or "UNKNOWN",
|
||
"severity": "P3",
|
||
"signals": [{"alert_name": "diagnosis_hypothesis", "description": _hypothesis_text}],
|
||
"affected_services": [],
|
||
"intent_hint": "diagnose",
|
||
}
|
||
|
||
from src.services.openclaw import get_openclaw
|
||
openclaw = get_openclaw()
|
||
try:
|
||
response_text, _provider, success = await asyncio.wait_for(
|
||
openclaw.call(prompt, alert_context=alert_context),
|
||
timeout=PHASE2_STEP_TIMEOUT_SEC,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning(
|
||
"solver_step_timeout",
|
||
snapshot_id=diagnosis.evidence_snapshot_id,
|
||
timeout_sec=PHASE2_STEP_TIMEOUT_SEC,
|
||
)
|
||
return self._degraded_plan(diagnosis, 0, "step_timeout")
|
||
|
||
if not success or not response_text:
|
||
return self._degraded_plan(diagnosis, 0, "llm_failed")
|
||
|
||
parsed = self._parse_response(sanitize(response_text, "solver_output"))
|
||
candidates = _extract_candidates(parsed)
|
||
|
||
if not candidates:
|
||
return self._degraded_plan(diagnosis, 0, "no_candidates")
|
||
|
||
return ActionPlan(
|
||
candidates=candidates,
|
||
diagnosis_report=diagnosis,
|
||
latency_ms=0,
|
||
vote=AgentVote.APPROVE,
|
||
)
|
||
|
||
def _build_prompt(self, context: dict[str, Any]) -> str:
|
||
# 2026-04-17 ogt + Claude Sonnet 4.6: 修復 Solver action 格式問題
|
||
# 根因：舊 prompt action 範例為 "restart_service:awoooi-api"（自訂格式）
|
||
# LLM 模仿範例輸出自然語言描述，而非 kubectl 命令
|
||
# → auto_approve Condition 1c 拒絕（無 kubectl 關鍵字）
|
||
# → blast_radius_calculator 永遠不被調用（fill rate = 0%）
|
||
# 修復：要求 action 必須是真實 kubectl 命令，並提供正確範例
|
||
# 2026-04-17 ogt + Claude Sonnet 4.6 (Checkpoint-2): 注入 K8s 實際 Deployment 清單
|
||
# LLM 必須從此清單選擇資源名稱，不可自行編造
|
||
_inventory = context.get("k8s_inventory", "")
|
||
_inventory_section = (
|
||
f"\n🔒 叢集實際 Deployment 清單（awoooi-prod）— 必須從此清單選擇資源名稱：\n{_inventory}\n"
|
||
if _inventory
|
||
else "\n⚠️ 無法取得叢集清單，請謹慎填寫資源名稱。\n"
|
||
)
|
||
return f"""你是 AWOOOI SRE 系統的軍師 Agent，專職修復方案設計。
|
||
|
||
根因假設：{context.get("hypothesis", "")}
|
||
告警類別：{context.get("category", "")}
|
||
診斷信心：{context.get("confidence", 0.0):.0%}
|
||
{_inventory_section}
|
||
你的工作：為此根因提出 1-3 個修復候選方案。
|
||
每個方案必須評估：
|
||
- blast_radius（0-100）：影響範圍（越高 = 風險越大）
|
||
- rollback_cost（0-100）：回滾難度（越高 = 越難還原）
|
||
|
||
blast_radius 參考：
|
||
- kubectl rollout restart deployment = 10
|
||
- kubectl scale deployment --replicas=N = 15
|
||
- kubectl rollout undo deployment = 25
|
||
- kubectl apply -f = 40
|
||
- kubectl delete deployment = 75
|
||
- kubectl delete pvc = 95
|
||
|
||
🔴 關鍵規則：action 欄位必須是真實的 kubectl 命令，不可用自然語言描述。
|
||
目標資源格式：deployment/<name>，命名空間統一用 awoooi-prod。
|
||
|
||
以 JSON 回覆：
|
||
{{
|
||
"candidates": [
|
||
{{
|
||
"action": "kubectl rollout restart deployment/awoooi-api -n awoooi-prod",
|
||
"blast_radius": 10,
|
||
"rollback_cost": 5,
|
||
"confidence": 0.8,
|
||
"rationale": "重啟可清除 OOM 導致的記憶體碎片化"
|
||
}}
|
||
]
|
||
}}"""
|
||
|
||
def _parse_response(self, response: str) -> dict[str, Any]:
|
||
return self._extract_json(response)
|
||
|
||
def analyze(self, context: dict[str, Any]) -> Any:
|
||
raise NotImplementedError("Use run() for Phase 2 agents")
|
||
|
||
def _degraded_plan(
|
||
self,
|
||
diagnosis: DiagnosisReport,
|
||
latency_ms: int,
|
||
reason: str = "unknown",
|
||
) -> ActionPlan:
|
||
"""熔斷降級：rule-based mock（依 category 推 RESTART 兜底）"""
|
||
category = diagnosis.top_hypothesis.category if diagnosis.top_hypothesis else "Unknown"
|
||
fallback_action = _default_action_for_category(category)
|
||
return ActionPlan(
|
||
candidates=[
|
||
CandidateAction(
|
||
action=fallback_action,
|
||
blast_radius=20,
|
||
rollback_cost=5,
|
||
confidence=0.2,
|
||
rationale=f"[降級] LLM 分析失敗（{reason}），使用類別 {category} 的預設兜底動作",
|
||
)
|
||
],
|
||
diagnosis_report=diagnosis,
|
||
latency_ms=latency_ms,
|
||
vote=AgentVote.DEGRADED,
|
||
degraded=True,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def _fetch_k8s_inventory(namespace: str = "awoooi-prod", timeout_sec: float = 5.0) -> str:
|
||
"""
|
||
取得 K8s 叢集實際 Deployment/StatefulSet 清單，供 Solver prompt 注入。
|
||
|
||
2026-04-17 ogt + Claude Sonnet 4.6 (Checkpoint-2 環境感知):
|
||
- 在生成 kubectl 指令前查詢叢集真實資源，防止 LLM 幻覺資源名（如 awooiii-api）
|
||
- 超時或失敗 → 返回 ""（呼叫端降級為警示模式，不中斷 Solver 主流程）
|
||
- 只執行唯讀 get 指令，不修改叢集
|
||
|
||
Returns:
|
||
"awoooi-api, awoooi-web, postgres, ..." 格式字串，失敗時返回 ""
|
||
"""
|
||
import asyncio as _asyncio
|
||
try:
|
||
cmd = f"kubectl get deployments,statefulsets -n {namespace} -o jsonpath='{{.items[*].metadata.name}}' 2>/dev/null"
|
||
proc = await _asyncio.create_subprocess_shell(
|
||
cmd,
|
||
stdout=_asyncio.subprocess.PIPE,
|
||
stderr=_asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
stdout, _ = await _asyncio.wait_for(proc.communicate(), timeout=timeout_sec)
|
||
except _asyncio.TimeoutError:
|
||
proc.kill()
|
||
logger.warning("k8s_inventory_timeout", namespace=namespace, timeout_sec=timeout_sec)
|
||
return ""
|
||
|
||
raw = (stdout or b"").decode("utf-8", errors="replace").strip()
|
||
if not raw:
|
||
return ""
|
||
|
||
# jsonpath 輸出以空格分隔，轉成可讀逗號格式
|
||
names = [n.strip() for n in raw.split() if n.strip()]
|
||
inventory = ", ".join(names)
|
||
logger.debug("k8s_inventory_fetched", namespace=namespace, count=len(names))
|
||
return inventory
|
||
|
||
except Exception as _e:
|
||
logger.warning("k8s_inventory_failed", namespace=namespace, error=str(_e))
|
||
return ""
|
||
|
||
|
||
def _extract_candidates(parsed: dict[str, Any]) -> list[CandidateAction]:
|
||
"""從 LLM 解析結果提取候選方案（按信心降序）。
|
||
|
||
支援兩種格式：
|
||
1. 標準格式：{"candidates": [{action, blast_radius, rollback_cost, confidence, rationale}]}
|
||
2. OpenClaw Nemo 格式：{"action_title": "...", "risk_level": "...", "confidence": 0.85}
|
||
|
||
2026-04-16 ogt + Claude Sonnet 4.6: 與 diagnostician 同步，修復 openclaw_nemo 格式不相容
|
||
"""
|
||
# OpenClaw Nemo 格式轉換
|
||
# 2026-04-17 ogt + Claude Sonnet 4.6: Nemo path kubectl 驗證
|
||
# 根因：Nemo 回傳 {"action_title": "重啟 Crash Looping Pod"} 自然語言
|
||
# 直接用 action_title → 無 kubectl → auto_approve 誤通過 → 死迴圈
|
||
# 2026-04-24 ogt + Claude Sonnet 4.6: 修復靜默丟棄 → 語意合成兜底
|
||
# 舊：action_title 無 kubectl → return [] → _degraded_plan confidence=0.2
|
||
# 新：先嘗試語意合成 kubectl 指令；真的無從映射才 return []
|
||
if "action_title" in parsed and "candidates" not in parsed:
|
||
action_title = str(parsed.get("action_title", ""))
|
||
confidence = float(parsed.get("confidence", 0.5))
|
||
risk_level = str(parsed.get("risk_level", "medium"))
|
||
risk_to_blast = {"critical": 60, "high": 40, "medium": 25, "low": 10}
|
||
blast = risk_to_blast.get(risk_level.lower(), 30)
|
||
|
||
if "kubectl" in action_title.lower():
|
||
if action_title and confidence > 0:
|
||
return [CandidateAction(
|
||
action=action_title[:200],
|
||
blast_radius=blast,
|
||
rollback_cost=20,
|
||
confidence=confidence,
|
||
rationale=f"OpenClaw Nemo 建議: {action_title}",
|
||
)]
|
||
return []
|
||
|
||
# action_title 無 kubectl → 嘗試語意合成 kubectl 指令
|
||
_at_lower = action_title.lower()
|
||
_synthesized: str | None = None
|
||
if any(w in _at_lower for w in ("rollback", "undo", "回滾", "還原")):
|
||
_synthesized = "kubectl rollout undo deployment -n awoooi-prod"
|
||
elif any(w in _at_lower for w in ("restart", "重啟", "重新啟動")):
|
||
_synthesized = "kubectl rollout restart deployment -n awoooi-prod"
|
||
elif any(w in _at_lower for w in ("scale", "擴容", "縮容", "replicas")):
|
||
_synthesized = "kubectl scale deployment -n awoooi-prod"
|
||
elif any(w in _at_lower for w in ("logs", "日誌", "log")):
|
||
_synthesized = "kubectl logs -n awoooi-prod --tail=100 --selector=app=awoooi-api"
|
||
elif any(w in _at_lower for w in ("describe", "診斷", "diagnos")):
|
||
_synthesized = "kubectl describe pods -n awoooi-prod"
|
||
|
||
if _synthesized:
|
||
logger.debug(
|
||
"solver_nemo_action_synthesized",
|
||
action_title=action_title[:80],
|
||
synthesized=_synthesized,
|
||
)
|
||
return [CandidateAction(
|
||
action=_synthesized,
|
||
blast_radius=blast,
|
||
rollback_cost=20,
|
||
confidence=min(confidence, 0.5), # 合成指令最高 0.5，避免誤入自動執行
|
||
rationale=f"[語意合成] Nemo 建議「{action_title[:80]}」→ 轉為 kubectl 指令",
|
||
)]
|
||
|
||
# 完全無從映射 → return []（交由 _degraded_plan 輸出 category-based 調查指令）
|
||
logger.debug(
|
||
"solver_nemo_no_kubectl_fallback",
|
||
action_title=action_title[:80],
|
||
reason="action_title 無 kubectl 且語意合成失敗，降級至 _degraded_plan",
|
||
)
|
||
return []
|
||
|
||
raw = parsed.get("candidates", [])
|
||
candidates = []
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
c = CandidateAction(
|
||
action=str(item.get("action", ""))[:200],
|
||
blast_radius=max(0, min(100, int(item.get("blast_radius", 50)))),
|
||
rollback_cost=max(0, min(100, int(item.get("rollback_cost", 50)))),
|
||
confidence=float(item.get("confidence", 0.0)),
|
||
rationale=str(item.get("rationale", ""))[:500],
|
||
)
|
||
candidates.append(c)
|
||
candidates.sort(key=lambda c: c.confidence, reverse=True)
|
||
return candidates
|
||
|
||
|
||
def _default_action_for_category(category: str) -> str:
|
||
"""降級時的預設調查指令 — 必須是真實 kubectl/ssh 命令（調查優先，不執行破壞性操作）
|
||
|
||
2026-04-17 ogt + Claude Sonnet 4.6: 改為真實 kubectl 指令
|
||
2026-04-24 ogt + Claude Sonnet 4.6: 擴展非 K8s 類別（ClickHouse/主機磁碟/DB）
|
||
根因：SentryClickHouseMemoryPressure/HostDiskUsageHigh 類別不符任何 K8s 關鍵字
|
||
→ 全部 fallback 到 "kubectl get pods"（無意義診斷指令）
|
||
修復：加入 clickhouse/database/sentry/host/node/infra 類別映射
|
||
"""
|
||
category_lower = category.lower()
|
||
# K8s workload 層
|
||
if "pod" in category_lower or "kube" in category_lower or "crash" in category_lower:
|
||
return "kubectl get pods -n awoooi-prod -o wide"
|
||
if "cpu" in category_lower or "load" in category_lower:
|
||
return "kubectl top pods -n awoooi-prod --sort-by=cpu"
|
||
if "memory" in category_lower or "oom" in category_lower:
|
||
return "kubectl top pods -n awoooi-prod --sort-by=memory"
|
||
if "network" in category_lower or "connect" in category_lower:
|
||
return "kubectl get services -n awoooi-prod"
|
||
if "disk" in category_lower or "storage" in category_lower or "pvc" in category_lower:
|
||
return "kubectl exec -n awoooi-prod deployment/postgresql -- df -h"
|
||
# 外部服務層（非 K8s — 唯讀診斷）
|
||
if "clickhouse" in category_lower or "sentry" in category_lower:
|
||
return "kubectl get pods -n awoooi-prod -l app=sentry -o wide"
|
||
if "database" in category_lower or "postgres" in category_lower or "redis" in category_lower:
|
||
return "kubectl get pods -n awoooi-prod -l tier=database -o wide"
|
||
if "rollback" in category_lower or "deploy" in category_lower or "version" in category_lower:
|
||
return "kubectl rollout history deployment -n awoooi-prod"
|
||
if "latency" in category_lower or "slow" in category_lower or "timeout" in category_lower:
|
||
return "kubectl top pods -n awoooi-prod --sort-by=cpu"
|
||
# 主機層（host/node/infra — 調查指令，kubectl 只查 node 資訊）
|
||
if "host" in category_lower or "node" in category_lower or "infra" in category_lower:
|
||
return "kubectl describe nodes | grep -A5 'Conditions\\|Allocatable'"
|
||
return "kubectl get pods -n awoooi-prod -o wide"
|
||
|
||
|
||
def compute_input_hash(diagnosis: DiagnosisReport) -> str:
|
||
"""計算 Solver 輸入的 fingerprint。"""
|
||
key = diagnosis.evidence_snapshot_id + (
|
||
diagnosis.top_hypothesis.description if diagnosis.top_hypothesis else ""
|
||
)
|
||
return hashlib.sha256(key.encode()).hexdigest()[:16]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Singleton
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_agent: SolverAgent | None = None
|
||
|
||
|
||
def get_solver_agent() -> SolverAgent:
|
||
global _agent
|
||
if _agent is None:
|
||
_agent = SolverAgent()
|
||
return _agent
|