LLM Applications for IoT Analytics: Use Cases and Best Practices

#learnai #oxlo #ai

Industrial IoT fleets generate constant telemetry, but most data never gets reviewed until equipment fails. In this tutorial, I will build an IoT Telemetry Analyst that ingests raw sensor JSON, detects anomalies, and returns structured maintenance recommendations using an LLM. We will run the entire pipeline on Oxlo.ai through its OpenAI-compatible API.

What you'll need

Python 3.10 or newer
An Oxlo.ai API key from https://portal.oxlo.ai
The OpenAI SDK installed with pip install openai
This script runs entirely with the Python standard library, so no extra dependencies are needed

1. Initialize the Oxlo.ai client and synthesize device telemetry

I start by pointing the OpenAI SDK at Oxlo.ai and generating synthetic pump data so we can test the pipeline without hardware. One device will contain injected thermal and vibration anomalies.

from openai import OpenAI
import json
import random
from datetime import datetime, timedelta

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

def generate_telemetry(device_id: str, hours: int = 24, anomalous: bool = False):
    readings = []
    base_temp = 60.0
    base_vib = 45.0
    base_volt = 230.0
    for i in range(hours):
        ts = (datetime.utcnow() - timedelta(hours=hours - i)).isoformat() + "Z"
        noise = random.gauss(0, 1)
        temp = base_temp + noise + (15 if (anomalous and i > 18) else 0)
        vib = base_vib + noise * 2 + (30 if (anomalous and i > 18) else 0)
        volt = base_volt + noise * 0.5
        readings.append({
            "timestamp": ts,
            "temperature_c": round(temp, 2),
            "vibration_hz": round(vib, 2),
            "voltage": round(volt, 2)
        })
    return {"device_id": device_id, "readings": readings}

devices = [
    generate_telemetry("pump-001", anomalous=False),
    generate_telemetry("pump-002", anomalous=True),
    generate_telemetry("pump-003", anomalous=False),
]
print(f"Loaded telemetry for {len(devices)} devices")

2. Compress time-series data into an LLM-friendly markdown table

To keep the prompt concise, I format the most recent 12 hours of each device as a markdown table. This reduces noise and makes the model's job easier without losing trend information.

def format_device_context(device: dict, max_hours: int = 12) -> str:
    lines = [
        f"Device: {device['device_id']}",
        "| Hour | Temp (C) | Vib (Hz) | Voltage |",
        "|------|----------|----------|---------|"
    ]
    for r in device["readings"][-max_hours:]:
        hour = r["timestamp"][11:16]
        lines.append(
            f"| {hour} | {r['temperature_c']} | {r['vibration_hz']} | {r['voltage']} |"
        )
    return "\n".join(lines)

context_blocks = [format_device_context(d) for d in devices]
user_message = "\n\n".join(context_blocks)
print("Context size:", len(user_message), "chars")

3. Define the system prompt for structured IoT analytics

The system prompt acts as the agent's instruction manual. I embed explicit thresholds so the model classifies devices consistently, and I require pure JSON output to simplify downstream parsing.

SYSTEM_PROMPT = """You are an industrial IoT reliability engineer analyzing pump telemetry.
Your task:
1. Review the hourly readings for each device.
2. Flag any device where temperature exceeds 70 C or vibration exceeds 75 Hz.
3. Return a JSON object with this exact structure:
   {
     "summary": "string",
     "devices": [
       {
         "device_id": "string",
         "status": "healthy" | "warning" | "critical",
         "anomalies_detected": ["string"],
         "recommended_action": "string"
       }
     ]
   }
Rules:
- Be concise. One sentence per recommended_action.
- If no anomalies are found, status must be "healthy" and anomalies_detected must be an empty list.
- Output ONLY the JSON object, with no markdown fences."""

4. Send the telemetry to Oxlo.ai and parse the structured report

Now I call Llama 3.3 70B on Oxlo.ai. I enable JSON mode to enforce valid output, and I keep temperature low for reproducible threshold checks. Because Oxlo.ai uses per-request pricing, this large prompt with three full telemetry tables costs the same flat rate as a one-line greeting.

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.2,
)

raw = response.choices[0].message.content
result = json.loads(raw)
print(json.dumps(result, indent=2))

5. Wrap the logic in a reusable fleet monitor

For production use, I package the analysis into a function that returns both the full report and a boolean alert flag. This drops cleanly into a scheduled job or webhook handler.

def analyze_fleet(devices: list, max_hours: int = 12) -> dict:
    blocks = [format_device_context(d, max_hours) for d in devices]
    msg = "\n\n".join(blocks)
    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.2,
    )
    return json.loads(resp.choices[0].message.content)

def should_alert(analysis: dict) -> bool:
    return any(d["status"] == "critical" for d in analysis.get("devices", []))

report = analyze_fleet(devices)
if should_alert(report):
    print("ALERT: Critical anomaly detected in fleet.")
else:
    print("Fleet status nominal.")

Run it

Saving the complete script as iot_analyst.py and executing python iot_analyst.py produces the following output. The model correctly flags pump-002 as critical due to temperature and vibration spikes in the final six hours, while keeping pump-001 and pump-003 as healthy.

ALERT: Critical anomaly detected in fleet.

{
  "summary": "Pump-002 shows thermal and mechanical stress in recent hours.",
  "devices": [
    {
      "device_id": "pump-001",
      "status": "healthy",
      "anomalies_detected": [],
      "recommended_action": "Continue standard monitoring."
    },
    {
      "device_id": "pump-002",
      "status": "critical",
      "anomalies_detected": [
        "temperature exceeded 70 C threshold",
        "vibration exceeded 75 Hz threshold"
      ],
      "recommended_action": "Schedule immediate maintenance and inspect bearings."
    },
    {
      "device_id": "pump-003",
      "status": "healthy",
      "anomalies_detected": [],
      "recommended_action": "Continue standard monitoring."
    }
  ]
}

Next steps

Swap the synthetic generator for a real MQTT subscriber or Kafka consumer, and schedule this script to run every fifteen minutes. If your fleet grows and telemetry tables get longer, Oxlo.ai's request-based pricing keeps costs flat regardless of prompt size, which makes it significantly cheaper than token-based providers for high-frequency IoT workloads. For zero-cost prototyping, test the same pipeline on DeepSeek V3.2 through Oxlo.ai's free tier, then move to Kimi K2.6 when you need advanced reasoning over larger historical windows. You can view the full pricing structure at https://oxlo.ai/pricing.

DEV Community