We are going to build a predictive maintenance agent that runs on an edge gateway and classifies equipment health from messy sensor logs. Because edge workloads produce variable-length telemetry bursts, token-based inference gets expensive fast. Oxlo.ai's flat per-request pricing, detailed at https://oxlo.ai/pricing, makes long log dumps predictable, so we can ship this without surprise bills.
What you'll need
- An Oxlo.ai API key from https://portal.oxlo.ai
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai python-dotenv
Step 1: Project setup and client initialization
Create a new directory and a .env file that holds your Oxlo.ai key. Then initialize the OpenAI-compatible client pointing to Oxlo.ai.
# .env
OXLO_API_KEY=sk-oxlo.ai-...
# edge_agent.py
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
Step 2: Define the system prompt
The agent must return strict JSON with a severity level and a recommended action. I keep the prompt short so it fits easily into context, which helps latency on the edge.
SYSTEM_PROMPT = """You are a predictive maintenance agent running on an industrial edge gateway.
Analyze the provided telemetry blob and return a single JSON object with exactly these keys:
- severity: one of OK, WARNING, CRITICAL
- action: a concise string describing the next step
- reason: one sentence explaining your diagnosis
Rules:
- If temperature is over 85 C and vibration is above 4.0, severity must be CRITICAL.
- Return only the JSON object, no markdown fences."""
Step 3: Build the telemetry formatter
Real edge devices dump unstructured text. This function collates temperature, vibration, and recent syslog lines into a plain text blob that we send to Oxlo.ai.
def format_telemetry(temp_c: float, vibration: float, logs: list[str]) -> str:
lines = [
f"Current temperature: {temp_c} C",
f"Vibration RMS: {vibration}",
"Recent logs:",
]
for line in logs[-5:]: # keep only last 5 lines to stay concise
lines.append(f"- {line}")
return "\n".join(lines)
Step 4: Inference with Oxlo.ai
I use Llama 3.3 70B because it handles structured instruction following reliably. A single request covers the entire log blob, and the flat pricing means a longer syslog trace does not change the cost.
def classify_health(telemetry_text: str) -> dict:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": telemetry_text},
],
temperature=0.1,
max_tokens=256,
)
raw = response.choices[0].message.content.strip()
import json
return json.loads(raw)
Step 5: Local action handler
The edge gateway needs to do something with the result. This stub prints to stdout and returns a relay command. In production, this would trigger a Modbus write or an MQTT alert.
def handle_result(result: dict):
severity = result.get("severity", "UNKNOWN")
action = result.get("action", "none")
reason = result.get("reason", "no reason given")
print(f"[{severity}] {action}")
print(f" Diagnosis: {reason}")
if severity == "CRITICAL":
print(" -> Relay command: SHUTDOWN_LINE")
elif severity == "WARNING":
print(" -> Relay command: SCHEDULE_MAINTENANCE")
else:
print(" -> Relay command: NOP")
Run it
Wire everything together and feed the agent two scenarios: a healthy machine and an overheating bearing.
if __name__ == "__main__":
# Scenario A: normal operation
normal_logs = [
"motor_controller: heartbeat ok",
"thermal_sensor: temp stable at 42 C",
"vibration_monitor: RMS 1.2 within limits",
]
text_a = format_telemetry(temp_c=42.0, vibration=1.2, logs=normal_logs)
result_a = classify_health(text_a)
handle_result(result_a)
print()
# Scenario B: overheating with high vibration
fault_logs = [
"thermal_sensor: temp rising 82 C",
"thermal_sensor: temp rising 89 C",
"vibration_monitor: RMS 4.5 ALARM",
"motor_controller: thermal throttling engaged",
]
text_b = format_telemetry(temp_c=89.0, vibration=4.5, logs=fault_logs)
result_b = classify_health(text_b)
handle_result(result_b)
Example output:
$ python edge_agent.py
[OK] Continue normal monitoring
Diagnosis: Temperature and vibration are within normal operating ranges.
-> Relay command: NOP
[CRITICAL] Immediate shutdown and inspect bearing
Diagnosis: Temperature exceeds 85 C and vibration RMS is above 4.0, indicating critical bearing failure.
-> Relay command: SHUTDOWN_LINE
Next steps
Swap the text formatter for a real OPC-UA or Modbus client so the agent reads live PLC registers. If you add a camera to the gateway, pipe the visual finding through an Oxlo.ai vision model such as Kimi K2.6 and merge the result into the same telemetry blob.
Top comments (0)