Building Edge AI Applications with LLM: Best Practices and Use Cases

#learnai #oxlo #ai

We are building an edge log anomaly detector. It runs on a lightweight gateway, tails local syslog files, and sends batched entries to Oxlo.ai for instant classification. This helps ops teams catch infrastructure issues at remote sites without shipping raw logs back to a central SIEM.

What you'll need

Prerequisites are minimal. You need Python 3.10 or newer, the OpenAI SDK, and an Oxlo.ai API key.

Python 3.10+
pip install openai
An Oxlo.ai API key from https://portal.oxlo.ai
A local log file to monitor, such as /var/log/syslog or a sample edge.log

Step 1: Scaffold the local log watcher

First, we need a generator that tails a log file and yields new lines as they appear. We will also add a small rolling buffer so we can batch entries before sending them to the API.

import time
from collections import deque

def tail_log(filepath):
    with open(filepath, "r") as f:
        f.seek(0, 2)
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.1)
                continue
            yield line.strip()

class LogBuffer:
    def __init__(self, maxlen=20):
        self.buffer = deque(maxlen=maxlen)

    def push(self, line):
        self.buffer.append(line)

    def flush(self):
        batch = list(self.buffer)
        self.buffer.clear()
        return batch

Step 2: Sanitize and format batches

Raw log lines carry repetitive timestamps and hostnames that waste context window space. We strip redundant metadata and format the batch as a clean markdown list so the model can scan it quickly.

import re

def sanitize(line):
    line = re.sub(r"\w{3}\s+\d{1,2}\s\d{2}:\d{2}:\d{2}", "", line)
    line = re.sub(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", "", line)
    return line.strip()

def format_batch(lines):
    cleaned = [sanitize(l) for l in lines if l.strip()]
    if not cleaned:
        return ""
    return "Log batch:\n" + "\n".join(f"- {l}" for l in cleaned)

Step 3: Initialize the Oxlo.ai client and system prompt

We point the OpenAI SDK at Oxlo.ai. For edge deployments where latency matters, Llama 3.3 70B is a solid default, and the flat per-request pricing keeps costs predictable even when verbose stack traces show up. The system prompt forces the model to return strict JSON so our edge script can act on the result without fragile regex parsing.

SYSTEM_PROMPT = """You are an edge log analyzer. Evaluate the provided log batch.
Return ONLY a JSON object with no markdown formatting and no explanation.
Use this exact schema:
{
  \"anomaly_detected\": true or false,
  \"severity\": \"low\", \"medium\", or \"high\",
  \"summary\": \"One sentence describing what happened.\",
  \"recommended_action\": \"One sentence suggesting a fix.\"
}
If the logs look normal, set anomaly_detected to false and severity to \"low\"."""

from openai import OpenAI

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

def classify_batch(user_message):
    if not user_message:
        return None
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
    )
    return response.choices[0].message.content

Step 4: Add local filtering and run the loop

Edge bandwidth is often metered. We filter out routine noise with a lightweight keyword check before burning an API call. When a batch does contain suspicious tokens, we send it to Oxlo.ai and print the structured result to stdout. In production you might write this to a local MQTT broker or SQLite cache.

import json

SUSPICIOUS_KEYWORDS = ["error", "failed", "unauthorized", "panic", "segfault", "timeout"]

def should_send(batch_text):
    text_lower = batch_text.lower()
    return any(k in text_lower for k in SUSPICIOUS_KEYWORDS)

def main(logfile="edge.log"):
    buf = LogBuffer(maxlen=15)
    for line in tail_log(logfile):
        buf.push(line)
        if len(buf.buffer) >= buf.buffer.maxlen:
            batch = buf.flush()
            text = format_batch(batch)
            if should_send(text):
                raw = classify_batch(text)
                if raw:
                    try:
                        result = json.loads(raw)
                        print(json.dumps(result))
                    except json.JSONDecodeError:
                        print("{\"anomaly_detected\": false, \"note\": \"bad json\"}")
            else:
                print("{\"anomaly_detected\": false, \"severity\": \"low\", \"summary\": \"routine traffic\"}")

if __name__ == "__main__":
    main()

Run it

Create a sample edge.log and append lines to it in another terminal. Run the agent:

python edge_agent.py

When the buffer fills with routine lines, you will see compact pass-through output. When you inject a suspicious line such as kernel: segfault at 0 ip 00007f..., the agent fires an Oxlo.ai request and emits structured JSON like this:

{"anomaly_detected": true, "severity": "high", "summary": "A kernel segfault was recorded indicating a possible memory violation or unstable process.", "recommended_action": "Identify the offending process from the address map and restart or update the associated service."}

Wrap-up and next steps

This pattern keeps heavy inference off the edge device while still letting local logic decide what gets sent upstream. Oxlo.ai fits here because request-based pricing means a sudden spike in verbose stack traces will not inflate your bill the way token-based metering would. See https://oxlo.ai/pricing for plan details. If you want to cut costs during a pilot, swap the model to deepseek-v3.2 on Oxlo.ai's free tier and lower the buffer size to 10. A solid next step is to replace the stdout print with a local SQLite cache and a small FastAPI health endpoint so central orchestration can poll each edge node for its latest status.

DEV Community