DEV Community

shashank ms
shashank ms

Posted on

Optimizing LLM for Edge Devices: Best Practices and Techniques

We are building a lightweight edge gateway that runs on a resource-constrained device, compresses local sensor logs, and offloads reasoning to Oxlo.ai. This pattern keeps bandwidth low and avoids running a multi-billion-parameter model on battery-powered hardware. If you deploy LLMs to factory floors, retail kiosks, or remote cameras, this is the architecture I actually ship.

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
  • SQLite (included in the Python standard library)

Step 1: Verify connectivity with Oxlo.ai

Before adding compression or caching, confirm the device can reach the API. I use deepseek-v3.2 here because it offers a free tier and responds quickly on low-bandwidth links.

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="deepseek-v3.2",
    messages=[
        {"role": "user", "content": "Say 'Edge online'"},
    ],
)

print(response.choices[0].message.content)

Step 2: Compress raw logs before transmission

Edge networks are slow. We summarize 100 plus lines of syslog into a short paragraph using Oxlo.ai, cutting payload size by roughly 90 percent. Because Oxlo.ai uses flat request-based pricing (see https://oxlo.ai/pricing), this summary step costs the same rate regardless of how long the original logs are.

from openai import OpenAI

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

def compress_logs(raw_logs: str) -> str:
    prompt = (
        "Compress the following device logs into a single paragraph. "
        "Preserve error codes, timestamps, and anomaly patterns. "
        "Discard routine heartbeat messages.\n\n" + raw_logs
    )

    response = client.chat.completions.create(
        model="qwen-3-32b",
        messages=[{"role": "user", "content": prompt}],
    )

    return response.choices[0].message.content.strip()

Step 3: Define the agent system prompt

The system prompt is the only brain the edge device borrows from the cloud. I keep it strict so the model outputs deterministic JSON that a local controller can act on without a heavy parser.

SYSTEM_PROMPT = """You are an edge monitoring agent.
Your inputs are compressed sensor logs from a remote Linux gateway.
Analyze them for anomalies, security events, or hardware degradation.
Respond with valid JSON containing exactly these keys:
  anomaly_detected: boolean
  severity: one of [low, medium, critical]
  recommended_action: string
  affected_service: string
Do not include markdown, explanations, or preamble."""

Step 4: Run structured inference with JSON mode

With the prompt locked, we call Oxlo.ai and force JSON mode. This removes the need to ship a validation library on the edge device.

import json
from openai import OpenAI

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

def diagnose(compressed_logs: str) -> dict:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": compressed_logs},
        ],
        response_format={"type": "json_object"},
    )

    return json.loads(response.choices[0].message.content)

Step 5: Add a local SQLite cache

Rural edge sites drop packets. Caching identical log fingerprints avoids redundant API calls and keeps the device functional when offline. Because Oxlo.ai charges per request, every cached hit is direct savings.

import sqlite3
import hashlib
import time
import json

DB_PATH = "edge_cache.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS cache "
        "(hash TEXT PRIMARY KEY, response TEXT, ts INTEGER)"
    )
    conn.commit()
    conn.close()

def get_cached(key: str):
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        "SELECT response FROM cache WHERE hash = ? AND ts > ?",
        (key, int(time.time()) - 3600),
    ).fetchone()
    conn.close()
    return json.loads(row[0]) if row else None

def set_cache(key: str, value: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "REPLACE INTO cache (hash, response, ts) VALUES (?, ?, ?)",
        (key, json.dumps(value), int(time.time())),
    )
    conn.commit()
    conn.close()

Step 6: Stream results for real-time dashboards

When a technician plugs into the edge gateway, they want to see tokens immediately. Streaming reduces perceived latency on slow last-mile links.

from openai import OpenAI

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

def diagnose_stream(compressed_logs: str):
    stream = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": compressed_logs},
        ],
        response_format={"type": "json_object"},
        stream=True,
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content
        if token:
            print(token, end="", flush=True)
    print()

Step 7: Tie it together in an EdgeAgent class

This is the module I actually import into the device's main loop. It compresses, checks cache, calls Oxlo.ai, and caches the result.

class EdgeAgent:
    def __init__(self):
        init_db()

    def run(self, raw_logs: str) -> dict:
        fingerprint = hashlib.sha256(raw_logs.encode()).hexdigest()
        cached = get_cached(fingerprint)
        if cached:
            return cached

        compressed = compress_logs(raw_logs)
        result = diagnose(compressed)
        set_cache(fingerprint, result)
        return result

Run it

Here is a sample syslog from a remote temperature controller. The agent compresses it and returns structured diagnostics.

if __name__ == "__main__":
    raw = """\
Mar 10 14:23:01 edge-gateway kernel: [temperature_sensor] reading 81.2C
Mar 10 14:23:05 edge-gateway kernel: [temperature_sensor] reading 82.1C
Mar 10 14:23:09 edge-gateway kernel: [temperature_sensor] reading 84.5C
Mar 10 14:23:12 edge-gateway kernel: [fan_controller] RPM drop detected 1200
Mar 10 14:23:15 edge-gateway kernel: [temperature_sensor] reading 87.0C
Mar 10 14:23:18 edge-gateway kernel: [thermal_daemon] THERMAL ALERT: throttling CPU
Mar 10 14:23:20 edge-gateway kernel: [temperature_sensor] reading 88.2C
"""

    agent = EdgeAgent()
    print(json.dumps(agent.run(raw), indent=2))

Expected output:

{
  "anomaly_detected": true,
  "severity": "critical",
  "recommended_action": "Inspect fan_controller and clean air vents immediately.",
  "affected_service": "thermal_daemon"
}

Next steps

Wire the EdgeAgent into a FastAPI endpoint so plant-floor HMIs can POST logs to localhost:8000/diagnose. If you want to cut cloud traffic further, add a local keyword filter before the compress step so routine heartbeats never leave the device, reserving Oxlo.ai for genuine anomalies.

Top comments (0)