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