shashank ms

Posted on Sep 6

Building Edge AI Applications with LLMs: A Step-by-Step Guide

#engineering #oxlo #ai

We are building a headless Edge Telemetry Diagnostic Agent that runs on a gateway device, tails local system logs, and uses an LLM to classify anomalies and suggest remediation. It is designed for teams managing remote edge nodes who need a tiny local footprint but heavy reasoning handled in the cloud. Because Oxlo.ai offers request-based pricing and full OpenAI SDK compatibility, we can ship this with no custom client code and predictable costs even when log payloads grow.

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 local log file to monitor. We will use /var/log/syslog as an example.

Step 1: Configure the Oxlo.ai client

First we initialize the OpenAI-compatible client pointing at Oxlo.ai. I keep the API key in the environment so the agent can be deployed via systemd or Docker without hard-coded secrets.

import os
from openai import OpenAI

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

Step 2: Tail and window local logs

Edge devices generate verbose logs. We read the last 100 lines and truncate to a 4,000 character window so we stay well within context limits while still giving the model enough signal.

def get_log_window(path="/var/log/syslog", lines=100, max_chars=4000):
    try:
        with open(path, "r") as f:
            raw = f.readlines()
        tail = raw[-lines:]
        block = "".join(tail)
        if len(block) > max_chars:
            block = block[-max_chars:]
        return block
    except FileNotFoundError:
        return "No log file found at {}".format(path)

Step 3: Define the system prompt

The system prompt constrains the model to emit strict JSON with severity, root cause, recommended action, and confidence. I use Llama 3.3 70B because its general-purpose reasoning is reliable for structured system administration tasks.

SYSTEM_PROMPT = """You are an edge site reliability engineer.
Analyze the provided system logs and return a single JSON object with no markdown formatting.
Use exactly these keys:
- severity: one of INFO, WARNING, CRITICAL
- root_cause: a 20-word max description
- recommended_action: a concrete shell command or configuration change
- confidence: a float between 0 and 1

If logs look healthy, set severity to INFO and root_cause to "Nominal".
"""

Step 4: Send telemetry to Oxlo.ai for structured analysis

We pack the log window into the user message and call Oxlo.ai. Because Oxlo.ai uses flat request-based pricing, sending a large log payload does not increase cost the way token-based providers would. This makes it practical to batch chunky edge telemetry into a single diagnostic request. See https://oxlo.ai/pricing for plan details.

import json

def diagnose(log_window: str):
    if not log_window.strip():
        return None

    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": log_window},
        ],
    )

    raw = response.choices[0].message.content.strip()
    raw = raw.removeprefix("

```json").removeprefix("```

").removesuffix("

```

").strip()
    return json.loads(raw)

Step 5: Act on the diagnosis locally

The edge agent should do more than print text. If confidence is high and severity is CRITICAL, we append the result to a local alerts file that a parent monitoring system can scrape.

import datetime

ALERTS_PATH = "/tmp/edge_alerts.jsonl"

def handle_diagnosis(diag: dict):
    if diag is None:
        return

    print(f"[{diag['severity']}] {diag['root_cause']} (confidence: {diag['confidence']})")

    if diag["severity"] == "CRITICAL" and diag["confidence"] >= 0.85:
        alert = {
            "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
            "severity": diag["severity"],
            "root_cause": diag["root_cause"],
            "action": diag["recommended_action"],
        }
        with open(ALERTS_PATH, "a") as f:
            f.write(json.dumps(alert) + "\n")
        print(f"Alert persisted to {ALERTS_PATH}")

Step 6: Package the agent as a polling service

Finally, we wire everything into a tight loop that wakes every 60 seconds. On memory-constrained edge nodes, this keeps the local footprint tiny while offloading reasoning to Oxlo.ai.

import time

def main():
    while True:
        window = get_log_window()
        diag = diagnose(window)
        handle_diagnosis(diag)
        time.sleep(60)

if __name__ == "__main__":
    main()

Run it

Save the full script as edge_agent.py, export your key, and run it against a live log file or a synthetic sample.

export OXLO_API_KEY="oxlo_xxxxxxxx"
python edge_agent.py

With a synthetic log containing disk and SSH anomalies, you should see structured output like this:

[INFO] Nominal (confidence: 0.94)
[WARNING] Repeated SSH authentication failures (confidence: 0.89)
[CRITICAL] Disk partition /dev/mmcblk0p2 at 98 percent capacity (confidence: 0.96)
Alert persisted to /tmp/edge_alerts.jsonl

Wrap-up and next steps

This agent is already useful as a standalone edge monitor, but two concrete extensions make it production-ready. First, replace the local JSONL file with a lightweight MQTT publisher so upstream systems receive push notifications instead of polling a file. Second, if your fleet spans regions with non-English kernel messages, swap the model to qwen-3-32b via Oxlo.ai. Its multilingual reasoning handles mixed-language logs without any code changes beyond the model string.

DEV Community