Anomaly detection usually relies on statistical thresholds that miss operational context. In this tutorial, we will build a small Python tool that feeds server metrics into an LLM running on Oxlo.ai, letting the model flag outliers and explain them in plain language. It is useful for SREs and data engineers who want semantic reasoning on top of their time-series data.
What you'll need
Python 3.10 or newer, the OpenAI SDK, and pandas. You will also need an Oxlo.ai API key from https://portal.oxlo.ai. The free tier gives you 60 requests per day, which is enough to follow along.
pip install openai pandasStep 1: Prepare sample metrics
We will work with an in-memory CSV so you do not need external files. The data contains CPU, memory, and latency readings for a web service, with two injected spikes.
import pandas as pd
import json
from io import StringIO
csv_data = """timestamp,cpu_percent,memory_percent,latency_ms
2024-01-15T08:00:00,45,62,120
2024-01-15T08:05:00,48,63,125
2024-01-15T08:10:00,85,70,340
2024-01-15T08:15:00,47,61,118
2024-01-15T08:20:00,50,64,122
2024-01-15T08:25:00,92,75,520
2024-01-15T08:30:00,46,62,119
2024-01-15T08:35:00,49,63,121"""
df = pd.read_csv(StringIO(csv_data))
print(f"Loaded {len(df)} rows")
print(df.head())Step 2: Define the system prompt
The prompt tells the model to act as an SRE analyst and return strict JSON. Keep it in a top-level constant so it is easy to iterate on.
SYSTEM_PROMPT = """You are an SRE anomaly detection assistant. Analyze the provided server metrics window and determine if the final row contains an anomaly compared to the baseline of earlier rows.
Respond ONLY with a JSON object in this exact format:
{
"anomaly_detected": true or false,
"severity": "low", "medium", or "high",
"metric": "name of the primary metric involved",
"reason": "one sentence explanation"
}
Be concise. Base your judgment on relative spikes, not absolute thresholds."""Step 3: Build the Oxlo.ai client
We point the OpenAI SDK at Oxlo.ai and add a helper that converts a pandas DataFrame into a text table, calls the model, and parses the JSON response.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def analyze_window(window_df: pd.DataFrame) -> dict:
# Build a simple text table
lines = [" | ".join(window_df.columns)]
lines.append("-" * 40)
for _, row in window_df.iterrows():
lines.append(" | ".join(str(v) for v in row))
table = "\n".join(lines)
user_message = f"Analyze these metrics:\n\n{table}"
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,
)
content = response.choices[0].message.content.strip()
if content.startswith("
```"):
content = content.split("```
")[1].replace("json", "").strip()
return json.loads(content)Step 4: Run a sliding-window analysis
Instead of judging one row at a time, we slide a three-row window across the dataset so the model sees recent baseline context before evaluating the latest reading.
window_size = 3
results = []
for i in range(window_size, len(df) + 1):
window = df.iloc[i - window_size:i]
result = analyze_window(window)
result["timestamp"] = window.iloc[-1]["timestamp"]
results.append(result)
print(f"Row {i-1}: {result}")
anomalies = [r for r in results if r["anomaly_detected"]]
print(f"\nTotal anomalies detected: {len(anomalies)}")Run it
Save the completed script as anomaly_detector.py, replace YOUR_OXLO_API_KEY, and run it. You should see output similar to this:
Loaded 8 rows
Row 2: {'anomaly_detected': True, 'severity': 'high', 'metric': 'latency_ms', 'reason': 'CPU and latency spiked well above the prior baseline.', 'timestamp': '2024-01-15T08:10:00'}
Row 3: {'anomaly_detected': False, 'severity': 'low', 'metric': 'cpu_percent', 'reason': 'All metrics returned to normal baseline levels.', 'timestamp': '2024-01-15T08:15:00'}
Row 5: {'anomaly_detected': True, 'severity': 'high', 'metric': 'latency_ms', 'reason': 'Severe latency spike to 520ms with elevated CPU.', 'timestamp': '2024-01-15T08:25:00'}
Row 7: {'anomaly_detected': False, 'severity': 'low', 'metric': 'cpu_percent', 'reason': 'Readings stabilized after the previous spike.', 'timestamp': '2024-01-15T08:35:00'}
Total anomalies detected: 2Next steps
Pipe live data into the script by replacing the CSV with a query to your metrics store, such as Prometheus or Datadog, and schedule the detector on a five-minute cron job. If you want to analyze longer history in a single shot, swap llama-3.3-70b for kimi-k2.6 on Oxlo.ai, which supports a 131K context window. Because Oxlo.ai uses flat per-request pricing, sending larger metric tables does not increase your API cost the way token-based providers would.
Top comments (0)