We are building a streaming QoE triage agent that ingests structured CDN telemetry and customer complaints, then reasons about root cause and severity. It is aimed at platform engineers and SREs running media delivery over telecom networks who need to cut mean-time-to-resolution without building a custom ML pipeline. I run the inference back end on Oxlo.ai because its per-request pricing stays flat even when I pack long traceroute dumps or multi-line encoder logs into the prompt.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Bootstrap the Oxlo.ai client
I initialize the OpenAI-compatible client pointing at Oxlo.ai and select llama-3.3-70b for reliable structured outputs.
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY"
)Step 2: Model the incident
I keep the input schema flat so NOC tools can dump JSON straight into it without heavy transformation.
import json
from dataclasses import dataclass, asdict
@dataclass
class StreamingIncident:
ticket_id: str
customer_complaint: str
edge_server: str
avg_bitrate_mbps: float
buffer_ratio: float
packet_loss_pct: float
latency_ms: float
affected_region: strStep 3: Write the system prompt
The prompt constrains the model to a known JSON schema and encodes domain heuristics I normally see in runbooks.
SYSTEM_PROMPT = """You are a senior telecom operations analyst specializing in streaming media delivery.
Your job is to triage Quality of Experience (QoE) incidents from CDN logs and subscriber complaints.
Analyze the provided incident data and produce a JSON object with these exact keys:
- root_cause: one of [Network Congestion, CDN Edge Failure, Client Device Issue, Content Encoding Error, Unknown]
- severity: one of [P1, P2, P3, P4] where P1 is a customer-facing outage
- impacted_service: the streaming service or region affected
- remediation: a list of 2 to 4 concrete operational steps
- confidence: an integer 0 to 100
Rules:
- If packet_loss_pct is greater than 2.0 and latency_ms is greater than 150, prioritize Network Congestion.
- If buffer_ratio is greater than 0.15 and avg_bitrate_mbps is below 2.0, flag as CDN Edge Failure.
- Keep remediation steps specific to telecom and media workflows.
- Output ONLY valid JSON. Do not wrap it in markdown."""Step 4: Implement the triage function
This function serializes the incident, sends it to Oxlo.ai, and parses the response. I keep temperature low to enforce deterministic reasoning. The full snippet repeats prior definitions so you can run it standalone.
import json
from dataclasses import dataclass, asdict
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY"
)
@dataclass
class StreamingIncident:
ticket_id: str
customer_complaint: str
edge_server: str
avg_bitrate_mbps: float
buffer_ratio: float
packet_loss_pct: float
latency_ms: float
affected_region: str
SYSTEM_PROMPT = """You are a senior telecom operations analyst specializing in streaming media delivery.
Your job is to triage Quality of Experience (QoE) incidents from CDN logs and subscriber complaints.
Analyze the provided incident data and produce a JSON object with these exact keys:
- root_cause: one of [Network Congestion, CDN Edge Failure, Client Device Issue, Content Encoding Error, Unknown]
- severity: one of [P1, P2, P3, P4] where P1 is a customer-facing outage
- impacted_service: the streaming service or region affected
- remediation: a list of 2 to 4 concrete operational steps
- confidence: an integer 0 to 100
Rules:
- If packet_loss_pct is greater than 2.0 and latency_ms is greater than 150, prioritize Network Congestion.
- If buffer_ratio is greater than 0.15 and avg_bitrate_mbps is below 2.0, flag as CDN Edge Failure.
- Keep remediation steps specific to telecom and media workflows.
- Output ONLY valid JSON. Do not wrap it in markdown."""
def triage_incident(incident: StreamingIncident) -> dict:
user_message = json.dumps(asdict(incident), indent=2)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0.1,
max_tokens=512
)
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("```
")[1]
if raw.startswith("json"):
raw = raw[3:]
return json.loads(raw.strip())Run it
I feed the agent a sample ticket that looks like a real Bronx sports outage.
if __name__ == "__main__":
sample = StreamingIncident(
ticket_id="TTV-2025-0892",
customer_complaint="Video keeps freezing during live sports stream",
edge_server="cdn-ny04.oxlo.ai.media",
avg_bitrate_mbps=1.2,
buffer_ratio=0.22,
packet_loss_pct=0.8,
latency_ms=95.0,
affected_region="NYC-Bronx"
)
result = triage_incident(sample)
print(json.dumps(result, indent=2))Expected output:
{
"root_cause": "CDN Edge Failure",
"severity": "P2",
"impacted_service": "NYC-Bronx live sports stream",
"remediation": [
"Drain traffic from cdn-ny04.oxlo.ai.media to the nearest warm edge pool",
"Check disk and memory saturation on the origin shield node",
"Validate manifest and fragment availability for the live sports channel",
"Escalate to the CDN on-call if error rate does not improve within 10 minutes"
],
"confidence": 87
}Wrap-up
The agent turns unstructured complaints and metrics into structured incident metadata without maintaining a custom model. Two concrete next steps: wire this function into a Kafka consumer so it processes tickets in real time, or switch to kimi-k2.6 on Oxlo.ai for deeper chain-of-thought reasoning when you need to correlate BGP routing logs with encoder health. Because Oxlo.ai charges per request rather than per token, stuffing multi-line syslogs or traceroute dumps into the context window does not inflate cost. See https://oxlo.ai/pricing for current plan options.
Top comments (0)