We are going to build a lightweight event-driven incident triage pipeline. It ingests JSON events from a message queue, uses an LLM to classify severity and suggest remediation, and routes alerts to either a high-priority pager mock or a standard ticketing mock. This pattern scales from observability webhooks to IoT telemetry without changing the core logic.
What You'll Need
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
Oxlo.ai uses request-based pricing, so processing thousands of events costs the same per request regardless of how large the logs are. See https://oxlo.ai/pricing for details.
Step 1: Scaffold the Event Bus
I start with a simple in-memory queue and typed event dataclass. In production you would swap this for Redis Streams or Kafka, but the interface stays identical.
import queue
from dataclasses import dataclass
from typing import Dict, Any
@dataclass
class PlatformEvent:
event_id: str
source: str
payload: Dict[str, Any]
event_bus = queue.Queue(maxsize=100)
def produce_sample_events():
samples = [
PlatformEvent("evt-1", "web-check", {"status": 500, "latency_ms": 12000, "region": "us-east-1", "message": "Database connection timeout"}),
PlatformEvent("evt-2", "cron", {"status": 200, "latency_ms": 45, "region": "eu-west-1", "message": "Nightly backup completed successfully"}),
PlatformEvent("evt-3", "api-gateway", {"status": 503, "latency_ms": 800, "region": "ap-south-1", "message": "Rate limit exceeded on /v1/orders"}),
]
for evt in samples:
event_bus.put(evt)
Step 2: Write the System Prompt
The prompt is the contract. I force JSON output so the downstream router does not need to parse free text.
SYSTEM_PROMPT = """You are an event classifier for a platform operations pipeline.
Analyze the incoming event payload and return a JSON object with exactly these keys:
- severity: one of "critical", "warning", "info"
- reason: a one-sentence explanation
- action: one of "page_oncall", "create_ticket", "ignore"
- remediation: a concrete shell command or configuration fix if applicable, otherwise "none"
Respond with JSON only. Do not wrap the output in markdown fences."""
Step 3: Build the Oxlo.ai Inference Client
I use the OpenAI SDK as a drop-in client for Oxlo.ai. I picked llama-3.3-70b because it handles structured JSON reliably at high throughput, though qwen-3-32b is a solid alternative if your events include multilingual logs. With Oxlo.ai's request-based pricing the cost per event stays flat even when the payload is large.
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def classify_event(payload: dict) -> dict:
user_message = json.dumps(payload, 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=256,
)
raw = response.choices[0].message.content.strip()
return json.loads(raw)
Step 4: Create the Event Consumer
The worker thread pulls events, calls Oxlo.ai, and routes based on the returned action. I keep the handlers as simple prints so the logic is obvious.
def route_action(event: PlatformEvent, decision: dict):
action = decision.get("action")
if action == "page_oncall":
print(f"[PAGER] {event.event_id} | severity={decision['severity']} | {decision['reason']}")
elif action == "create_ticket":
print(f"[TICKET] {event.event_id} | severity={decision['severity']} | {decision['reason']}")
else:
print(f"[IGNORE] {event.event_id} | {decision['reason']}")
def consumer_worker():
while True:
try:
event = event_bus.get(timeout=5)
except queue.Empty:
break
try:
decision = classify_event(event.payload)
route_action(event, decision)
except Exception as e:
print(f"[ERROR] {event.event_id}: {e}")
finally:
event_bus.task_done()
Step 5: Wire the Producer and Run Loop
Finally, I assemble everything into a single runnable script. I feed the queue and start the consumer. You can spawn a pool of workers because Oxlo.ai has no cold starts on popular models.
import json
import queue
import threading
from dataclasses import dataclass
from typing import Dict, Any
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are an event classifier for a platform operations pipeline.
Analyze the incoming event payload and return a JSON object with exactly these keys:
- severity: one of "critical", "warning", "info"
- reason: a one-sentence explanation
- action: one of "page_oncall", "create_ticket", "ignore"
- remediation: a concrete shell command or configuration fix if applicable, otherwise "none"
Respond with JSON only. Do not wrap the output in markdown fences."""
@dataclass
class PlatformEvent:
event_id: str
source: str
payload: Dict[str, Any]
event_bus = queue.Queue(maxsize=100)
def produce_sample_events():
samples = [
PlatformEvent("evt-1", "web-check", {"status": 500, "latency_ms": 12000, "region": "us-east-1", "message": "Database connection timeout"}),
PlatformEvent("evt-2", "cron", {"status": 200, "latency_ms": 45, "region": "eu-west-1", "message": "Nightly backup completed successfully"}),
PlatformEvent("evt-3", "api-gateway", {"status": 503, "latency_ms": 800, "region": "ap-south-1", "message": "Rate limit exceeded on /v1/orders"}),
]
for evt in samples:
event_bus.put(evt)
def classify_event(payload: dict) -> dict:
user_message = json.dumps(payload, 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=256,
)
raw = response.choices[0].message.content.strip()
return json.loads(raw)
def route_action(event: PlatformEvent, decision: dict):
action = decision.get("action")
if action == "page_oncall":
print(f"[PAGER] {event.event_id} | severity={decision['severity']} | {decision['reason']}")
elif action == "create_ticket":
print(f"[TICKET] {event.event_id} | severity={decision['severity']} | {decision['reason']}")
else:
print(f"[IGNORE] {event.event_id} | {decision['reason']}")
def consumer_worker():
while True:
try:
event = event_bus.get(timeout=5)
except queue.Empty:
break
try:
decision = classify_event(event.payload)
route_action(event, decision)
except Exception as e:
print(f"[ERROR] {event.event_id}: {e}")
finally:
event_bus.task_done()
if __name__ == "__main__":
produce_sample_events()
thread = threading.Thread(target=consumer_worker)
thread.start()
event_bus.join()
thread.join()
print("Event-driven pipeline completed.")
Run It
Replace YOUR_OXLO_API_KEY in the script, then execute:
python pipeline.py
Example output:
[PAGER] evt-1 | severity=critical | Database timeout indicates infrastructure failure requiring immediate human intervention.
[TICKET] evt-3 | severity=warning | Rate limiting is service degradation that needs capacity planning but does not wake the on-call.
[IGNORE] evt-2 | severity=info | Successful backup is expected behavior.
Event-driven pipeline completed.
Next Steps
Swap the in-memory queue for Redis Streams or Kafka to survive restarts and scale horizontally. You can also add an outbox table to track every Oxlo.ai request ID against the event ID for audit trails.
If your payloads grow, remember that Oxlo.ai charges per request, not per token. Feeding large stack traces or 1M context windows into models like DeepSeek V4 Flash or Kimi K2.6 does not change the cost structure, which makes long-context enrichment predictable.
Top comments (0)