We are going to build a lightweight telemetry agent that runs on a resource-constrained edge device, compresses local system data into minimal prompts, and calls Oxlo.ai for fast diagnostic analysis. This is for engineers who need LLM intelligence on factory floors or remote sites without shipping gigabyte-sized models to every node.
What you'll need
- Python 3.10+ on your edge device (Raspberry Pi 4, NUC, or any Linux ARM/x86 node)
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Verify connectivity and choose a fast model
Before we install anything else, we confirm the edge node can reach Oxlo.ai over its upstream link. We also test the smallest viable model so we do not waste bandwidth on heavy weights for a ping test.
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": "system", "content": "Reply only OK."},
{"role": "user", "content": "Ping"},
],
max_tokens=3,
stream=False
)
print(response.choices[0].message.content)Step 2: Collect and compress telemetry
Edge devices generate verbose logs. We strip everything except the top five processes, load averages, and free memory, then fold them into a terse markdown table. We immediately test this format against Oxlo.ai to make sure the model can still parse it.
import subprocess
import json
def get_telemetry():
top = subprocess.run(["ps", "aux", "--sort=-%mem"], capture_output=True, text=True)
lines = top.stdout.strip().split("\n")[1:6]
rows = []
for line in lines:
parts = line.split()
rows.append({"user": parts[0], "pid": parts[1], "cpu": parts[2], "mem": parts[3], "cmd": parts[10]})
load = subprocess.run(["cat", "/proc/loadavg"], capture_output=True, text=True).stdout.split()[:3]
mem = subprocess.run(["free", "-m"], capture_output=True, text=True).stdout.split("\n")[1].split()
return {"load": load, "mem_free_mb": mem[3], "top": rows}
snapshot = get_telemetry()
prompt = f"Load: {snapshot['load']}\nMem free: {snapshot['mem_free_mb']}MB\nTop procs:\n{json.dumps(snapshot['top'])}"
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": "system", "content": "Confirm you received telemetry by replying RECEIVED."},
{"role": "user", "content": prompt},
],
max_tokens=5
)
print(response.choices[0].message.content)Step 3: Write the diagnostic system prompt
The system prompt is the only place where we give the model its role and output rules. Keep it short so the payload stays small on every request. Here is the exact prompt we will use in production.
SYSTEM_PROMPT = """You are an edge-device diagnostician.
Analyze the telemetry snapshot and return a JSON object with exactly these keys:
- summary: one sentence describing the health state
- risk_level: LOW, MEDIUM, or HIGH
- action: one concrete command or fix the engineer should run next
Do not include markdown code blocks, only raw JSON."""Step 4: Stream responses to survive slow edge links
On a factory floor uplink, waiting for a full response to buffer can trigger client timeouts. We enable streaming so the edge node starts processing tokens immediately. We parse the stream into a complete string before local JSON parsing.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def diagnose_stream(telemetry_text):
stream = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": telemetry_text},
],
stream=True,
max_tokens=200
)
chunks = []
for chunk in stream:
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
return "".join(chunks)
# Example call
import json
sample = json.dumps({"load": ["0.45", "0.30", "0.25"], "mem_free_mb": "412", "top": [{"cmd": "python3"}]})
print(diagnose_stream(sample))Step 5: Cache identical telemetry to cut redundant requests
Edge devices often report the same stable state for minutes at a time. We hash the compressed telemetry and store the Oxlo.ai response in a local dictionary. If the hash matches, we skip the network round-trip entirely.
import hashlib
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
cache = {}
def get_diagnosis(telemetry_text):
h = hashlib.sha256(telemetry_text.encode()).hexdigest()[:16]
if h in cache:
return cache[h]
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": telemetry_text},
],
max_tokens=200
)
result = response.choices[0].message.content
cache[h] = result
return result
# Test cache miss then hit
txt = "Load: 0.1 0.1 0.1\nMem free: 800MB"
print("miss:", get_diagnosis(txt))
print("hit :", get_diagnosis(txt))Step 6: Wire the agent loop and handle offline gaps
We poll telemetry every thirty seconds and send it to Oxlo.ai only when the cache misses or when the load average crosses a threshold. We wrap the call in a try block so a temporary outage does not crash the edge service.
import time
import subprocess
import json
import hashlib
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
cache = {}
def get_telemetry():
top = subprocess.run(["ps", "aux", "--sort=-%mem"], capture_output=True, text=True)
lines = top.stdout.strip().split("\n")[1:6]
rows = [{"cmd": l.split()[10], "cpu": l.split()[2], "mem": l.split()[3]} for l in lines]
load = subprocess.run(["cat", "/proc/loadavg"], capture_output=True, text=True).stdout.split()[:3]
mem = subprocess.run(["free", "-m"], capture_output=True, text=True).stdout.split("\n")[1].split()[3]
return {"load": [float(x) for x in load], "mem_free_mb": int(mem), "top": rows}
def get_diagnosis(telemetry_text):
h = hashlib.sha256(telemetry_text.encode()).hexdigest()[:16]
if h in cache:
return cache[h]
try:
resp = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": telemetry_text},
],
max_tokens=200
)
result = resp.choices[0].message.content
cache[h] = result
return result
except Exception as e:
return json.dumps({"summary": "Network or API error", "risk_level": "UNKNOWN", "action": str(e)})
if __name__ == "__main__":
while True:
snap = get_telemetry()
text = json.dumps(snap, separators=(",", ":"))
if snap["load"][0] > 0.8 or not cache:
diag = get_diagnosis(text)
print(diag)
time.sleep(30)Run it
Save the final script as edge_agent.py, export your key, and start it on the edge node.
export OXLO_API_KEY="sk-..."
python3 edge_agent.pyExample output on a healthy node:
{"summary": "System is stable with low load and adequate memory", "risk_level": "LOW", "action": "No action required, continue monitoring"}Next steps
Try swapping deepseek-v3.2 for qwen-3-32b if you need multilingual diagnostics on global edge sites. You can also batch telemetry from ten edge nodes into a single prompt to take advantage of Oxlo.ai request-based pricing, which stays flat no matter how many device reports you pack into one call.
Top comments (0)