We are going to build a maintenance coordinator agent that ingests raw PLC telemetry, flags anomalies, and emits structured work orders. It helps plant engineers and reliability teams reduce unplanned downtime without replacing existing SCADA infrastructure. Every call runs against Oxlo.ai's OpenAI-compatible endpoint, so you can drop this into your stack without vendor lock-in.
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: Generate mock line telemetry
I need realistic pump station data to test against. This function returns temperature, vibration, and suction pressure readings in the same shape an MQTT gateway would publish from the plant floor.
import json
import random
from datetime import datetime, timezone
def read_pump_station(station_id: str = "P-101"):
"""Return a single telemetry snapshot."""
return {
"station_id": station_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"temperature_c": round(random.gauss(78, 12), 2),
"vibration_mm_s": round(random.gauss(4.5, 1.2), 2),
"suction_pressure_kpa": round(random.gauss(210, 35), 2),
}
def batch_readings(count: int = 3):
return [read_pump_station() for _ in range(count)]
if __name__ == "__main__":
print(json.dumps(batch_readings(), indent=2))Step 2: Define the agent's system prompt and initialize the Oxlo.ai client
The agent acts as a reliability engineer. It must classify severity, name the likely fault mode, and recommend a corrective action and spare part. I will store the prompt as a constant and create the client pointing at Oxlo.ai.
from openai import OpenAI
import json
SYSTEM_PROMPT = """You are a senior reliability engineer monitoring rotating equipment.
Analyze the provided sensor telemetry and produce a maintenance assessment.
Follow these rules exactly:
1. Classify severity as one of: Nominal, Warning, Critical.
2. Identify the most likely fault mode.
3. Recommend one concrete corrective action.
4. Recommend one spare part that should be staged.
Return your reasoning in JSON with keys: severity, fault_mode, corrective_action, spare_part, reasoning.
If all readings are within normal ranges, return severity Nominal and set the other fields to None."""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY",
)
if __name__ == "__main__":
print("Client ready. Prompt loaded.")
print(json.dumps({"prompt_length": len(SYSTEM_PROMPT)}))Step 3: Classify anomalies with an LLM call
Now I feed the telemetry batch to a reasoning model. I use deepseek-v3.2 because it handles structured reasoning well and is available on Oxlo.ai's free tier, so you can prototype without guessing token costs.
from openai import OpenAI
import json
import random
from datetime import datetime, timezone
# Telemetry helpers from Step 1
def read_pump_station(station_id: str = "P-101"):
return {
"station_id": station_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"temperature_c": round(random.gauss(78, 12), 2),
"vibration_mm_s": round(random.gauss(4.5, 1.2), 2),
"suction_pressure_kpa": round(random.gauss(210, 35), 2),
}
def batch_readings(count: int = 3):
return [read_pump_station() for _ in range(count)]
# Agent config from Step 2
SYSTEM_PROMPT = """You are a senior reliability engineer monitoring rotating equipment.
Analyze the provided sensor telemetry and produce a maintenance assessment.
Follow these rules exactly:
1. Classify severity as one of: Nominal, Warning, Critical.
2. Identify the most likely fault mode.
3. Recommend one concrete corrective action.
4. Recommend one spare part that should be staged.
Return your reasoning in JSON with keys: severity, fault_mode, corrective_action, spare_part, reasoning.
If all readings are within normal ranges, return severity Nominal and set the other fields to None."""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY",
)
# New in Step 3
def assess_telemetry(readings: list[dict]) -> dict:
user_message = f"Sensor batch:\n{json.dumps(readings, indent=2)}"
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
content = response.choices[0].message.content.strip()
if content.startswith("
```"):
content = content.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(content)
if __name__ == "__main__":
readings = batch_readings(3)
result = assess_telemetry(readings)
print(json.dumps(result, indent=2))Step 4: Lock output shape with JSON mode
To guarantee parseable tickets in production, I switch on JSON mode and add a schema reminder. Oxlo.ai supports this OpenAI-compatible feature, and because Oxlo.ai charges a flat rate per request, sending a large telemetry payload does not inflate cost the way token-based billing would.
from openai import OpenAI
import json
import random
from datetime import datetime, timezone
# Telemetry helpers
def read_pump_station(station_id: str = "P-101"):
return {
"station_id": station_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"temperature_c": round(random.gauss(78, 12), 2),
"vibration_mm_s": round(random.gauss(4.5, 1.2), 2),
"suction_pressure_kpa": round(random.gauss(210, 35), 2),
}
def batch_readings(count: int = 3):
return [read_pump_station() for _ in range(count)]
# Agent config
SYSTEM_PROMPT = """You are a senior reliability engineer monitoring rotating equipment.
Analyze the provided sensor telemetry and produce a maintenance assessment.
Follow these rules exactly:
1. Classify severity as one of: Nominal, Warning, Critical.
2. Identify the most likely fault mode.
3. Recommend one concrete corrective action.
4. Recommend one spare part that should be staged.
Return your reasoning in JSON with keys: severity, fault_mode, corrective_action, spare_part, reasoning.
If all readings are within normal ranges, return severity Nominal and set the other fields to None."""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY",
)
# Step 4: strict JSON mode
def assess_telemetry_strict(readings: list[dict]) -> dict:
user_message = (
"Analyze the following sensor batch and return valid JSON only.\n\n"
f"{json.dumps(readings, indent=2)}"
)
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
response_format={"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
if __name__ == "__main__":
readings = batch_readings(3)
ticket = assess_telemetry_strict(readings)
print(json.dumps(ticket, indent=2))Step 5: Build the polling loop
In production, this loop would pull from an MQTT broker or OPC-UA historian. For now, it polls our mock generator every few seconds and prints maintenance tickets to stdout.
from openai import OpenAI
import json
import random
import time
from datetime import datetime, timezone
# Telemetry helpers
def read_pump_station(station_id: str = "P-101"):
return {
"station_id": station_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"temperature_c": round(random.gauss(78, 12), 2),
"vibration_mm_s": round(random.gauss(4.5, 1.2), 2),
"suction_pressure_kpa": round(random.gauss(210, 35), 2),
}
def batch_readings(count: int = 3):
return [read_pump_station() for _ in range(count)]
# Agent config
SYSTEM_PROMPT = """You are a senior reliability engineer monitoring rotating equipment.
Analyze the provided sensor telemetry and produce a maintenance assessment.
Follow these rules exactly:
1. Classify severity as one of: Nominal, Warning, Critical.
2. Identify the most likely fault mode.
3. Recommend one concrete corrective action.
4. Recommend one spare part that should be staged.
Return your reasoning in JSON with keys: severity, fault_mode, corrective_action, spare_part, reasoning.
If all readings are within normal ranges, return severity Nominal and set the other fields to None."""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY",
)
# Assessment function
def assess_telemetry_strict(readings: list[dict]) -> dict:
user_message = (
"Analyze the following sensor batch and return valid JSON only.\n\n"
f"{json.dumps(readings, indent=2)}"
)
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
response_format={"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
# Main loop
if __name__ == "__main__":
for cycle in range(3):
readings = batch_readings(3)
ticket = assess_telemetry_strict(readings)
print(f"\n--- Cycle {cycle + 1} ---")
print(json.dumps(ticket, indent=2))
time.sleep(2)Run it
Save the final script as maintenance_agent.py, replace YOUR_OXLO_API_KEY with your key from https://portal.oxlo.ai, and run:
python maintenance_agent.pyExample output:
--- Cycle 1 ---
{
"severity": "Warning",
"fault_mode": "Bearing degradation",
"corrective_action": "Schedule vibration analysis and inspect pump bearings within 48 hours",
"spare_part": "Angular contact bearing set P-101-BRG-01",
"reasoning": "Vibration readings averaged 6.8 mm/s with peaks above 7.2 mm/s, exceeding ISO 10816 limits for this pump class."
}
--- Cycle 2 ---
{
"severity": "Nominal",
"fault_mode": "None",
"corrective_action": "None",
"spare_part": "None",
"reasoning": "All temperature, vibration, and pressure readings fall within established baseline bands."
}
--- Cycle 3 ---
{
"severity": "Critical",
"fault_mode": "Cavitation",
"corrective_action": "Reduce pump speed and inspect suction strainer immediately",
"spare_part": "Suction strainer mesh 40-micron P-101-STR-01",
"reasoning": "Suction pressure dropped to 142 kPa while flow demand remained constant, indicating imminent cavitation risk."
}Next steps
Swap the mock generator for a real MQTT subscriber using paho-mqtt, and publish the JSON tickets to your CMMS webhook instead of stdout. If you need deeper chain-of-thought reasoning for root-cause analysis, switch the model string to kimi-k2.6 or qwen-3-32b on Oxlo.ai without changing any client code.
Top comments (0)