Building a Model-Based Reflex Agent in Python for IoT Control

I built this pattern for HydroSmart, an IoT water management system where an ESP32 streams flow rate, temperature, current and voltage to a Python backend that decides whether the motor should be running. The decision layer is a model-based reflex agent — a term from Russell and Norvig's agent taxonomy that describes exactly the amount of intelligence this problem needs: no more, no less.

Why a simple reflex agent isn't enough

A simple reflex agent maps the current percept straight to an action:

if flow_rate == 0:
    motor.off()

That looks reasonable until you run it against reality. Flow is zero for a second or two every time the motor starts, because water has inertia. A simple reflex agent sees that zero, concludes the pump is running dry, and shuts down — every single start. It has no way to know that it just turned the motor on, because it has no memory of anything.

The world is partially observable. The sensor tells you the flow right now; it does not tell you whether the tank is filling, whether this reading is the third zero in a row or the first, or whether the motor has been running for six hours. All of that has to be inferred and kept.

The four parts of a model-based agent

A model-based reflex agent adds two things to the simple version: a picture of the world it maintains, and knowledge of how the world changes. Concretely, four components:

• State — the agent's current best guess about the world, including things it can't sense.
• Transition model — how the world evolves on its own (a tank drains, water cools).
• Sensor model — how percepts map onto that state, including how much to trust them.
• Condition-action rules — what to do given the state.

The loop is always the same: update state from the percept → match a rule → act → record what you did. That last step is the one people leave out, and it's the one that makes the agent model-based rather than merely stateful.

Modelling internal state in Python

Use a dataclass. It keeps the state explicit, typed and easy to log — which matters enormously when you are debugging why a motor switched off at 3am.

from dataclasses import dataclass, field
from collections import deque
from time import monotonic


@dataclass
class WorldState:
    """The agent's belief about the system, not raw sensor output."""
    flow_lpm: float = 0.0
    temperature_c: float = 0.0
    current_a: float = 0.0
    voltage_v: float = 0.0

    # Inferred, not sensed:
    motor_on: bool = False
    motor_started_at: float | None = None
    dry_run_strikes: int = 0
    last_seen_at: float = 0.0
    flow_history: deque = field(default_factory=lambda: deque(maxlen=10))

    @property
    def uptime_s(self) -> float:
        if not self.motor_on or self.motor_started_at is None:
            return 0.0
        return monotonic() - self.motor_started_at

    @property
    def sensors_stale(self) -> bool:
        return monotonic() - self.last_seen_at > 15.0

Now the update step. This is the transition and sensor model combined: fold a new percept into the belief, applying what we know about how the system behaves.

SPIN_UP_GRACE_S = 8.0   # water has inertia; ignore flow while it builds
DRY_RUN_LIMIT   = 3     # consecutive bad readings before we act


def update_state(state: WorldState, percept: dict) -> WorldState:
    state.flow_lpm      = percept["flow"]
    state.temperature_c = percept["temp"]
    state.current_a     = percept["current"]
    state.voltage_v     = percept["voltage"]
    state.last_seen_at  = monotonic()
    state.flow_history.append(percept["flow"])

    # The key inference: zero flow only counts as "dry" once the pump has had
    # time to prime. Without this the agent fights its own last action.
    if state.motor_on and state.uptime_s > SPIN_UP_GRACE_S:
        if percept["flow"] < 0.5:
            state.dry_run_strikes += 1
        else:
            state.dry_run_strikes = 0
    else:
        state.dry_run_strikes = 0

    return state

Two details worth copying. First, strikes instead of a single reading — one anomalous sample from a hall-effect flow sensor is noise, three in a row is a fact. Second, the grace period is keyed to uptime_s, which only exists because the agent recorded its own action. That is the model-based part doing real work.

Writing condition-action rules you can debug

Resist the temptation to write one long if/elif chain. Make each rule a named, ordered object. You get priority for free, and — more importantly — the agent can tell you which rule fired, which is the difference between a system you trust and one you unplug.

from typing import Callable, NamedTuple


class Rule(NamedTuple):
    name: str
    condition: Callable[[WorldState], bool]
    action: str


# Order matters: safety rules first, optimisation last.
RULES: list[Rule] = [
    Rule("sensor_timeout",
         lambda s: s.sensors_stale,
         "MOTOR_OFF"),

    Rule("overvoltage",
         lambda s: s.voltage_v > 260 or s.voltage_v < 180,
         "MOTOR_OFF"),

    Rule("overcurrent",
         lambda s: s.current_a > 8.0,
         "MOTOR_OFF"),

    Rule("dry_run",
         lambda s: s.dry_run_strikes >= DRY_RUN_LIMIT,
         "MOTOR_OFF"),

    Rule("overheat",
         lambda s: s.temperature_c > 65,
         "MOTOR_OFF"),

    Rule("max_runtime",
         lambda s: s.uptime_s > 3600,
         "MOTOR_OFF"),
]


def decide(state: WorldState) -> tuple[str, str]:
    for rule in RULES:
        if rule.condition(state):
            return rule.action, rule.name
    return "NO_CHANGE", "default"

Putting it together: the agent loop

import logging

log = logging.getLogger("agent")


class MotorAgent:
    def __init__(self, actuator):
        self.state = WorldState()
        self.actuator = actuator

    def step(self, percept: dict) -> str:
        self.state = update_state(self.state, percept)
        action, rule_name = decide(self.state)

        if action == "MOTOR_OFF" and self.state.motor_on:
            self.actuator.off()
            self.state.motor_on = False
            self.state.motor_started_at = None
            log.warning("motor stopped by rule=%s state=%s", rule_name, self.state)

        elif action == "MOTOR_ON" and not self.state.motor_on:
            self.actuator.on()
            self.state.motor_on = True
            self.state.motor_started_at = monotonic()   # record our own action
            log.info("motor started by rule=%s", rule_name)

        return action

Every state change is logged with the rule that caused it. When someone asks why the pump stopped overnight, the answer is one grep away instead of a guess.

Talking to the ESP32

The ESP32 publishes a small JSON payload; the Python side never trusts it blindly. Validate before it reaches the agent — a malformed packet must not become a belief.

import json

REQUIRED = ("flow", "temp", "current", "voltage")


def parse_percept(raw: bytes) -> dict | None:
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        log.warning("bad JSON from device: %r", raw[:80])
        return None

    if not all(k in data for k in REQUIRED):
        log.warning("percept missing keys: %s", data.keys())
        return None

    try:
        return {k: float(data[k]) for k in REQUIRED}
    except (TypeError, ValueError):
        log.warning("non-numeric percept: %s", data)
        return None

Note that a dropped packet does not reset the state — it simply means no update happened, and sensors_stale will eventually fire on its own. Silence is handled by the timeout rule rather than by special-casing it in the parser.

On the firmware side, the ESP32 does no decision-making at all beyond a hard local cutoff. That split is deliberate: the microcontroller handles anything that must be instant even with the network down, and Python handles everything that benefits from memory and history. If you are setting up the ESP32 toolchain, my ESP32 Arduino IDE setup guide covers the install, and this troubleshooting guide covers the COM port problems that tend to bite first.

What I'd do differently

• Persist state across restarts. My first version kept everything in memory, so a backend restart forgot that the motor had already been running for 50 minutes. A small SQLite table fixes it.
• Add hysteresis to every threshold. A fixed cut-off at 65 °C means a sensor hovering at 64.9/65.1 will chatter the relay. Turn off at 65, allow restart only below 58.
• Make the grace period adaptive. Eight seconds was tuned for one pump. Measuring the actual time-to-flow on the first few successful starts and using that would generalise across installations.
• Test the rules without hardware. Because decide() is a pure function of state, every safety rule can be unit-tested with a constructed WorldState — no ESP32, no water. That is the single biggest practical benefit of separating state from rules.

The broader point: reaching for machine learning here would have been a mistake. The system has clear physical constraints and a handful of well-understood failure modes. A model-based reflex agent is transparent, testable and explainable — and when it is switching mains-powered equipment, those properties matter more than sophistication.