Field operations centers for utilities and telecom still translate raw outage reports into dispatch tickets by hand. I built a lightweight triage agent that reads alarm feeds, crew notes, and customer complaints, then returns validated JSON with severity scores, safety flags, and crew recommendations. It runs on Oxlo.ai, where flat per-request pricing means a long incident log costs the same as a short ping.
What you'll need
- Python 3.10+
- The OpenAI SDK:
pip install openai - Pydantic for validation:
pip install pydantic - An Oxlo.ai API key from https://portal.oxlo.ai
Oxlo.ai bills per request, not per token, so feeding the model a ten-page incident history does not inflate your cost. See https://oxlo.ai/pricing for current plan details.
Step 1: Connect and test the Oxlo.ai endpoint
Before ingesting real feeds, verify that the SDK reaches Oxlo.ai and authenticates correctly. This snippet uses the exact client pattern you will reuse in every downstream step.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = "You are a connectivity test assistant."
user_message = "Respond with exactly: connection_ok"
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
print(response.choices[0].message.content)Step 2: Define the triage schema
The agent must emit structured data, not prose, so downstream dispatch systems can consume it. I use Pydantic to enforce types and to give the LLM a clear target in the system prompt.
from pydantic import BaseModel, Field
from typing import Literal, List
class OutageTriage(BaseModel):
severity: Literal["low", "medium", "high", "critical"] = Field(
description="Estimated severity of the outage."
)
probable_cause: Literal["weather", "equipment_failure", "vegetation", "unknown", "cyber"] = Field(
description="Most likely root cause."
)
affected_assets: List[str] = Field(
default_factory=list, description="Specific equipment or locations affected."
)
estimated_customers_impacted: int = Field(
description="Number of customers out. Estimate if unclear."
)
required_crew_type: Literal["lineman", "substation_tech", "fiber_splicer", "none"] = Field(
description="Type of crew required for the fix."
)
safety_hazards: List[str] = Field(
default_factory=list, description="Any hazards mentioned or implied."
)
dispatch_notes: str = Field(
description="A 1-2 sentence directive for the dispatch desk."
)Step 3: Write the system prompt
The system prompt is the only place where we teach the model the schema and the domain rules. Keep it strict: require valid JSON, define every enum, and forbid markdown wrappers.
SYSTEM_PROMPT = """You are a utility and telecom outage triage engineer.
Your job is to read raw field reports, SCADA alarms, and customer complaints, then produce a structured triage assessment.
Follow these rules:
- severity: classify as low, medium, high, or critical based on customer count, safety risk, and infrastructure type.
- probable_cause: choose from weather, equipment_failure, vegetation, cyber, or unknown.
- affected_assets: list specific equipment (for example, "Transformer T-104", "Feeder F-12", "Cell Tower 85B").
- estimated_customers_impacted: provide a number. If unclear, estimate conservatively.
- required_crew_type: choose lineman, substation_tech, fiber_splicer, or none.
- safety_hazards: list any hazards mentioned or implied (for example, "downed line", "gas leak proximity").
- dispatch_notes: write a 1-2 sentence directive for the dispatch desk.
Output strictly valid JSON matching the requested schema. Do not add markdown formatting around the JSON."""Step 4: Normalize incoming reports
Real feeds arrive as multiline blobs with inconsistent spacing and mixed sources. A thin normalizer strips empty lines and adds a consistent header so the LLM sees a uniform document.
def normalize_report(raw: str) -> str:
lines = raw.strip().splitlines()
cleaned = [line.strip() for line in lines if line.strip()]
return "FIELD REPORT:\n" + "\n".join(cleaned)Step 5: Call the LLM and parse structured output
Here we route the normalized report to Oxlo.ai and coerce the response into our Pydantic model. A small sanitizer removes accidental markdown fences before parsing.
import json
from pydantic import ValidationError
def triage_outage(raw_report: str) -> OutageTriage:
user_message = normalize_report(raw_report)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
content = response.choices[0].message.content.strip()
# Strip markdown code fences if the model includes them
if content.startswith("
```"):
content = content.split("```
json")[-1].split("
```
")[0].strip()
data = json.loads(content)
return OutageTriage(**data)Step 6: Add dispatch rules
Once the triage object is validated, plain Python logic maps severity to SLA windows and safety flags. This keeps the LLM focused on classification while engineering owns the business rules.
def dispatch_recommendation(triage: OutageTriage) -> dict:
sla_minutes = {
"low": 1440,
"medium": 240,
"high": 90,
"critical": 30,
}
return {
"dispatch_within_minutes": sla_minutes[triage.severity],
"crew": triage.required_crew_type,
"assets": triage.affected_assets,
"notes": triage.dispatch_notes,
"safety_flag": len(triage.safety_hazards) > 0,
}Run it
The snippet below passes a realistic multi-source report through the full pipeline. You should see structured JSON that a dispatch dashboard can ingest directly.
if __name__ == "__main__":
raw = """
Substation 7 alarm at 14:03. Breaker BR-702 tripped.
Field crew reports loud bang from transformer T-7A.
1200 homes out in sector 4.
Tree branch on primary line near Maple St.
Weather clear. No injuries reported but wire is down across road.
"""
result = triage_outage(raw)
print(result.model_dump_json(indent=2))
dispatch = dispatch_recommendation(result)
print(json.dumps(dispatch, indent=2))Example output:
{
"severity": "critical",
"probable_cause": "vegetation",
"affected_assets": [
"Substation 7",
"Breaker BR-702",
"Transformer T-7A",
"Primary line near Maple St"
],
"estimated_customers_impacted": 1200,
"required_crew_type": "lineman",
"safety_hazards": [
"downed line across road"
],
"dispatch_notes": "Dispatch lineman crew immediately to Substation 7. Downed primary line presents public safety hazard."
}
{
"dispatch_within_minutes": 30,
"crew": "lineman",
"assets": [
"Substation 7",
"Breaker BR-702",
"Transformer T-7A",
"Primary line near Maple St"
],
"notes": "Dispatch lineman crew immediately to Substation 7. Downed primary line presents public safety hazard.",
"safety_flag": true
}Wrap-up
This agent turns unstructured outage noise into structured dispatch data. Two concrete next steps: deploy it as a FastAPI service and point your SCADA or NOC webhooks at the endpoint, or add a second Oxlo.ai stage using qwen-3-32b to generate Spanish dispatch notes for bilingual crews without worrying about token costs on long reports. Both paths stay on the same OpenAI-compatible client, so the only change is the model string.
Top comments (0)