Getting Started with Edge AI and LLMs

#learnai #oxlo #ai

We're building a lightweight edge log analyzer that preprocesses system logs on a low-power device and forwards condensed error batches to an LLM for structured incident reports. It helps DevOps teams running remote edge servers who need actionable alerts without shipping gigabytes of raw logs to the cloud.

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
A sample log file (we will generate one in Step 1)

Step 1: Generate synthetic edge logs

We need a sample log stream that mimics an edge gateway. This script creates a mixed log file with info lines, warnings, and repeated errors so we have realistic noise to filter.

import random
from datetime import datetime, timedelta

LOG_PATH = "edge_gateway.log"

def generate_logs(path: str, lines: int = 500):
    levels = ["INFO", "INFO", "INFO", "WARN", "ERROR", "ERROR", "FATAL"]
    msgs = [
        "Packet forwarded to upstream",
        "MQTT broker latency 120ms",
        "Temperature sensor 0x4A unreachable",
        "Disk usage 91% on /var/log",
        "Connection timeout to cloud relay",
        "Unhandled exception in worker thread",
        "Kernel panic on node 0"
    ]
    with open(path, "w") as f:
        t = datetime.now() - timedelta(hours=6)
        for i in range(lines):
            t += timedelta(seconds=random.randint(1, 30))
            lvl = random.choice(levels)
            msg = random.choice(msgs)
            f.write(f"[{t.isoformat()}] [{lvl}] {msg}\n")

if __name__ == "__main__":
    generate_logs(LOG_PATH)
    print(f"Wrote {path}")

Step 2: Build the edge filter

Before we burn bandwidth or tokens, we filter for ERROR and FATAL lines locally and collapse repeated messages. This lightweight preprocessing is what makes the workflow edge-friendly.

from collections import Counter
import re

def filter_log(path: str, max_lines: int = 50) -> str:
    """Return a condensed block of the most frequent error signatures."""
    pattern = re.compile(r"\[(.*?)\] \[(ERROR|FATAL)\] (.*)")
    counts = Counter()
    with open(path, "r") as f:
        for line in f:
            m = pattern.search(line)
            if m:
                counts[m.group(3)] += 1

    top = counts.most_common(max_lines)
    if not top:
        return "No errors found."

    out = []
    for msg, cnt in top:
        out.append(f"(x{cnt}) {msg}")
    return "\n".join(out)

Step 3: Define the agent system prompt

The prompt tells the LLM to act as an on-call SRE and return strictly formatted JSON. Keeping the output structured makes it easy to pipe into other tools.

SYSTEM_PROMPT = """You are an edge infrastructure SRE assistant.
Analyze the provided log summary and produce a JSON object with exactly these keys:
- summary: a one-sentence description of the incident
- severity: LOW, MEDIUM, HIGH, or CRITICAL
- root_cause: the most likely cause
- next_steps: a list of concrete remediation actions
Do not include markdown formatting or explanation outside the JSON."""

Step 4: Wire up the Oxlo.ai client

We use the OpenAI SDK with Oxlo.ai's base URL and Llama 3.3 70B. Oxlo.ai's request-based pricing is a good fit here because we can stuff a large filtered batch into one API call without the cost scaling by token count.

import os
from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"))

def analyze_batch(batch_text: str) -> str:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": batch_text},
        ],
    )
    return response.choices[0].message.content

Step 5: Assemble the full pipeline

This ties the filter, prompt, and client together into a single callable agent. Running it reads the local log, distills the errors, and returns a structured incident report.

import json

def run_agent(log_path: str = "edge_gateway.log"):
    batch = filter_log(log_path)
    if "No errors found." in batch:
        print("Nothing to report.")
        return

    raw = analyze_batch(batch)
    raw = raw.replace("

```json", "").replace("```

", "").strip()
    try:
        report = json.loads(raw)
        print(json.dumps(report, indent=2))
    except json.JSONDecodeError:
        print("Raw response:", raw)

if __name__ == "__main__":
    run_agent()

Run it

Execute the pipeline against our synthetic log file and inspect the structured report. You should see a JSON object with a severity rating and actionable next steps.

$ python step1_generate.py
Wrote edge_gateway.log

$ export OXLO_API_KEY="sk-oxlo.ai-..."
$ python agent.py
{
  "summary": "Multiple connection timeouts and unhandled worker exceptions are saturating the edge gateway.",
  "severity": "HIGH",
  "root_cause": "Intermittent upstream network relay failure causing cascading worker thread crashes.",
  "next_steps": [
    "Restart the cloud relay service and verify TLS certificate validity.",
    "Check kernel logs for storage controller faults.",
    "Reduce MQTT keep-alive interval to detect dropped connections faster."
  ]
}

Wrap up

The agent now gives you structured incident reports from noisy edge logs without shipping the raw stream to the cloud. Two concrete next steps: add a local SQLite cache that stores error fingerprints so you only pay for one Oxlo.ai request per unique signature per hour, or wire the JSON output to a webhook for PagerDuty integration.

DEV Community