Using LLMs for Predictive Maintenance

#learnai #oxlo #ai

We are going to build a lightweight predictive maintenance agent that reads machine sensor logs and recent work orders, then returns a structured risk assessment. It is aimed at plant engineers and ops teams who need to catch bearing degradation or thermal runaway before scheduled inspections. The whole pipeline runs against Oxlo.ai, so stuffing long telemetry histories into the prompt does not inflate your cost the way token-based providers do.

What you'll need

Python 3.10 or newer
The openai SDK. Install it with pip install openai
An Oxlo.ai API key from https://portal.oxlo.ai
A few minutes to copy, paste, and run the code

Step 1: Collect and normalize sensor data

In production this would pull from your SCADA or historian database. For this tutorial we will mock a helper that returns the last 24 hours of readings for a given asset tag.

import random
from datetime import datetime, timedelta

def get_recent_telemetry(machine_id: str):
    # Mock telemetry: vibration mm/s, temperature C, runtime hours
    base = {"machine_id": machine_id, "readings": []}
    now = datetime.utcnow()
    for i in range(6):
        base["readings"].append({
            "timestamp": (now - timedelta(hours=4 * i)).isoformat() + "Z",
            "vibration_ms": round(2.5 + random.random() * 3.0, 2),
            "temp_c": round(65 + random.random() * 25, 1),
            "runtime_hrs": 1200 + i * 4
        })
    return base

Step 2: Build the maintenance log context

LLMs reason better over plain text logs than raw CSV rows. We will render the telemetry into a concise text block, then append any free-text work orders.

def build_context(telemetry: dict, work_orders: list[str]) -> str:
    lines = [
        f"Machine: {telemetry['machine_id']}",
        "Recent telemetry (last 24 hours):",
    ]
    for r in telemetry["readings"]:
        lines.append(
            f"  {r['timestamp']} | vibration: {r['vibration_ms']} mm/s | "
            f"temp: {r['temp_c']} C | runtime: {r['runtime_hrs']} hrs"
        )
    if work_orders:
        lines.append("Recent work orders:")
        for wo in work_orders:
            lines.append(f"  - {wo}")
    return "\n".join(lines)

Step 3: Write the system prompt

The system prompt defines the agent as a reliability engineer and forces JSON output so downstream scripts can parse it without regex.

SYSTEM_PROMPT = """You are a reliability engineer specializing in rotating equipment.
Analyze the provided telemetry and maintenance history.
Respond with a single JSON object containing exactly these keys:
- risk_score: integer from 1 to 10
- likely_fault: one of [none, bearing_wear, misalignment, thermal_overload, lubrication_issue]
- recommended_action: one of [none, schedule_inspection, immediate_shutdown, order_parts]
- reasoning: a concise sentence explaining your decision

Use only the data provided. Do not invent measurements."""

Step 4: Wire the Oxlo.ai client

We use the OpenAI SDK as a drop-in client pointed at Oxlo.ai. Because Oxlo.ai uses flat per-request pricing, you can pass in long maintenance histories without watching input costs scale. See https://oxlo.ai/pricing for plan details.

from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

def assess_machine(machine_id: str, work_orders: list[str]):
    telemetry = get_recent_telemetry(machine_id)
    context = build_context(telemetry, work_orders)

    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": context},
        ],
        temperature=0.2,
    )

    return response.choices[0].message.content

Step 5: Parse the response and route the result

The model returns a JSON string. We will load it and print a human-readable maintenance ticket. If parsing fails, we fall back to raw text so the script never crashes silently.

import json

def run_maintenance_agent(machine_id: str, work_orders: list[str] = None):
    if work_orders is None:
        work_orders = []
    raw = assess_machine(machine_id, work_orders)
    try:
        result = json.loads(raw)
    except json.JSONDecodeError:
        result = {
            "risk_score": 0,
            "likely_fault": "unknown",
            "recommended_action": "manual_review",
            "reasoning": f"Unparseable response: {raw}"
        }

    print(f"Asset: {machine_id}")
    print(f"Risk Score: {result['risk_score']}/10")
    print(f"Likely Fault: {result['likely_fault']}")
    print(f"Action: {result['recommended_action']}")
    print(f"Reasoning: {result['reasoning']}")
    return result

Run it

Copy the snippets above into a single file named maintenance_agent.py, replace YOUR_OXLO_API_KEY, and execute it. Here is a sample invocation with a compressor that has recent thermal observations.

if __name__ == "__main__":
    work_orders = [
        "2024-05-10: Replaced discharge valve seal",
        "2024-05-14: Observed elevated casing temp during rounds"
    ]
    run_maintenance_agent("COMP-001-A", work_orders)

Example output:

Asset: COMP-001-A
Risk Score: 7/10
Likely Fault: thermal_overload
Action: schedule_inspection
Reasoning: Casing temperatures trending upward combined with recent thermal observation warrant inspection before next runtime milestone.

Next steps

Swap in kimi-k2.6 or deepseek-v3.2 if you want heavier reasoning over longer maintenance backlogs. Both are available on Oxlo.ai with the same request-based pricing.

For production, replace the mock telemetry with an MQTT or OPC-UA listener, and enable Oxlo.ai JSON mode or function calling to guarantee schema compliance instead of hand-rolling try/except blocks.

DEV Community