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)