215 lines
7.5 KiB
Python
215 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from app.correlation import CorrelationEventRecord
|
|
from app.models import NotificationDecision, ProcessorForwardEnvelope
|
|
|
|
|
|
@dataclass
|
|
class LLMCorrelationResult:
|
|
ok: bool
|
|
role: str
|
|
kind: str
|
|
reason: str | None
|
|
confidence: str | None
|
|
parent_event_id: str | None = None
|
|
parent_correlation_id: str | None = None
|
|
suppress_child: bool = False
|
|
error: str | None = None
|
|
|
|
|
|
class LLMCorrelationAdapter:
|
|
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,
|
|
recent_events: list[CorrelationEventRecord],
|
|
) -> LLMCorrelationResult:
|
|
prompt = self._build_prompt(envelope, decision, recent_events)
|
|
|
|
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()
|
|
|
|
payload = response.json()
|
|
raw_text = payload.get("response", "")
|
|
parsed = self._parse_json_object(raw_text)
|
|
|
|
role = self._normalize_role(parsed.get("role"))
|
|
kind = self._normalize_text(parsed.get("kind")) or "unknown"
|
|
reason = self._normalize_text(parsed.get("reason"))
|
|
confidence = self._normalize_confidence(parsed.get("confidence"))
|
|
parent_event_id = self._normalize_text(parsed.get("parent_event_id"))
|
|
parent_correlation_id = self._normalize_text(parsed.get("parent_correlation_id"))
|
|
suppress_child = bool(parsed.get("suppress_child", False))
|
|
|
|
if role == "child" and not (parent_event_id or parent_correlation_id):
|
|
return LLMCorrelationResult(
|
|
ok=False,
|
|
role="standalone",
|
|
kind=kind,
|
|
reason=reason,
|
|
confidence=confidence,
|
|
error="LLM correlation returned child without parent reference",
|
|
)
|
|
|
|
return LLMCorrelationResult(
|
|
ok=True,
|
|
role=role,
|
|
kind=kind,
|
|
reason=reason,
|
|
confidence=confidence,
|
|
parent_event_id=parent_event_id,
|
|
parent_correlation_id=parent_correlation_id,
|
|
suppress_child=suppress_child,
|
|
error=None,
|
|
)
|
|
|
|
def _build_prompt(
|
|
self,
|
|
envelope: ProcessorForwardEnvelope,
|
|
decision: NotificationDecision,
|
|
recent_events: list[CorrelationEventRecord],
|
|
) -> str:
|
|
event = envelope.event
|
|
zbx_context = event.zabbix_context or {}
|
|
|
|
current_event = {
|
|
"event_id": event.event_id,
|
|
"correlation_id": event.correlation_id,
|
|
"severity": decision.severity,
|
|
"routing_class": decision.routing_class,
|
|
"host": event.host,
|
|
"service": zbx_context.get("service") or event.service,
|
|
"scope": zbx_context.get("scope"),
|
|
"component": zbx_context.get("component"),
|
|
"domain": zbx_context.get("domain"),
|
|
"trigger_name": event.trigger_name,
|
|
"value": event.value,
|
|
"tags": event.tags,
|
|
"item_key": zbx_context.get("item_key"),
|
|
"alert_scope": zbx_context.get("alert_scope"),
|
|
"correlation_scopes": zbx_context.get("correlation_scopes"),
|
|
}
|
|
|
|
recent_payload = [
|
|
{
|
|
"event_id": item.event_id,
|
|
"correlation_id": item.correlation_id,
|
|
"kind": item.kind,
|
|
"severity": item.severity,
|
|
"routing_class": item.routing_class,
|
|
"root_candidate": item.root_candidate,
|
|
"role": item.role,
|
|
"group_id": item.group_id,
|
|
"parent_event_id": item.parent_event_id,
|
|
"parent_correlation_id": item.parent_correlation_id,
|
|
"host": item.host,
|
|
"service": item.service,
|
|
"scope": item.scope,
|
|
"component": item.component,
|
|
"domain": item.domain,
|
|
"tags": item.tags,
|
|
}
|
|
for item in recent_events[-12:]
|
|
]
|
|
|
|
return (
|
|
"You are a cautious AIOps correlation assistant.\n"
|
|
"/no_think\n"
|
|
"Decide whether the current alert is standalone, a root cause candidate, or a downstream child event.\n"
|
|
"Rules:\n"
|
|
"1. Output JSON only.\n"
|
|
"2. Allowed role values: standalone, root, child.\n"
|
|
"3. Use the current event and recent events in the same service/scope/domain context.\n"
|
|
"4. Prefer standalone when uncertain.\n"
|
|
"5. child role requires a parent_event_id or parent_correlation_id from recent_events.\n"
|
|
"6. Never suggest suppress_child for High or Disaster alerts.\n"
|
|
"7. suppress_child may be true only for role=child and only when confidence is high.\n"
|
|
"8. confidence must be one of: low, medium, high.\n"
|
|
"9. Synthetic website alerts may be children of recent app/container outages if service/scope/domain matches.\n\n"
|
|
"Return exactly this JSON schema:\n"
|
|
"{\n"
|
|
' "role": "standalone|root|child",\n'
|
|
' "kind": "short_kind_name_or_unknown",\n'
|
|
' "reason": "short explanation",\n'
|
|
' "confidence": "low|medium|high",\n'
|
|
' "parent_event_id": "optional",\n'
|
|
' "parent_correlation_id": "optional",\n'
|
|
' "suppress_child": false\n'
|
|
"}\n\n"
|
|
f"Current event:\n{json.dumps(current_event, ensure_ascii=False, indent=2)}\n\n"
|
|
f"Recent events:\n{json.dumps(recent_payload, ensure_ascii=False, indent=2)}"
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_json_object(raw_text: str) -> dict:
|
|
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_role(value: object) -> str:
|
|
role = str(value or "standalone").strip().lower()
|
|
if role not in {"standalone", "root", "child"}:
|
|
return "standalone"
|
|
return role
|
|
|
|
@staticmethod
|
|
def _normalize_confidence(value: object) -> str:
|
|
confidence = str(value or "low").strip().lower()
|
|
if confidence not in {"low", "medium", "high"}:
|
|
return "low"
|
|
return confidence
|
|
|
|
@staticmethod
|
|
def _normalize_text(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|