We are going to build a lightweight edge diagnostics agent that runs on constrained hardware, collects local telemetry, and calls Oxlo.ai for structured maintenance recommendations. It is designed for factory floors or remote sites where bandwidth is spotty and you need deterministic output you can act on locally.
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
Step 1: Local queue and client setup
At the edge, connectivity comes and goes. I use a small SQLite database as a local FIFO queue so telemetry never gets dropped when the plant floor router drops out. I also initialize the Oxlo.ai client once and reuse it.
import sqlite3, json, os, time, random
from openai import OpenAI
DB_PATH = "edge_queue.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS pending (
id INTEGER PRIMARY KEY AUTOINCREMENT,
payload TEXT NOT NULL,
created_at REAL DEFAULT (unixepoch())
)
""")
conn.commit()
conn.close()
init_db()
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"]
)
Step 2: Simulate edge telemetry
Real edge hardware would read Modbus or MQTT. For this tutorial, I simulate three sensors: motor temperature, vibration RMS, and line voltage. I pack the last ten readings into a single structured payload so we can send one request per window instead of ten separate calls.
def get_sensor_window(n=10):
readings = []
for _ in range(n):
readings.append({
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"motor_temp_c": round(random.uniform(55.0, 95.0), 1),
"vibration_rms": round(random.uniform(0.2, 4.5), 2),
"line_voltage": round(random.uniform(218.0, 242.0), 1)
})
time.sleep(0.01)
return readings
def format_payload(readings):
return json.dumps({
"device_id": "edge-node-07",
"location": "assembly_line_b",
"readings": readings
}, indent=2)
Step 3: The diagnostics system prompt
The agent must return machine-readable output. I force JSON via the prompt itself so the edge script can parse alerts without brittle regex. I also give it a clear rubric for severity so local actuators know when to stop the line.
SYSTEM_PROMPT = """You are an industrial edge diagnostics engine.
Analyze the provided sensor window and return a single JSON object with exactly these keys:
- summary: a one-sentence human-readable assessment.
- severity: one of "normal", "warning", or "critical".
- recommended_action: one of "none", "schedule_maintenance", or "stop_immediately".
- root_cause: a concise technical guess, or "none" if severity is normal.
Use these thresholds:
- motor_temp_c above 85 is warning, above 92 is critical.
- vibration_rms above 3.0 is warning, above 4.0 is critical.
- line_voltage outside 220-240 is warning, outside 210-250 is critical.
Return only the JSON object. Do not wrap it in markdown."""
Step 4: Inference dispatcher with offline fallback
This is the core. If the request succeeds, we return the parsed result. If the network is down, we serialize the payload to SQLite and move on. Because Oxlo.ai uses per-request pricing, not per-token pricing, we can stuff the full ten-reading window and detailed prompt into one flat-cost call.
def dispatch(payload_str: str):
try:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": payload_str},
],
)
return response.choices[0].message.content
except Exception:
conn = sqlite3.connect(DB_PATH)
conn.execute("INSERT INTO pending (payload) VALUES (?)", (payload_str,))
conn.commit()
conn.close()
return None
Step 5: Drain the queue and parse results
When connectivity returns, we replay queued payloads in order. I process them sequentially to avoid hammering the gateway on a weak connection. Once we have a result, I parse the JSON and print an alert. In production, this is where you would write to a local PLC or trigger a buzzer.
def drain_queue():
conn = sqlite3.connect(DB_PATH)
cursor = conn.execute("SELECT id, payload FROM pending ORDER BY id")
rows = cursor.fetchall()
results = []
for row_id, payload in rows:
result = dispatch(payload)
if result is not None:
conn.execute("DELETE FROM pending WHERE id = ?", (row_id,))
conn.commit()
results.append(result)
else:
break
conn.close()
return results
def handle_result(raw: str):
try:
data = json.loads(raw.strip())
severity = data.get("severity", "normal")
action = data.get("recommended_action", "none")
print(f"[{severity.upper()}] {data['summary']}")
print(f" Root cause: {data['root_cause']}")
print(f" Action: {action}")
if severity == "critical":
print(" >>> LOCAL ALERT: Triggering emergency stop relay.")
return data
except Exception as exc:
print("Parse error:", exc)
return None
Step 6: Main loop
Finally, I tie everything together. I generate a sensor window, dispatch it, drain any backlog, and sleep. This loop runs indefinitely on the edge device.
if __name__ == "__main__":
while True:
readings = get_sensor_window()
payload = format_payload(readings)
print("Sending window with", len(readings), "readings...")
result = dispatch(payload)
if result:
handle_result(result)
backlog = drain_queue()
for item in backlog:
handle_result(item)
time.sleep(30)
Run it
Save the script as edge_agent.py, set your key, and run it.
export OXLO_API_KEY="sk-oxlo.ai-..."
python edge_agent.py
Example output when a temperature spike occurs:
Sending window with 10 readings...
[WARNING] Motor temperature elevated but within operational bounds.
Root cause: Insufficient cooling airflow suspected.
Action: schedule_maintenance
When the network is unreachable, the script silently persists the payload to SQLite. On the next successful iteration, drain_queue replays it automatically.
Next steps
Wire the critical branch to a GPIO pin or OPC-UA tag so the LLM decision stops the line without human intervention. If you run multiple edge nodes, centralize the queue drain logic on a plant gateway so you only pay for the requests that actually leave the floor. You can view Oxlo.ai request-based pricing at https://oxlo.ai/pricing, which makes heavy-context diagnostics cheap even when you batch large sensor windows.
Top comments (0)