We are going to build a high-accuracy incident triage agent that reads unstructured alerts and outputs structured severity classifications with confidence scores. This helps on-call engineers reduce false positives and catch real outages faster without waking the team for noise.
What you'll need
Before starting, make sure you have the following:
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK and Pydantic installed:
pip install openai pydantic
Step 1: Define the structured schema
Locking down the output shape with Pydantic prevents the model from hallucinating fields and gives us native validation. We define an enum for severity and a result model that requires chain-of-thought reasoning.
from pydantic import BaseModel, Field
from typing import List
from enum import Enum
class Severity(str, Enum):
P0 = "P0"
P1 = "P1"
P2 = "P2"
P3 = "P3"
P4 = "P4"
class TriageResult(BaseModel):
severity: Severity
affected_services: List[str] = Field(..., description="List of affected microservices")
root_cause_category: str = Field(..., description="One of: infrastructure, deployment, dependency, data, unknown")
remediation: str = Field(..., description="Immediate action to take")
reasoning: str = Field(..., description="Brief chain-of-thought reasoning")
confidence: float = Field(..., ge=0.0, le=1.0, description="Model confidence 0.0-1.0")Step 2: Initialize the Oxlo.ai client
Oxlo.ai is fully OpenAI SDK compatible, so switching providers takes a single line. Point the base URL to Oxlo.ai and load your key.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)Step 3: Craft the system prompt
The system prompt is the highest-leverage optimization for accuracy. We force explicit reasoning, ban vague language, and tie severity levels to concrete business impact.
SYSTEM_PROMPT = """You are an expert Site Reliability Engineer triaging production incidents.
Your job is to read an alert or error report and produce a precise structured assessment.
Rules:
- Always think step by step in the reasoning field before selecting severity.
- Severity definitions:
P0: Revenue-impacting outage affecting all users. No workaround.
P1: Major feature degraded for many users. Workaround exists but is painful.
P2: Partial degradation or isolated customer impact. Workaround is easy.
P3: Minor bug or cosmetic issue. No user impact.
P4: Noise, monitoring artifact, or informational log.
- affected_services must be specific microservice names. If unknown, use ["unknown"].
- root_cause_category must be exactly one of: infrastructure, deployment, dependency, data, unknown.
- remediation must be a concrete, actionable command or step, not generic advice.
- confidence must reflect your certainty given ambiguous input. Low confidence when logs are unclear.
Output strictly valid JSON matching the requested schema.
"""Step 4: Single-pass extraction with JSON mode
We start with one call using Qwen 3 32B. Its reasoning capabilities handle agentic workflows well, and Oxlo.ai supports JSON mode natively so we can enforce schema compliance at the API level.
import json
def extract_once(alert_text: str) -> TriageResult:
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": alert_text},
],
response_format={"type": "json_object"},
temperature=0.1,
)
raw = response.choices[0].message.content
return TriageResult.model_validate_json(raw)
# Quick sanity check
alert = """
[ALERT] payment-gateway-prod
HTTP 500 spike started 2024-05-21 03:42 UTC.
Error rate: 94%. Latency p99: 12s.
All regions affected. Checkout flow down.
Rollback to v2.3.1 did not resolve.
"""
result = extract_once(alert)
print(result.model_dump_json(indent=2))Step 5: Self-consistency voting
A single sample can be unlucky. We run three independent extractions and take the majority vote for severity. Because Oxlo.ai uses flat per-request pricing, three calls cost exactly three requests regardless of how long the log dump is, which keeps costs predictable when you add reasoning depth.
from collections import Counter
def triage_with_voting(alert_text: str, n: int = 3) -> TriageResult:
candidates: list[TriageResult] = []
for i in range(n):
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": alert_text},
],
response_format={"type": "json_object"},
temperature=0.2,
seed=i,
)
raw = response.choices[0].message.content
candidates.append(TriageResult.model_validate_json(raw))
# Majority vote on severity
severities = [c.severity for c in candidates]
winner_sev = Counter(severities).most_common(1)[0][0]
# Pick the candidate with winning severity and highest confidence
best = max(
[c for c in candidates if c.severity == winner_sev],
key=lambda x: x.confidence
)
best.confidence = round(best.confidence * (Counter(severities)[winner_sev] / n), 2)
return best
voted = triage_with_voting(alert)
print(voted.model_dump_json(indent=2))Step 6: Judge verification for low-confidence cases
When voting confidence stays below 0.8, we escalate to Kimi K2.6. Its advanced reasoning works well as a judge that critiques the proposed triage against the original alert.
JUDGE_PROMPT = """You are a senior SRE reviewing an automated triage.
Severity definitions: P0 is all-users revenue outage with no workaround. P1 is major degradation with painful workaround. P2 is partial degradation with easy workaround. P3 is minor bug. P4 is noise.
Given the original alert and a proposed triage, verify if the severity and root cause are correct.
Respond with JSON containing:
- approved: bool
- corrected_severity: string or null
- corrected_root_cause: string or null
- critique: string explaining your reasoning
"""
def judge_triage(alert_text: str, proposal: TriageResult) -> TriageResult:
review_input = f"ALERT:\n{alert_text}\n\nPROPOSED TRIAGE:\n{proposal.model_dump_json()}"
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": JUDGE_PROMPT},
{"role": "user", "content": review_input},
],
response_format={"type": "json_object"},
temperature=0.1,
)
review = json.loads(response.choices[0].message.content)
if not review.get("approved", True):
proposal.severity = review.get("corrected_severity", proposal.severity)
proposal.root_cause_category = review.get("corrected_root_cause", proposal.root_cause_category)
proposal.confidence = 0.95
proposal.reasoning += f" | Judge critique: {review.get('critique', 'No issues')}"
return proposal
def triage_pipeline(alert_text: str) -> TriageResult:
result = triage_with_voting(alert_text)
if result.confidence < 0.8:
result = judge_triage(alert_text, result)
return resultRun it
Feed a messy, ambiguous alert into the pipeline and inspect the final structured result.
messy_alert = """
[PagerDuty] service: user-auth-api
Latency p95 elevated to 890ms (threshold 500ms).
Error rate at 3%. Only region: eu-west-1.
Customers reporting slow login but not failure.
Last deploy: user-auth-api v1.4.2 45 min ago.
"""
final = triage_pipeline(messy_alert)
print(final.model_dump_json(indent=2))Example output:
{
"severity": "P2",
"affected_services": ["user-auth-api"],
"root_cause_category": "deployment",
"remediation": "Consider rolling back user-auth-api v1.4.2 and monitor latency in eu-west-1.",
"reasoning": "Latency spike is regional, error rate is low, and users report slowness not failure. This matches P2 partial degradation. | Judge critique: Agreed. Deployment correlation is strong.",
"confidence": 0.85
}Wrap-up and next steps
First, wire this into your existing PagerDuty or Opsgenie webhook so it runs on every incoming alert and posts the structured triage to Slack. Second, add a feedback loop by logging judge overrides to a dataset, then refine the system prompt weekly based on real mismatches to push accuracy higher without retraining.
Top comments (0)