AI and IoT: The Complete Guide to AIoT, from Edge Models to LLM Agents

AI and IoT complement each other: IoT devices produce continuous streams of data about the physical world, and AI turns those streams into detections, forecasts and decisions. This guide explains where AI models should run, how data and commands flow between devices and models, and how to build a real-time anomaly detector and a safe command path for AI agents using MQTT.

What AIoT Means

AIoT, short for the Artificial Intelligence of Things, describes systems where connected devices and machine learning work as one loop. Sensors measure, models interpret, and the results drive actions: an alert, a maintenance ticket, a changed setpoint or a stopped conveyor.

Neither half is very useful alone at scale. Thousands of sensors produce more data than people can watch, so without analytics most of it is never looked at. And a model without live data is just a report about the past. Combining them makes it possible to:

• Detect unusual behavior as it happens, instead of discovering it in a monthly review.
• Predict failures, demand or energy consumption early enough to act.
• Recognize objects, people and defects in camera and audio streams.
• Explain and interact, letting people ask questions about their equipment in natural language.

The hard part is rarely the model. It is getting clean, contextualized data to the model reliably, and getting the model's output back to the right place. That is an architecture problem, and it is the focus of this guide.

Where AI Runs: Device, Edge or Cloud

The first design decision in any AI and IoT project is where inference happens. Each location has clear strengths.

In practice most systems use all three. A device filters and compresses, an edge node runs the time-critical model, and the cloud trains models and analyzes the fleet. The edge AI vs cloud AI comparison helps you decide case by case, and what is edge AI and what is TinyML explain the edge options in detail.

A useful rule is to move computation as close to the sensor as the model allows, and send only what upstream systems need. An accelerometer sampling at several kilohertz produces far too much data to stream continuously, but a microcontroller can compute RMS, peak and a few frequency bands every second and publish those features in a small JSON message. If you want to try on-device inference yourself, TinyML on the ESP32 walks through a complete example, and the edge AI hardware comparison helps you choose a gateway when the model outgrows a microcontroller.

MQTT as the Data and Command Channel for AI and IoT

Whatever the split, devices, edge nodes and cloud services must exchange data. MQTT is a natural fit for that role:

• Decoupling. A model subscribes to the topics it needs. Adding a second model, or replacing one, does not touch devices.
• Bidirectional. The same broker carries telemetry up and commands or model results down.
• Efficient. Small headers and persistent connections suit constrained devices and cellular links.
• Results as data. A model publishes its outputs, such as an anomaly score or a detected object, as new topics that dashboards, alerting and other models can consume.

A typical topic layout separates raw data, model outputs and commands:

plant1/line2/pump3/telemetry          raw sensor readings (JSON)
plant1/line2/pump3/ai/anomaly         model output: score, label, model version
plant1/line2/pump3/ai/forecast        predicted values
plant1/line2/pump3/command/set        validated commands to the device

Publishing model outputs to their own topics is an underrated pattern. It makes AI results observable like any other signal, which matters when someone asks why an alert fired. It also lets you run two model versions side by side and compare them.

Keep in mind that the broker carries data; it does not run models. ConnectMQ is a managed MQTT broker with no built-in AI features, and that is the honest, modular pattern: your models run in Python services, edge devices or ML platforms that connect to the broker as ordinary MQTT clients.

Preparing IoT Data for AI

Models are only as good as the data they see, and raw IoT data is messy. Before any training or inference, make sure each stream meets a few basic standards. Getting these right early saves weeks of cleanup later.

Timestamps at the source. Stamp every reading where it is measured, not where it is received. Network delays, retries and edge buffering reorder messages, and a model that uses arrival time will learn patterns that are really network artifacts. Synchronize device clocks with NTP and include the timestamp in the payload.

Explicit units and asset context. A value of 71.2 means nothing to a model in six months. Put units in the payload or in a documented schema, and make the topic path identify the asset unambiguously. A hierarchical namespace such as site, area, line and asset lets you group data by machine type or location when building training sets.

Consistent sampling. Many models assume regular intervals. Decide whether devices report periodically, on change, or both, and document it. When devices report on change, a periodic heartbeat with the current value helps separate "no change" from "no data".

Operating context. A pump drawing high current during startup is normal; the same current at steady state is not. Publish machine state, product type or batch ID alongside sensor readings so models can condition on them.

Labels and events. Supervised models need examples of the outcomes you care about. Capture maintenance actions, operator notes and confirmed faults as timestamped events on their own topics, so they can be joined with sensor history later. The labeling time-series sensor data guide covers practical workflows.

You can rehearse the whole data pipeline on a public MQTT test broker with simulated values before connecting real equipment, which is exactly what the hands-on example below does.

Core AIoT Use Cases

Anomaly detection

Anomaly detection flags readings or patterns that differ from normal behavior: a bearing running hotter than usual, an unusual vibration spectrum, an energy profile that changes overnight. It is the most common first AI project because it does not need labeled failure data. Methods range from rolling statistics to Isolation Forests and autoencoders. Start with the IoT anomaly detection guide.

Predictive maintenance

Predictive maintenance estimates when equipment will fail so maintenance can be planned. It needs history, ideally including failures, and good context about operating conditions. The predictive maintenance with machine learning article covers features, models and the data you need to collect first.

Forecasting

Forecasting predicts future values such as energy load, tank levels or production demand. Classic time-series methods often perform well and are easier to explain than deep models.

Computer vision

Cameras with object detection models count parts, check for defects, detect people in restricted areas or read gauges. Vision usually runs at the edge because streaming video to the cloud is expensive, and only the detections travel over MQTT, as shown in YOLO object detection with MQTT.

LLM agents and natural-language interfaces

Large language models can summarize sensor data, answer questions about equipment and, with tool calling, interact with devices. This is powerful and also the area where mistakes have physical consequences. The LLM agent for IoT devices guide covers the architecture, and the section below shows the safety layer that must sit between an agent and real hardware.

Hands-On: Real-Time Anomaly Detection over MQTT

This example builds a streaming anomaly detector in Python. A simulator publishes vibration readings with occasional faults, and a detector subscribes, scores each reading with a rolling z-score and publishes anomaly events to a separate topic. Both run against the free public area of broker.connectmq.com, which accepts anonymous clients with a unique, non-empty client ID. Data there is readable by anyone, so keep it simulated.

pip install "paho-mqtt>=2.0"

The sensor simulator

# sensor_sim.py
import json
import random
import sys
import time

import paho.mqtt.client as mqtt

RUN_ID = sys.argv[1]  # any short unique string, e.g. a7f3k2
TOPIC = f"connectmq/demo/{RUN_ID}/line2/pump3/telemetry"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pump3-sim-{RUN_ID}")
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

while True:
    vibration = random.gauss(2.0, 0.15)  # normal RMS velocity in mm/s
    if random.random() < 0.03:
        vibration += random.uniform(1.0, 2.5)  # injected fault spike
    payload = {"ts": int(time.time() * 1000), "vibrationMmS": round(vibration, 3)}
    client.publish(TOPIC, json.dumps(payload), qos=0)
    time.sleep(0.5)

The streaming detector

# detector.py
import json
import math
import sys
from collections import deque

import paho.mqtt.client as mqtt

RUN_ID = sys.argv[1]
BASE = f"connectmq/demo/{RUN_ID}/line2/pump3"
WINDOW, THRESHOLD, MODEL_VERSION = 120, 4.0, "zscore-v1"
history = deque(maxlen=WINDOW)


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{BASE}/telemetry", qos=0)
    print("detector listening:", reason_code)


def on_message(client, userdata, msg):
    reading = json.loads(msg.payload)
    value = reading["vibrationMmS"]

    if len(history) >= 30:  # wait for a minimal baseline
        mean = sum(history) / len(history)
        std = math.sqrt(sum((x - mean) ** 2 for x in history) / len(history)) or 1e-6
        score = (value - mean) / std
        if abs(score) > THRESHOLD:
            event = {
                "ts": reading["ts"],
                "value": value,
                "score": round(score, 2),
                "baselineMean": round(mean, 3),
                "model": MODEL_VERSION,
            }
            client.publish(f"{BASE}/ai/anomaly", json.dumps(event), qos=1)
            print("ANOMALY", event)
            return  # keep anomalies out of the baseline

    history.append(value)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"detector-{RUN_ID}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

Run python sensor_sim.py a7f3k2 in one terminal and python detector.py a7f3k2 in another, replacing a7f3k2 with your own random string. Watch the results with:

mosquitto_sub -h broker.connectmq.com -p 1883 -i viewer-a7f3k2 -t "connectmq/demo/a7f3k2/line2/pump3/ai/#" -v

Three details make this more than a toy. Anomalous values are excluded from the baseline, so a fault does not teach the detector that faults are normal. Each event carries the model version, so you can tell which detector produced it. And the output is a normal MQTT topic, so a dashboard, an alerting service or a maintenance system can subscribe without changes to the detector. For a version using Isolation Forest on several features, see real-time anomaly detection in Python.

Letting AI Agents Act Safely on Devices

When an LLM agent or any model sends commands to physical equipment, the most important component is not the model. It is the gate between the model and the device. Never let a model publish directly to command topics. Instead, route its proposals through a validator that enforces an allowlist, limits and rate controls:

# command_gate.py - validates AI-proposed commands before they reach devices
import json
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
POLICY = {
    "line2/pump3": {"setpointRpm": (800, 1600)},
    "line2/fan1": {"speedPct": (0, 80)},
}
MIN_INTERVAL_S = 10
last_sent = {}


def validate(proposal):
    asset, param, value = proposal["asset"], proposal["param"], proposal["value"]
    if asset not in POLICY or param not in POLICY[asset]:
        return f"{asset}/{param} is not on the allowlist"
    lo, hi = POLICY[asset][param]
    if not isinstance(value, (int, float)) or not lo <= value <= hi:
        return f"value {value} outside [{lo}, {hi}]"
    if time.time() - last_sent.get(asset, 0) < MIN_INTERVAL_S:
        return "rate limit: too many commands for this asset"
    return None


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/ai/proposals", qos=1)


def on_message(client, userdata, msg):
    try:
        proposal = json.loads(msg.payload)
        error = validate(proposal)
    except (ValueError, KeyError) as exc:
        proposal, error = {"raw": msg.payload.decode(errors="replace")}, f"malformed: {exc}"
    audit = {"ts": int(time.time() * 1000), "proposal": proposal, "accepted": error is None, "reason": error}
    client.publish(f"{PREFIX}/ai/audit", json.dumps(audit), qos=1)
    if error is None:
        topic = f"{PREFIX}/{proposal['asset']}/command/set"
        client.publish(topic, json.dumps({proposal["param"]: proposal["value"]}), qos=1)
        last_sent[proposal["asset"]] = time.time()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"gate-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

The agent, whichever LLM or framework it uses, publishes proposals like {"asset": "line2/pump3", "param": "setpointRpm", "value": 1450} to ai/proposals. Only the gate publishes to device command topics, and every decision is written to an audit topic. In production the gate runs with its own broker credentials, and devices only accept commands from it. Safety functions still belong in the PLC or controller, where no software above them can override them. Safe AI control of IoT devices extends this with human approval steps, and building an MCP server for MQTT shows how to expose read and command tools to an agent in a structured way.

An AIoT Project Roadmap

1. Define the decision. What will change when the model produces a result? If nothing, stop here.
2. Collect and contextualize data. Consistent topics, units, timestamps and asset metadata. Most AI projects fail on data quality, not algorithms.
3. Establish a baseline with simple rules or statistics. It tells you whether a complex model adds anything.
4. Pick the inference location based on latency, bandwidth and connectivity.
5. Publish model outputs as topics with model version and confidence.
6. Close the loop carefully: alerts first, recommendations next, automated commands last and only through a validation gate.
7. Monitor and retrain. Sensors drift, processes change and models degrade. The IoT MLOps guide covers versioning, monitoring and redeployment.

Common Pitfalls

• Training on unrepresentative data. A model trained in summer may flag every winter morning as an anomaly.
• Ignoring missing data. A sensor that goes silent is itself a signal. Use Last Will status topics so models do not mistake gaps for stable readings.
• Black-box alerts. Operators ignore alerts they cannot understand. Include the value, baseline and reason in every event.
• Over-trusting LLMs. Language models can produce plausible but wrong commands. Validate everything they propose against hard limits.
• Sending everything to the cloud. High-frequency vibration or video should be processed at the edge, with features or detections sent upstream.

Power Your AIoT Pipeline with ConnectMQ

Your models need a dependable stream of data in and a controlled channel for results and commands out. Create a free account at app.connectmq.com on the Serverless plan, no credit card required, then connect devices, detectors and agents with the generated credentials over TLS on port 8883. ConnectMQ supports MQTT 3.1.1 and 5.0, and paid plans add shared subscriptions so several detector instances can split the load of one telemetry stream. For more connections or an isolated environment, compare the cloud MQTT broker plans.

Frequently Asked Questions

What is the difference between AI and IoT?

IoT is the network of connected devices that sense and act in the physical world, while AI is the set of techniques that learn patterns from data and make predictions. AIoT combines them so device data feeds models and model outputs drive actions.

Should AI run on the edge or in the cloud?

Run time-critical and bandwidth-heavy inference, such as vision or high-frequency vibration analysis, at the edge. Use the cloud for training, fleet-wide analytics, forecasting and large language models. Most systems combine both.

Does ConnectMQ include AI or machine learning features?

No. ConnectMQ is the MQTT messaging layer. Your models run in your own services, edge devices or ML platforms, which connect to ConnectMQ as MQTT clients to receive data and publish results.

Can an LLM control IoT devices directly?

It should not. Route an agent's proposed commands through a validation gate that enforces allowlists, value limits and rate limits, records an audit trail, and is the only client permitted to publish to device command topics.

What is the easiest first AI and IoT project?

Real-time anomaly detection on an existing sensor stream, because it needs no labeled failure data. A rolling statistical baseline like the example in this guide is often enough to deliver useful alerts.