Edge nodes generate noisy telemetry, but most cannot run large models locally. We are going to build a lightweight edge agent that preprocesses local sensor logs and uses an LLM to flag anomalies in plain language. The agent runs on minimal hardware and offloads heavy reasoning to Oxlo.ai, so you get state-of-the-art diagnostics without state-of-the-art GPUs in the field.
What you'll need
- An Oxlo.ai API key from https://portal.oxlo.ai
- Python 3.10+
pip install openai- A machine to act as the edge node. I tested this on a Raspberry Pi 4, but any Linux gateway or laptop works.
Step 1: Generate synthetic edge telemetry
We need a reproducible log stream to mimic a temperature and vibration sensor. The script below writes 120 normal readings followed by a sudden heat spike.
import json
import random
from datetime import datetime, timedelta, timezone
LOG_FILE = "telemetry.jsonl"
def seed_data():
base = datetime(2024, 5, 21, 10, 0, 0, tzinfo=timezone.utc)
with open(LOG_FILE, "w") as f:
for i in range(120):
reading = {
"ts": (base + timedelta(seconds=i)).isoformat().replace("+00:00", "Z"),
"temp_c": round(random.uniform(22.0, 24.0), 2),
"vibration_ms2": round(random.uniform(0.1, 0.3), 3),
"sensor_id": "edge-01"
}
f.write(json.dumps(reading) + "\n")
# Inject anomaly
for i in range(5):
reading = {
"ts": (base + timedelta(seconds=120 + i)).isoformat().replace("+00:00", "Z"),
"temp_c": round(random.uniform(68.0, 72.0), 2),
"vibration_ms2": round(random.uniform(0.1, 0.3), 3),
"sensor_id": "edge-01"
}
f.write(json.dumps(reading) + "\n")
if __name__ == "__main__":
seed_data()
print(f"Wrote 125 lines to {LOG_FILE}")
Step 2: Build the local preprocessor
Edge AI means doing the cheap work locally. We read the last 60 lines, compute a five-number summary, and emit a compact text block so we do not burn bandwidth or tokens shipping raw JSON.
import json
LOG_FILE = "telemetry.jsonl"
def compress_telemetry(path=LOG_FILE, limit=60):
"""Return a human-readable summary of the last N readings."""
readings = []
with open(path) as f:
for line in f:
readings.append(json.loads(line))
buffer = readings[-limit:]
temps = [r["temp_c"] for r in buffer]
vibs = [r["vibration_ms2"] for r in buffer]
return (
f"Sensor {buffer[0]['sensor_id']} last {len(buffer)} readings "
f"from {buffer[0]['ts']} to {buffer[-1]['ts']}: "
f"temp mean {round(sum(temps)/len(temps), 2)}C "
f"(min {min(temps)}C, max {max(temps)}C), "
f"vibration mean {round(sum(vibs)/len(vibs), 3)} m/s2 "
f"(max {max(vibs)} m/s2)."
)
Step 3: Write the system prompt
I keep the prompt in its own variable so I can tune diagnostics without touching the request logic. It forces JSON output with explicit fields, which makes downstream automation trivial.
SYSTEM_PROMPT = """You are an industrial edge diagnostic agent.
Analyze the telemetry summary provided by the user.
Respond ONLY with a JSON object containing these keys:
- summary: a one-sentence human-readable assessment.
- anomaly_detected: boolean.
- severity: one of low, medium, high.
- recommended_action: one sentence.
Base your judgment on these thresholds:
- temp_max > 60C is critical.
- temp_mean > 30C is suspicious.
- vibration is informational unless accompanied by high temperature.
"""
Step 4: Wire the Oxlo.ai client
Now we send the compressed context to Oxlo.ai. We use the OpenAI SDK as a drop-in replacement and request JSON mode so the edge node receives machine-parseable output. Because Oxlo.ai charges per request, not per token, sending a detailed system prompt or a large telemetry block does not inflate cost. See https://oxlo.ai/pricing.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)
def analyze_with_oxlo(context: str):
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": context},
],
response_format={"type": "json_object"},
temperature=0.2,
)
return response.choices[0].message.content
Step 5: Wrap it in a single command
The final script ties the preprocessor and LLM call together. It writes a local alert.json only when the model flags an issue, so the edge node acts autonomously even with intermittent connectivity.
import json
import os
from openai import OpenAI
# --- config ---
LOG_FILE = "telemetry.jsonl"
ALERT_FILE = "alert.json"
OXLO_API_KEY = os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
# --- client ---
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key=OXLO_API_KEY)
# --- prompt ---
SYSTEM_PROMPT = """You are an industrial edge diagnostic agent.
Analyze the telemetry summary provided by the user.
Respond ONLY with a JSON object containing these keys:
- summary: a one-sentence human-readable assessment.
- anomaly_detected: boolean.
- severity: one of low, medium, high.
- recommended_action: one sentence.
Base your judgment on these thresholds:
- temp_max > 60C is critical.
- temp_mean > 30C is suspicious.
- vibration is informational unless accompanied by high temperature.
"""
# --- preprocessor ---
def compress_telemetry(path=LOG_FILE, limit=60):
readings = []
with open(path) as f:
for line in f:
readings.append(json.loads(line))
buffer = readings[-limit:]
temps = [r["temp_c"] for r in buffer]
vibs = [r["vibration_ms2"] for r in buffer]
return (
f"Sensor {buffer[0]['sensor_id']} last {len(buffer)} readings "
f"from {buffer[0]['ts']} to {buffer[-1]['ts']}: "
f"temp mean {round(sum(temps)/len(temps), 2)}C "
f"(min {min(temps)}C, max {max(temps)}C), "
f"vibration mean {round(sum(vibs)/len(vibs), 3)} m/s2 "
f"(max {max(vibs)} m/s2)."
)
# --- oxlo.ai inference ---
def analyze_with_oxlo(context: str):
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": context},
],
response_format={"type": "json_object"},
temperature=0.2,
)
return response.choices[0].message.content
# --- main ---
if __name__ == "__main__":
context = compress_telemetry()
print("Context:", context)
raw = analyze_with_oxlo(context)
result = json.loads(raw)
print(json.dumps(result, indent=2))
if result.get("anomaly_detected"):
with open(ALERT_FILE, "w") as f:
json.dump(result, f, indent=2)
print(f"Alert written to {ALERT_FILE}")
else:
print("No anomaly detected.")
Run it
Export your key, generate the data, then run the agent.
export OXLO_API_KEY="sk-..."
python -c "import step1; step1.seed_data()" # or run the Step 1 script
python edge_agent.py
Example output:
Context: Sensor edge-01 last 60 readings from 2024-05-21T10:00:00Z to 2024-05-21T10:01:24Z: temp mean 45.12C (min 22.15C, max 70.5C), vibration mean 0.203 m/s2 (max 0.298 m/s2).
{
"summary": "Maximum temperature reached 70.5C, indicating a critical overheating event.",
"anomaly_detected": true,
"severity": "high",
"recommended_action": "Trigger emergency thermal shutdown and inspect cooling fan."
}
Alert written to alert.json
Next steps
Add a cron job or systemd timer to run the agent every minute against rotating sensor buffers. If you want the agent to act instead of just report, give it function calling tools so it can restart services via local API endpoints. Oxlo.ai supports tool use out of the box, so the same client code can drive closed-loop edge automation.
Top comments (0)