I built this agent for a reliability team that wanted condition-based alerts without deploying a custom ML pipeline. It reads raw equipment logs, flags anomalies, and returns structured repair recommendations. Because Oxlo.ai charges a flat rate per request instead of per token, we can feed the model an entire shift's sensor history and pay the same cost as a short ping.
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: Set up the client
I start by instantiating the OpenAI-compatible client pointing at Oxlo.ai and verifying the endpoint with a lightweight ping.
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="llama-3.3-70b",
messages=[{"role": "user", "content": "Respond with OK"}],
max_tokens=10,
)
print(response.choices[0].message.content)Step 2: Define the system prompt
The agent needs strict engineering guardrails. This prompt forces it to reason only from the provided telemetry and return nothing else.
SYSTEM_PROMPT = """You are a senior reliability engineer analyzing equipment telemetry.
Rules:
1. Base every conclusion only on the logs provided below.
2. Classify severity as INFO, WARNING, CRITICAL, or EMERGENCY.
3. State the most likely root cause in one sentence.
4. Recommend a single concrete next action.
5. Output valid JSON with keys: severity, root_cause, next_action, confidence (0-1).
Do not add markdown formatting or explanations outside the JSON."""Step 3: Ingest and format logs
Real SCADA or vibration logs are noisy. I bundle the last N readings into a single text block so the LLM sees the trend.
def format_log(asset_id, readings):
# readings is a list of dicts with keys: timestamp, sensor, value, unit
lines = [f"Asset: {asset_id}", "Recent readings:"]
for r in readings:
lines.append(
f"- {r['timestamp']} | {r['sensor']}: {r['value']} {r['unit']}"
)
return "\n".join(lines)
sample_readings = [
{"timestamp": "2024-06-01T08:00:00Z", "sensor": "vibration_rms", "value": 2.1, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:10:00Z", "sensor": "vibration_rms", "value": 2.8, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:20:00Z", "sensor": "vibration_rms", "value": 4.5, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:30:00Z", "sensor": "oil_temp", "value": 92, "unit": "degC"},
]
user_message = format_log("PUMP-001", sample_readings)
print(user_message)Step 4: Run the diagnosis
With the context ready, I send it to Oxlo.ai. Llama 3.3 70B handles structured instructions reliably at low latency.
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="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0.1,
max_tokens=512,
)
raw_output = response.choices[0].message.content
print(raw_output)Step 5: Lock output with JSON mode
Parsing free text in a PLC or MES pipeline is fragile. Oxlo.ai supports JSON mode, so I lock the output schema and validate it with Python.
import json
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="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=512,
)
try:
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2))
except json.JSONDecodeError as e:
print("Parse error:", e)Step 6: Batch process a fleet
In practice you monitor dozens of assets. I wrap the call in a function that loops over a list of logs and collects JSON results for a dashboard.
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def diagnose(asset_id, readings):
msg = format_log(asset_id, readings)
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": msg},
],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=512,
)
return json.loads(resp.choices[0].message.content)
fleet_logs = {
"PUMP-001": sample_readings,
"COMPRESSOR-A": [
{"timestamp": "2024-06-01T09:00:00Z", "sensor": "discharge_pressure", "value": 8.2, "unit": "bar"},
{"timestamp": "2024-06-01T09:10:00Z", "sensor": "discharge_pressure", "value": 7.9, "unit": "bar"},
]
}
fleet_results = {aid: diagnose(aid, logs) for aid, logs in fleet_logs.items()}
print(json.dumps(fleet_results, indent=2))Run it
Here is the complete script and the output I received when running it against Oxlo.ai.
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a senior reliability engineer analyzing equipment telemetry.
Rules:
1. Base every conclusion only on the logs provided below.
2. Classify severity as INFO, WARNING, CRITICAL, or EMERGENCY.
3. State the most likely root cause in one sentence.
4. Recommend a single concrete next action.
5. Output valid JSON with keys: severity, root_cause, next_action, confidence (0-1).
Do not add markdown formatting or explanations outside the JSON."""
def format_log(asset_id, readings):
lines = [f"Asset: {asset_id}", "Recent readings:"]
for r in readings:
lines.append(
f"- {r['timestamp']} | {r['sensor']}: {r['value']} {r['unit']}"
)
return "\n".join(lines)
def diagnose(asset_id, readings):
msg = format_log(asset_id, readings)
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": msg},
],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=512,
)
return json.loads(resp.choices[0].message.content)
if __name__ == "__main__":
logs = {
"PUMP-001": [
{"timestamp": "2024-06-01T08:00:00Z", "sensor": "vibration_rms", "value": 2.1, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:10:00Z", "sensor": "vibration_rms", "value": 2.8, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:20:00Z", "sensor": "vibration_rms", "value": 4.5, "unit": "mm/s"},
{"timestamp": "2024-06-01T08:30:00Z", "sensor": "oil_temp", "value": 92, "unit": "degC"},
],
"COMPRESSOR-A": [
{"timestamp": "2024-06-01T09:00:00Z", "sensor": "discharge_pressure", "value": 8.2, "unit": "bar"},
{"timestamp": "2024-06-01T09:10:00Z", "sensor": "discharge_pressure", "value": 7.9, "unit": "bar"},
]
}
for aid, data in logs.items():
result = diagnose(aid, data)
print(f"{aid}: {json.dumps(result)}")PUMP-001: {"severity": "CRITICAL", "root_cause": "Bearing degradation indicated by rapid vibration RMS increase combined with elevated oil temperature.", "next_action": "Schedule immediate bearing inspection and vibration analysis within 24 hours.", "confidence": 0.91}
COMPRESSOR-A: {"severity": "INFO", "root_cause": "Normal pressure fluctuation within operating envelope.", "next_action": "Continue routine monitoring; no action required.", "confidence": 0.85}Wrap up
Two concrete next steps. First, wire the diagnose() function into an MQTT subscriber or a cron job so it triggers on every new SCADA payload. Second, if you need deeper multi-sensor reasoning, swap Llama 3.3 70B for DeepSeek V3.2 or Qwen 3 32B on Oxlo.ai. Because Oxlo.ai uses request-based pricing, sending long sensor histories costs the same flat rate as a short ping, which keeps fleet-wide monitoring affordable.
Top comments (0)