Getting Started with Edge AI and LLM

#learnai #oxlo #ai

We are going to build a lightweight edge log analyzer that tails a local application log, batches recent error lines, and sends them to an LLM for root-cause analysis. It is useful for developers and DevOps engineers who want immediate, intelligent summaries from noisy log streams without running heavy infrastructure. The agent runs entirely on your machine and only ships filtered text to Oxlo.ai, so data stays local until it is needed.

What you'll need

Python 3.10 or newer installed on your edge device or laptop. An Oxlo.ai API key from https://portal.oxlo.ai. The OpenAI SDK installed with pip install openai.

Step 1: Generate sample logs

Create a local log file that mixes INFO, WARNING, and ERROR lines. This simulates the edge data source our agent will monitor.

import random
import time
from datetime import datetime

LOG_PATH = "app.log"
LEVELS = ["INFO", "WARNING", "ERROR"]
MESSAGES = {
    "INFO": ["Request completed", "Cache hit", "User logged in"],
    "WARNING": ["Retry attempt 2", "High memory usage", "Slow query detected"],
    "ERROR": ["Connection timeout to db:3306", "NullPointerException in payment.py:42", "Disk full on /var/log"]
}

with open(LOG_PATH, "w") as f:
    for _ in range(200):
        level = random.choices(LEVELS, weights=[70, 20, 10])[0]
        msg = random.choice(MESSAGES[level])
        line = f"{datetime.now().isoformat()} [{level}] {msg}\n"
        f.write(line)
        time.sleep(0.001)

print(f"Wrote {LOG_PATH}")

Step 2: Filter logs locally

We will write a small edge-side preprocessor that tails the file and keeps only lines with ERROR or WARNING tags. Stripping noise before the API call keeps latency low and respects the flat request pricing on Oxlo.ai.

from collections import deque

def tail_lines(path, n=20):
    with open(path, "r") as f:
        return list(deque(f, maxlen=n))

def extract_issues(lines):
    filtered = [l for l in lines if "[ERROR]" in l or "[WARNING]" in l]
    return "\n".join(filtered) if filtered else ""

sample = tail_lines("app.log", 30)
batch = extract_issues(sample)
print(batch)

Step 3: Define the agent prompt

The system prompt instructs the model to return structured JSON. Locking the output format makes it easy to parse the diagnosis downstream without fragile regex.

SYSTEM_PROMPT = """You are an on-call site reliability engineer.
Analyze the provided log lines and produce a concise diagnosis.
Respond in valid JSON with exactly these keys:
- summary: one sentence describing the problem
- severity: LOW, MEDIUM, or HIGH
- recommended_fix: a concrete, actionable remediation step
If there are no issues, set severity to LOW and summary to 'No actionable issues found'."""

Step 4: Call Oxlo.ai for diagnosis

We send the filtered batch to Oxlo.ai using the standard OpenAI client. Because Oxlo.ai charges per request rather than per token, packing thirty lines into one call costs the same as sending a single sentence. That makes long-context log dumps practical.

import json
from openai import OpenAI

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

def diagnose(batch):
    if not batch.strip():
        return {"summary": "No actionable issues found", "severity": "LOW", "recommended_fix": "Monitor logs"}
    
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": batch},
        ],
        response_format={"type": "json_object"},
    )
    
    raw = response.choices[0].message.content
    return json.loads(raw)

result = diagnose(batch)
print(json.dumps(result, indent=2))

Step 5: Assemble the monitor

The final script combines local tailing, filtering, and the Oxlo.ai diagnosis loop into one continuous edge agent. It deduplicates identical windows so we do not waste requests on unchanged logs.

import json
import time
from collections import deque
from openai import OpenAI

LOG_PATH = "app.log"
POLL_INTERVAL = 5
SYSTEM_PROMPT = """You are an on-call site reliability engineer.
Analyze the provided log lines and produce a concise diagnosis.
Respond in valid JSON with exactly these keys:
- summary: one sentence describing the problem
- severity: LOW, MEDIUM, or HIGH
- recommended_fix: a concrete, actionable remediation step
If there are no issues, set severity to LOW and summary to 'No actionable issues found'."""

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

def tail_lines(path, n=20):
    with open(path, "r") as f:
        return list(deque(f, maxlen=n))

def extract_issues(lines):
    filtered = [l for l in lines if "[ERROR]" in l or "[WARNING]" in l]
    return "\n".join(filtered) if filtered else ""

def diagnose(batch):
    if not batch.strip():
        return {"summary": "No actionable issues found", "severity": "LOW", "recommended_fix": "Monitor logs"}
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": batch},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

def run_monitor():
    seen = set()
    while True:
        lines = tail_lines(LOG_PATH, 30)
        fingerprint = hash(tuple(lines))
        if fingerprint in seen:
            time.sleep(POLL_INTERVAL)
            continue
        seen.add(fingerprint)
        batch = extract_issues(lines)
        if batch:
            result = diagnose(batch)
            print(f"[{result['severity']}] {result['summary']}")
            print(f"Fix: {result['recommended_fix']}\n")
        else:
            print("No issues in latest window.")
        time.sleep(POLL_INTERVAL)

if __name__ == "__main__":
    run_monitor()

Run it

Start the agent in one terminal, then append a new error to app.log in another to see the live diagnosis.

$ python agent.py
[MEDIUM] Multiple connection timeouts to db:3306 detected in latest window.
Fix: Check network path to db:3306 and verify the database service is accepting connections.

If you want to test against a reasoning model, swap the model string to deepseek-v3.2 or qwen-3-32b in the client call. Both are available on Oxlo.ai with the same SDK and base URL, so the only change is the model name.

Next steps

Hook the agent into a real log directory by changing LOG_PATH, or add a local SQLite cache to store historical diagnoses so the LLM can compare across time windows. If you move this to a fleet of edge devices, the flat per-request pricing on Oxlo.ai keeps costs predictable even when individual devices ship large batched payloads. You can review plans at https://oxlo.ai/pricing.

DEV Community