We are going to build a temporal log anomaly detector that maintains a sliding window of server events and flags cascading failures before they escalate. This kind of high-temporal dependency task is exactly where long-context reasoning pays off, and it is a workload that gets expensive fast on token-based providers. Oxlo.ai's request-based pricing removes that penalty, so we can ship a large-context agent without watching input tokens drain the budget.
What you'll need
- Python 3.10+
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
1. Initialize the Oxlo.ai client and the memory buffer
Every temporal agent needs working memory. I use a deque with a maxlen of 40 so the model always sees the most recent events without unbounded growth.
import os
import json
import time
import random
from datetime import datetime, timedelta
from collections import deque
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)
BUFFER = deque(maxlen=40)
2. Generate synthetic logs with a buried failure sequence
To test whether the model catches dependencies across time, I inject a known anomaly pattern: a slow memory leak that leads to an out-of-memory kill. The rest of the stream is normal traffic noise.
def generate_logs(n=50):
logs = []
base_time = datetime.utcnow()
for i in range(n):
ts = base_time + timedelta(seconds=i * 2)
if i < 35:
level = random.choice(["INFO", "INFO", "INFO", "WARN"])
msg = random.choice([
"GET /health 200",
"POST /api/v1/login 200",
"DB connection pooled",
"Cache hit ratio 94%",
])
if level == "WARN":
msg = "High memory usage: 62%"
else:
if i == 35:
level, msg = "WARN", "High memory usage: 87%"
elif i == 36:
level, msg = "WARN", "High memory usage: 91%"
elif i == 37:
level, msg = "WARN", "High memory usage: 96%; considering OOM kill"
else:
level, msg = "ERROR", "OutOfMemoryError: Killed process 4127 (java)"
logs.append({
"timestamp": ts.isoformat() + "Z",
"level": level,
"message": msg,
})
return logs
3. Write the system prompt for temporal reasoning
The system prompt is the only place where we teach the model to treat the log list as a timeline. I explicitly ask it to look for deltas between timestamps, repeated warnings, and sequences that precede a failure.
SYSTEM_PROMPT = """You are a site-reliability engineer monitoring a timeline of server logs.
Analyze the events in chronological order. Pay close attention to time deltas, repeated warnings, and sequences that lead to failures.
If you detect an anomaly that is explained by earlier events in the window, set anomaly_detected to true and explain the causal chain.
Return ONLY valid JSON with this schema:
{
"anomaly_detected": bool,
"severity": "none" | "warning" | "critical",
"summary": string,
"root_cause": string
}
"""4. Format the sliding window into a single request
I format the buffer as a plain text timeline before sending it. Because Oxlo.ai charges a flat rate per request, I can afford to ship the entire 40-event window every cycle. On token-based providers, this long-context polling would scale linearly with window size, but here the cost stays predictable.
def format_window(buffer):
lines = []
for entry in buffer:
lines.append(
f"{entry['timestamp']} [{entry['level']}] {entry['message']}"
)
return "\n".join(lines)
5. Query the model and enforce JSON output
I use Kimi K2.6 because its 131K context window and advanced reasoning handle long event sequences well. I also set response_format to JSON so the agent returns structured data we can act on programmatically.
def analyze_buffer():
user_message = format_window(BUFFER)
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
response_format={"type": "json_object"},
temperature=0.1,
)
return json.loads(response.choices[0].message.content)
6. Run the streaming detection loop
The main loop pushes a new log into the buffer every tenth of a second and calls the agent every 10 events. In production you would read from Kafka or syslog, but this loop proves the concept.
if __name__ == "__main__":
stream = generate_logs(50)
for idx, log in enumerate(stream):
BUFFER.append(log)
print(f" [{log['level']}] {log['message']}")
if (idx + 1) % 10 == 0 and len(BUFFER) >= 10:
result = analyze_buffer()
print(json.dumps(result, indent=2))
time.sleep(0.1)
Run it
Save the full script as temporal_agent.py, export your key, and run it. You should see a clean JSON verdict once the memory-leak sequence enters the window.
export OXLO_API_KEY="sk-oxlo.ai-..."
python temporal_agent.py
When the sliding window contains the escalating memory warnings, the output will look something like this:
{
"anomaly_detected": true,
"severity": "critical",
"summary": "Memory usage escalated from 62% to 96% over several minutes, followed by an OOM kill.",
"root_cause": "Repeated high-memory warnings were ignored until the kernel terminated the JVM process."
}Wrap-up and next steps
This agent works because the model sees the full temporal context, not just a single error line. Two concrete next steps: pipe real logs from journald or Fluent Bit into the buffer, and add a webhook that pages the on-call engineer when anomaly_detected flips to true. If you are evaluating providers for this workload, compare your current token bill against Oxlo.ai's flat per-request pricing at https://oxlo.ai/pricing. For long-context monitoring agents, the difference is usually significant.
Top comments (0)