Sense Data

The Sense stage collects data from your external sources and passes it to the Reason stage. Each cycle, all configured Sense substages run in order, and their results are merged together for analysis.

Where does this data land in agent state? Every Sense substage writes into environmentalData.payload using a per-source-type keying scheme. See Where Sensed Data Lands in Agent State at the end of this page for the exact mapping from MQTT topic, database query, API endpoint, or CSV file to the in-memory location your Reason prompts and conditions read. The per-substage notes below give a quick pointer.

How to Configure It

The Sense stage lives under stages.sense.substages in your data-defined agent (DDA). Add one or more substages of different types to collect from multiple sources in a single cycle.

Substage Types

Type What it does

mqtt

Subscribes to MQTT topics and collects messages

api

Makes HTTP requests to a REST API

database

Runs a SQL query against a configured database

csv

Reads and parses a CSV file from a local path or URL

agent-messages

Collects messages from the agent bus discussion channel

Sandbox Mode: To run agents without live data sources during development or testing, set HIVEMQ_SANDBOX_MODE=true. The platform replaces configured sense substages with synthetic data so the full cycle executes without needing real MQTT brokers, databases, or APIs. See Sandbox Mode.

MQTT

Subscribe to MQTT topics and collect messages that arrive within the timeout window.

sense:
  substages:
    - type: mqtt
      name: quality-metrics
      connection: factory-mqtt # Name from your connections: block
      config:
        topics:
          - factory/quality/metrics
          - factory/quality/+/inspection # MQTT wildcards supported
          - factory/sensors/#
        timeoutMs: 15000 # Wait up to 15 seconds for messages

Messages are collected into environmentalData.payload, keyed by topic. In an LLM Reason prompt, the payload object is exposed as {{environmentalData}}, so reference it with {{environmentalData | json}}. See Where Sensed Data Lands in Agent State for the exact keying.

Set timeoutMs based on how frequently your broker publishes. Too short and you may miss messages; too long and each cycle takes longer. For high-frequency topics, use windowing (see below) to get statistical summaries instead of raw messages.

API

sense:
  substages:
    - type: api
      name: erp-data
      connection: erp-api # Name from your connections: block
      config:
        endpoints:
          - name: active-orders # ← becomes the payload key
            path: /api/v1/production-orders/active
            method: GET
            params:
              site: berlin
              status: running

The substage places each endpoint’s response body in environmentalData.payload, keyed by the endpoint name (here, environmentalData.payload["active-orders"]).

Database

sense:
  substages:
    - type: database
      name: recent-defects
      connection: quality-db
      config:
        query:
          name: recent-defects # ← becomes the payload key
          sql: |
            SELECT defect_type, count(*) as count, avg(severity) as avg_severity
            FROM defects
            WHERE created_at > NOW() - INTERVAL '1 hour'
            GROUP BY defect_type
            ORDER BY count DESC
            LIMIT 20

The substage places the query result in environmentalData.payload, keyed by the query name, as an array of row objects (here, environmentalData.payload["recent-defects"]).

Windowing and Aggregation

By default, each cycle sees only that cycle’s raw observations. Windowing buffers observations across multiple cycles and gives the Reason stage aggregated statistics (average, max, trend) instead of raw values.

This is useful when you want to detect trends rather than react to individual spikes.

sense:
  window:
    size: 10 # Buffer the last 10 observations
    slide: 1 # Advance by 1 each cycle (rolling window)
    minSize: 5 # Wait until at least 5 observations before forwarding

    aggregations:
      - name: avg_temp
        field: payload.temperature
        function: avg

      - name: max_temp
        field: payload.temperature
        function: max

      - name: temp_slope
        field: payload.temperature
        function: slope # Positive = rising, negative = falling

      - name: reading_count
        function: count # Number of observations in the window

Aggregated values are available in Reason as _aggregations.<substage-name>.<aggregation-name>:

{{_aggregations.quality-metrics.avg_temp}}
{{_aggregations.quality-metrics.temp_slope}}

Available Aggregation Functions

Function Result

avg

Mean value

sum

Sum of all values

min

Smallest value

max

Largest value

count

Number of observations

first

First value in window

last

Most recent value

stddev

Standard deviation

variance

Statistical variance

median

50th percentile

p95

95th percentile

p99

99th percentile

slope

Linear trend direction

range

Max minus min

Payload Interpretation

When your incoming data has a variable or unpredictable structure, the interpret block uses the LLM to normalise it into a consistent shape before Reason sees it.

This is useful when you subscribe to topics published by multiple different device types with different payload formats.

sense:
  substages:
    - type: mqtt
      name: heterogeneous-sensors
      connection: factory-mqtt
      config:
        topics:
          - sensors/+/telemetry
        timeoutMs: 10000

      interpret:
        instructions: |
          Normalise the incoming sensor payload into a consistent structure.
          Extract temperature (in °C), pressure (in hPa), and any alert flags.
        output_schema:
          type: object
          properties:
            temperature_c:
              type: number
            pressure_hpa:
              type: number
            alert:
              type: boolean

The LLM normalises each unique payload shape once, then caches the result. Subsequent messages with the same structure are normalised without another LLM call.

The Reason stage receives the normalised interpreted data alongside the raw original payload.

Agent Messages

Collect messages from the agent bus discussion channel. Use this when you want an agent to respond to instructions from other agents or from operators who send commands via the chat portal.

sense:
  substages:
    - type: agent-messages
      name: instruction-collector
      config:
        maxMessages: 10
        timeoutMs: 5000

Messages are placed in environmentalData.payload.agentMessages.

Where Sensed Data Lands in Agent State

Each cycle runs against a shared state object. The Sense stage writes everything it collects into environmentalData.payload, and the Reason and Reflect stages read from there. environmentalData is fully replaced every cycle. The previous cycle’s sensed data is gone unless you carried it forward through memory or a window.

Where each source lands in payload depends on the substage type and on whether you configure a window.

Without a Window

Source type Source identifier Memory location

mqtt

config.topics: [topic1, topic2]

environmentalData.payload[<topic>]: The latest message’s payload per topic

database

config.query.name

environmentalData.payload[<queryName>]: Array of row objects

api

config.endpoints[].name

environmentalData.payload[<endpointName>]: The response body (status, headers, and latency are stripped)

csv

config.name

environmentalData.payload[<csvName>]: { rows, columns, totalRows }

agent-messages

(fixed key)

environmentalData.payload.agentMessages: Array of messages

Only the latest message per MQTT topic is kept. If several readings arrived during the Sense window, the older ones are discarded. Configure a window if you need all of them.

When several substages run in the same cycle, the platform merges their outputs flat into the one payload object. If two substages produce the same key, the later one wins, so keep your topic, query, and endpoint names distinct.

Slash-keyed topics need bracket notation. MQTT topic keys contain /. In prompts and in rule conditions, always use bracket notation, such as {{state['factory/line1/temperature']}}. Dot notation (state.factory.line1.temperature) silently returns undefined and your condition never fires.

With a Window

When you add a window block, the keying flips. Instead of one entry per topic, query, or endpoint, you get one entry per substage, which holds the full window of observations. Aggregations and window metadata appear in their own top-level fields.

Field Holds

environmentalData.payload[<substageName>]

Array of the N observations the window collected

environmentalData._aggregations[<substageName>]

One entry per aggregation you configured ({ <aggregationName>: <number> })

environmentalData._windowMetadata[<substageName>]

Window size and first/last timestamps

Because the keying changes completely, choose either windowed or non-windowed per agent. The same Reason prompt or condition cannot serve both. A windowed substage also produces nothing until its buffer fills, so design your rules and prompts to tolerate a no-data cycle (treat _aggregations[<substageName>] as possibly undefined).