LLM Edge Computing 101

We are going to build a lightweight edge log analyzer that runs on a local device and uses an LLM to turn raw system logs into structured alerts. This is useful for anyone monitoring remote servers, IoT gateways, or factory edge nodes where you want to keep raw data local and only emit compressed intelligence. Because edge workloads often process large log chunks, Oxlo.ai's flat per-request pricing makes it a natural fit: the cost stays the same whether you summarize 500 tokens or 10,000 tokens in a single request, and you can see current plans at https://oxlo.ai/pricing.

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 local log file to analyze. We will use /var/log/syslog by default, but any text log works.
• Outbound HTTPS access to https://api.oxlo.ai. No inbound ports are required, which keeps firewall rules simple.

Step 1: Set up the edge client

I use the OpenAI SDK because it keeps the client code tiny, which matters on a Raspberry Pi or embedded gateway with limited disk space. Pointing it at Oxlo.ai is a single line change, and because Oxlo.ai serves popular models with no cold starts, the first request after an idle period returns immediately. That is important for edge agents that may sleep for hours between log bursts. Create a new file named edge_agent.py and add the imports and client setup.

import os
import json
import hashlib
from openai import OpenAI

OXLO_API_KEY = os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")

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

Step 2: Ingest and chunk local logs

Edge nodes generate logs continuously, but loading an entire multi-gigabyte file into RAM will crash a small device. I stream the file line by line and yield fixed-size chunks so memory stays flat. A chunk of 40 to 50 lines is usually enough context for the model to spot patterns without turning a single inference into multiple billed requests. I also set errors="ignore" because edge logs often contain binary noise from sensors or serial devices.

def read_log_chunks(path: str, lines_per_chunk: int = 50):
    if not os.path.exists(path):
        # Fallback sample so you can test without a real syslog file.
        sample = [
            "Jan 10 09:12:01 edge-node kernel: [UFW BLOCK] IN=eth0 OUT= MAC=...",
            "Jan 10 09:12:15 edge-node systemd[1]: Started OpenBSD Secure Shell server.",
            "Jan 10 09:13:42 edge-node python3[2041]: Connection reset by peer",
            "Jan 10 09:14:01 edge-node CRON[2050]: (root) CMD (cd / && run-parts --report /etc/cron.hourly)",
        ]
        yield "\n".join(sample)
        return

    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        buffer = []
        for line in f:
            buffer.append(line.rstrip())
            if len(buffer) >= lines_per_chunk:
                yield "\n".join(buffer)
                buffer = []
        if buffer:
            yield "\n".join(buffer)

Step 3: Write the system prompt

The system prompt is the only part of the agent that needs to stay constant. I force JSON output so that downstream automation on the edge node can parse the result with the standard library json module. No markdown, no explanations, just structured data. Keeping the prompt short also reduces payload size, which is good practice even on a flat per-request plan.

SYSTEM_PROMPT = """You are an edge log analyzer running on a constrained device.
Your job is to read a chunk of system logs and emit a structured JSON assessment.
Follow these rules exactly:
- severity: one of INFO, WARNING, CRITICAL
- summary: a one-sentence description of what happened
- tags: an array of relevant keywords
- recommended_action: a short ops command or fix if applicable

Output only valid JSON. Do not wrap it in markdown."""

Step 4: Build the classification agent

This is the inference call. I use llama-3.3-70b because it is reliable for structured extraction and runs without cold starts on Oxlo.ai. Because Oxlo.ai charges per request rather than per token, I can stuff an entire 40-line log chunk into the prompt without worrying about ballooning costs. If you are running on a battery-powered device and need lower latency, you could swap in qwen-3-32b for faster turnaround with only a small accuracy trade-off on simple log patterns. The response is parsed as JSON and returned as a native Python dictionary.

def analyze_chunk(chunk: str) -> dict:
    from openai import OpenAI

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

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

    raw = response.choices[0].message.content
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {
            "severity": "UNKNOWN",
            "summary": raw,
            "tags": ["parse-error"],
            "recommended_action": "Inspect raw output manually.",
        }

Step 5: Buffer and deduplicate on the edge

Bandwidth at remote sites is often metered or slow. I add a simple SHA-256 deduplication cache so that identical log patterns are not sent to the API twice. In my last deployment, roughly 60 percent of log chunks were repeats of cron jobs and health checks, so this filter cut API usage by more than half. This pairs naturally with Oxlo.ai's flat per-request pricing: once we pay to analyze a pattern, we cache the verdict locally and avoid both duplicate charges and unnecessary network traffic. On a device with very little RAM you could replace the set with an LRU cache from functools to cap memory usage.

seen_hashes = set()

def is_new_chunk(chunk: str) -> bool:
    h = hashlib.sha256(chunk.encode("utf-8")).hexdigest()[:16]
    if h in seen_hashes:
        return False
    seen_hashes.add(h)
    return True

Step 6: Run the main loop

The main loop ties everything together. It streams chunks from the log file, skips anything already seen, sends new chunks to Oxlo.ai, and prints the structured result. I keep the loop single-threaded to minimize CPU and power draw on the edge device. I also wrap the analysis call in a broad try/except so that a temporary network blip does not kill a long-running agent. On a real edge node you would replace the print statement with a webhook to your central monitoring stack or a local MQTT publish.

if __name__ == "__main__":
    LOG_PATH = "/var/log/syslog"

    for chunk in read_log_chunks(LOG_PATH, lines_per_chunk=40):
        if not is_new_chunk(chunk):
            continue

        try:
            result = analyze_chunk(chunk)
            print(json.dumps(result, indent=2))
        except Exception as e:
            print(json.dumps({"error": str(e), "severity": "UNKNOWN"}))

Run it

Set your API key and run the agent. If you do not have a real syslog file, the fallback sample data from Step 2 will still exercise the full pipeline. You should see a JSON object printed within a second or two because Oxlo.ai does not queue or cold-start the model.

export OXLO_API_KEY="sk-oxlo.ai-..."
python3 edge_agent.py

When I run this against the sample data, the output looks like this:

{
  "severity": "WARNING",
  "summary": "Connection reset by peer detected in application logs.",
  "tags": ["network", "python", "connection"],
  "recommended_action": "Check firewall rules and application timeout settings."
}

Next steps

Two concrete ways to harden this for production. First, replace the static file reader with a streaming tail -f implementation using Python's subprocess module so the agent runs continuously as a systemd service and processes logs in real time. Second, add a local SQLite cache that stores the last 24 hours of alerts and only forwards CRITICAL severity events to your cloud dashboard, which keeps bandwidth usage near zero for routine noise while preserving full local history.

If you want to scale this pattern across a larger fleet, package the script in a Docker container and deploy it via Balena or Raspberry Pi Imager. Oxlo.ai's OpenAI-compatible endpoint means you do not need to change client code if you later switch models or upgrade from the free tier to a Pro plan. You can explore the full model catalog and request-based pricing at https://oxlo.ai/pricing.