We are going to build a lightweight edge log analyzer that runs on a local gateway, batches anomalies, and forwards them to an LLM for root-cause analysis. This pattern works for IoT gateways, remote servers, or any resource-constrained node that needs cloud intelligence without shipping every byte over the wire. Oxlo.ai powers the cloud reasoning layer with flat per-request pricing, so the cost stays predictable even when the context grows.
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 - A sample log file (we will generate one in the first step)
Step 1: Simulate the edge log stream
On a real edge device this would read from /var/log/syslog or an MQTT topic. For this tutorial we will generate a noisy log stream locally. The important part is that filtering happens before any network call, which keeps bandwidth and cost down.
import re
import time
import random
from datetime import datetime
RAW_LOGS = [
"2024-05-20T14:23:05Z ERROR connection timeout to upstream db",
"2024-05-20T14:23:06Z WARN retry attempt 3/5",
"2024-05-20T14:23:10Z CRITICAL disk usage 98% on /var/log",
"2024-05-20T14:23:12Z INFO health check passed",
"2024-05-20T14:23:15Z ERROR failed to write to persistent store",
]
def edge_log_source():
while True:
line = random.choice(RAW_LOGS)
yield f"{datetime.now().isoformat()} {line}"
time.sleep(0.5)
def is_anomaly(line: str) -> bool:
return bool(re.search(r"\b(ERROR|CRITICAL|WARN)\b", line))
if __name__ == "__main__":
for line in edge_log_source():
if is_anomaly(line):
print(f"[EDGE] {line}")
Step 2: Batch events before sending them to the cloud
Sending one API call per log line is wasteful. We will accumulate anomalies in a buffer and flush every 30 seconds or when we hit 5 events. This reduces network overhead and makes each Oxlo.ai request carry more context, which plays well with flat per-request pricing because the cost does not scale with prompt length. You can see the exact rates at https://oxlo.ai/pricing.
from collections import deque
from datetime import datetime, timedelta
BUFFER_SIZE = 5
FLUSH_INTERVAL = timedelta(seconds=30)
class EdgeBuffer:
def __init__(self):
self.events = deque(maxlen=BUFFER_SIZE)
self.last_flush = datetime.now()
def add(self, line: str):
self.events.append(line)
def should_flush(self) -> bool:
if len(self.events) >= BUFFER_SIZE:
return True
if (datetime.now() - self.last_flush) >= FLUSH_INTERVAL and len(self.events) > 0:
return True
return False
def drain(self) -> str:
payload = "\n".join(self.events)
self.events.clear()
self.last_flush = datetime.now()
return payload
buffer = EdgeBuffer()
Step 3: Define the agent system prompt
The system prompt lives on the edge device as a configurable constant. It tells the LLM to act as a site reliability engineer and to return structured output that our local script can parse without extra dependencies.
SYSTEM_PROMPT = """You are a senior SRE monitoring an edge gateway.
Analyze the batched log lines below and produce exactly two lines:
1. A one-sentence severity assessment (SEVERITY: LOW|MEDIUM|HIGH|CRITICAL).
2. A one-sentence recommended action.
If there is nothing actionable, respond with exactly:
SEVERITY: LOW
ACTION: No action required.
"""
Step 4: Wire the buffer to Oxlo.ai
Now we add the Oxlo.ai client. We use the OpenAI SDK with Oxlo.ai's base URL and a model that handles system prompts reliably. I picked llama-3.3-70b because it is a strong general-purpose flagship, but qwen-3-32b or deepseek-v3.2 work just as well.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def analyze_batch(payload: str) -> str:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": payload},
],
)
return response.choices[0].message.content
Step 5: Close the loop with local alerting
The final piece is the main loop. We tail the log, buffer anomalies, flush to Oxlo.ai, and parse the response. If the LLM reports CRITICAL severity, we print a local alert. This is the edge-to-cloud pattern: heavy reasoning happens remotely, but decisions are enforced locally.
def parse_response(text: str) -> tuple[str, str]:
severity = "UNKNOWN"
action = text.strip()
for line in text.splitlines():
if line.startswith("SEVERITY:"):
severity = line.split(":", 1)[1].strip()
elif line.startswith("ACTION:"):
action = line.split(":", 1)[1].strip()
return severity, action
def main():
print("Edge agent starting. Press Ctrl+C to stop.")
for line in edge_log_source():
if is_anomaly(line):
buffer.add(line)
if buffer.should_flush():
payload = buffer.drain()
print(f"\n[FLUSH] Sending {len(payload.splitlines())} lines to Oxlo.ai...")
try:
result = analyze_batch(payload)
severity, action = parse_response(result)
print(f"[RESULT] Severity: {severity}")
print(f"[RESULT] Action: {action}")
if severity == "CRITICAL":
print("!!! LOCAL ALERT: CRITICAL condition detected !!!")
except Exception as e:
print(f"[ERROR] Failed to reach Oxlo.ai: {e}")
if __name__ == "__main__":
main()
Run it
Save everything into a single file named edge_agent.py, replace YOUR_OXLO_API_KEY with your key from https://portal.oxlo.ai, and run:
python edge_agent.py
Example output after a few seconds:
Edge agent starting. Press Ctrl+C to stop.
[EDGE] 2024-05-20T14:23:16.123456 ERROR connection timeout to upstream db
[EDGE] 2024-05-20T14:23:16.623789 WARN retry attempt 3/5
[EDGE] 2024-05-20T14:23:17.124123 CRITICAL disk usage 98% on /var/log
[EDGE] 2024-05-20T14:23:17.624456 ERROR failed to write to persistent store
[EDGE] 2024-05-20T14:23:18.124789 WARN retry attempt 3/5
[FLUSH] Sending 5 lines to Oxlo.ai...
[RESULT] Severity: CRITICAL
[RESULT] Action: Free disk space on /var/log immediately and investigate the upstream database connection timeout.
!!! LOCAL ALERT: CRITICAL condition detected !!!
Next steps
Try swapping llama-3.3-70b for kimi-k2.6 or deepseek-v3.2 in the client call to see which model gives the most actionable alerts for your log format. If you want to keep historical context across flushes, you could index past incidents with Oxlo.ai's embeddings endpoint and query them locally before each batch call.
Top comments (0)