162 lines
4.9 KiB
Python
162 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.models import NotificationDecision, ProcessorForwardEnvelope
|
|
|
|
|
|
@dataclass
|
|
class TriageResult:
|
|
ok: bool
|
|
verdict: str
|
|
classification: str | None
|
|
reason: str | None
|
|
error: str | None = None
|
|
|
|
|
|
class LLMTriageAdapter:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
model: str,
|
|
timeout_seconds: float = 45,
|
|
verify_tls: bool = True,
|
|
temperature: float = 0.1,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.model = model
|
|
self.temperature = temperature
|
|
|
|
self.client = httpx.AsyncClient(
|
|
timeout=timeout_seconds,
|
|
verify=verify_tls,
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
await self.client.aclose()
|
|
|
|
async def generate(
|
|
self,
|
|
envelope: ProcessorForwardEnvelope,
|
|
decision: NotificationDecision,
|
|
) -> TriageResult:
|
|
prompt = self._build_prompt(envelope, decision)
|
|
|
|
response = await self.client.post(
|
|
f"{self.base_url}/api/generate",
|
|
json={
|
|
"model": self.model,
|
|
"prompt": prompt,
|
|
"stream": False,
|
|
"format": "json",
|
|
"think": False,
|
|
"options": {
|
|
"temperature": self.temperature,
|
|
},
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
raw_text = data.get("response", "")
|
|
parsed = self._parse_json_object(raw_text)
|
|
|
|
verdict = self._normalize_verdict(parsed.get("verdict"))
|
|
classification = self._normalize_text(parsed.get("classification"))
|
|
reason = self._normalize_text(parsed.get("reason"))
|
|
|
|
if verdict not in {"notify", "suppress", "hold"}:
|
|
return TriageResult(
|
|
ok=False,
|
|
verdict="hold",
|
|
classification=None,
|
|
reason=None,
|
|
error=f"Invalid triage verdict: {verdict}",
|
|
)
|
|
|
|
return TriageResult(
|
|
ok=True,
|
|
verdict=verdict,
|
|
classification=classification,
|
|
reason=reason,
|
|
error=None,
|
|
)
|
|
|
|
def _build_prompt(
|
|
self,
|
|
envelope: ProcessorForwardEnvelope,
|
|
decision: NotificationDecision,
|
|
) -> str:
|
|
event = envelope.event
|
|
zbx_context = event.zabbix_context or {}
|
|
|
|
context = {
|
|
"severity": decision.severity,
|
|
"reason": decision.reason,
|
|
"event_phase": decision.event_phase,
|
|
"host": event.host,
|
|
"service": event.service,
|
|
"trigger_name": event.trigger_name,
|
|
"fingerprint": decision.fingerprint,
|
|
"repeat_count": decision.repeat_count,
|
|
"item_id": event.item_id,
|
|
"trigger_id": event.trigger_id,
|
|
"event_id": event.event_id,
|
|
"value": event.value,
|
|
"tags": event.tags,
|
|
"zabbix_context": zbx_context,
|
|
"alert_scope": zbx_context.get("alert_scope", "unknown"),
|
|
}
|
|
|
|
return (
|
|
"You are a cautious AIOps triage assistant. /no_think\n"
|
|
"Decide whether a low-severity Zabbix alert should notify, suppress, or stay on hold.\n"
|
|
"Rules:\n"
|
|
"1. Output JSON only.\n"
|
|
"2. Allowed verdict values: notify, suppress, hold.\n"
|
|
"3. notify = actionable low-severity event worth sending to operator now.\n"
|
|
"4. suppress = obvious noise, repetition, or insignificant deviation.\n"
|
|
"5. hold = uncertain, informational, or not enough evidence.\n"
|
|
"6. Never escalate to email; low-severity notify means Matrix only.\n"
|
|
"7. Be conservative.\n\n"
|
|
"Return exactly this JSON schema:\n"
|
|
"{\n"
|
|
' "verdict": "notify|suppress|hold",\n'
|
|
' "classification": "actionable|noise|informational|unknown",\n'
|
|
' "reason": "short explanation"\n'
|
|
"}\n\n"
|
|
f"Alert context:\n{json.dumps(context, ensure_ascii=False, indent=2)}"
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_json_object(raw_text: str) -> dict[str, Any]:
|
|
text = raw_text.strip()
|
|
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
fenced = re.search(r"\{.*\}", text, flags=re.DOTALL)
|
|
if fenced:
|
|
return json.loads(fenced.group(0))
|
|
|
|
raise ValueError(f"Unable to parse LLM JSON response: {raw_text}")
|
|
|
|
@staticmethod
|
|
def _normalize_verdict(value: Any) -> str:
|
|
if value is None:
|
|
return "hold"
|
|
return str(value).strip().lower()
|
|
|
|
@staticmethod
|
|
def _normalize_text(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None |