"""Causal reasoning engine - orchestrates LLM analysis.""" import json import logging import time from datetime import datetime, timezone from typing import Optional import requests from ..config import LLMConfig from ..models import Investigation, LogEntry, MetricSnapshot, GitChange, CausalChain, Severity from . import prompts logger = logging.getLogger(__name__) class ReasoningEngine: """LLM-powered causal reasoning engine.""" def __init__(self, config: LLMConfig): self.config = config self._session = requests.Session() self._session.headers.update({ "Content-Type": "application/json", "Accept": "application/json", }) def _call_llm(self, system_prompt: str, user_prompt: str) -> str: """Call the LLM API for analysis.""" payload = { "model": self.config.model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "max_tokens": self.config.max_tokens, "temperature": self.config.temperature, "stream": False, } start = time.time() try: response = self._session.post( f"{self.config.base_url}/chat/completions", json=payload, timeout=self.config.timeout ) elapsed = time.time() - start logger.info("LLM call took %.1fs", elapsed) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"] except requests.exceptions.ConnectionError: logger.error("Cannot connect to LLM at %s", self.config.base_url) raise RuntimeError(f"LLM endpoint unreachable: {self.config.base_url}") except requests.exceptions.Timeout: logger.error("LLM call timed out after %ds", self.config.timeout) raise TimeoutError(f"LLM call timed out ({self.config.timeout}s)") except (requests.exceptions.HTTPError, KeyError, json.JSONDecodeError) as e: logger.error("LLM call failed: %s", e) raise RuntimeError(f"LLM call failed: {e}") def _format_logs(self, logs: list[LogEntry], max_entries: int = 100) -> str: """Format log entries for LLM consumption.""" # Prioritize errors and warnings error_logs = [l for l in logs if l.severity in (Severity.ERROR, Severity.CRITICAL)] warning_logs = [l for l in logs if l.severity == Severity.WARNING] info_logs = [l for l in logs if l.severity == Severity.INFO] lines = [] # Errors first, then warnings, then info to fill remaining for log in error_logs[:max_entries]: lines.append(f"[{log.timestamp.isoformat()}] [{log.severity.value.upper()}] {log.source_name}: {log.message[:300]}") remaining = max_entries - len(error_logs) if remaining > 0: for log in warning_logs[:remaining]: lines.append(f"[{log.timestamp.isoformat()}] [{log.severity.value.upper()}] {log.source_name}: {log.message[:300]}") remaining -= 1 if remaining > 0 and lines: lines.append(f"... and {len(info_logs)} info entries") elif not lines: lines.append("No logs collected.") return "\n".join(lines) def _format_metrics(self, metrics: list[MetricSnapshot]) -> str: """Format metrics for LLM consumption.""" if not metrics: return "No metrics collected." latest = metrics[-1] lines = [ f"Timestamp: {latest.timestamp.isoformat()}", f"CPU: {latest.cpu_percent:.1f}%", f"Memory: {latest.memory_percent:.1f}%", f"Disk: {latest.disk_percent:.1f}%", f"Load Average: {latest.load_avg}", ] if latest.top_processes: lines.append("\nTop Processes:") for proc in latest.top_processes[:5]: lines.append(f" - {proc['name']} (PID {proc['pid']}): CPU {proc['cpu']:.1f}%, MEM {proc['memory']:.1f}%") return "\n".join(lines) def _format_git_changes(self, changes: list[GitChange]) -> str: """Format git changes for LLM consumption.""" if not changes: return "No recent git changes." lines = [] for change in changes: files = ", ".join(change.files_changed[:5]) if len(change.files_changed) > 5: files += f" (+{len(change.files_changed) - 5} more)" lines.append( f"[{change.timestamp.isoformat()}] {change.commit_hash} by {change.author}\n" f" {change.message}\n" f" Files: {files}" ) return "\n".join(lines) def analyze( self, query: str, logs: list[LogEntry], metrics: list[MetricSnapshot], git_changes: list[GitChange], ) -> Investigation: """Perform full causal analysis. Args: query: What the user is investigating logs: Collected log entries metrics: System metrics snapshots git_changes: Recent git changes Returns: Investigation with causal chains """ user_prompt = prompts.ANALYSIS_PROMPT.format( query=query, metrics=self._format_metrics(metrics), container_logs=self._format_logs(logs), system_logs="Included above", git_changes=self._format_git_changes(git_changes), ) logger.info("Running causal analysis for: %s", query[:100]) response = self._call_llm(prompts.SYSTEM_PROMPT, user_prompt) investigation = Investigation( query=query, timestamp=datetime.now(timezone.utc), logs=logs, metrics=metrics, git_changes=git_changes, raw_response=response, summary=response[:1000], # Quick summary ) # Try to extract structured causal chains from response investigation.causal_chains = self._extract_causal_chains(response) return investigation def triage(self, query: str, logs: list[LogEntry]) -> str: """Quick triage - assess what we know and what we need. Args: query: What the user is investigating logs: Available log entries Returns: Triage assessment """ user_prompt = prompts.TRIAGE_PROMPT.format( query=query, logs=self._format_logs(logs), ) return self._call_llm(prompts.SYSTEM_PROMPT, user_prompt) def _extract_causal_chains(self, response: str) -> list[CausalChain]: """Try to extract structured causal chains from LLM response. This is a best-effort parse — the full response is still the source of truth. """ chains = [] # Simple extraction: look for chain-like patterns lines = response.split("\n") current_chain = [] for line in lines: stripped = line.strip() # Look for bullet points that describe events if stripped.startswith(("- ", "• ", "* ")) and ("caused" in stripped.lower() or "led to" in stripped.lower() or "because" in stripped.lower()): current_chain.append(stripped[2:]) elif current_chain and (stripped.startswith("###") or stripped.startswith("##")): # New section — save chain if we have one if len(current_chain) >= 2: chains.append(CausalChain( symptom="See analysis", root_cause="See analysis", chain=current_chain, confidence=0.7, )) current_chain = [] return chains def close(self): """Close the HTTP session.""" self._session.close()