TensorFlow Lite Micro: Primeiros Passos em um Microcontrolador com Saída MQTT

O TensorFlow Lite Micro é o runtime que a maioria dos projetos de TinyML usa para executar redes neurais em microcontroladores, e sua API é pequena depois que você entende quatro objetos: o modelo, o op resolver, a tensor arena e o interpretador. Este guia de primeiros passos explica cada um deles, percorre um exemplo completo em uma placa ESP32-S3 e mostra como enviar previsões para um broker MQTT, seja diretamente via Wi-Fi, seja por meio de uma ponte serial para placas sem rede.

O que é o TensorFlow Lite Micro

O TensorFlow Lite for Microcontrollers (muitas vezes abreviado como TFLM e hoje parte da família LiteRT do Google) é um mecanismo de inferência em C++ projetado para dispositivos sem sistema operacional, sem alocação dinâmica de memória durante a inferência e com pouquíssima RAM. Ele lê o mesmo formato flatbuffer .tflite que o TensorFlow Lite em celulares, mas suporta um conjunto menor de operadores e foi feito para ser compilado diretamente no seu firmware.

O TFLM só faz inferência. O treinamento acontece no TensorFlow, em um computador; o microcontrolador recebe um modelo convertido, geralmente quantizado em int8. Se quiser entender o panorama de por que alguém roda ML em chips tão pequenos, veja o que é TinyML.

Os quatro conceitos centrais

Um programa típico faz o seguinte uma vez, na inicialização: mapeia o modelo, registra os operadores que ele usa, constrói o interpretador e aloca os tensores. Depois, a cada inferência: grava o tensor de entrada, chama Invoke() e lê o tensor de saída. Essa é toda a API de runtime de que você precisa para a maioria dos projetos.

Passo 1: treine e converta um modelo

O primeiro modelo clássico de TFLM aprende sin(x). Ele é pequeno, fácil de verificar a olho e exercita entrada e saída quantizadas. Este script o treina, converte para int8 completo e grava arquivos-fonte C++ para o firmware.

pip install tensorflow numpy

# train_sine.py
import numpy as np
import tensorflow as tf

rng = np.random.default_rng(1)
x = rng.uniform(0, 2 * np.pi, 3000).astype(np.float32)
y = (np.sin(x) + 0.05 * rng.standard_normal(x.size)).astype(np.float32)

model = tf.keras.Sequential([
    tf.keras.Input(shape=(1,)),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1),
])
model.compile(optimizer="adam", loss="mse")
model.fit(x, y, epochs=200, batch_size=64, validation_split=0.2, verbose=0)
print("val mse:", model.evaluate(x[:500], y[:500], verbose=0))


def rep():
    for v in np.linspace(0, 2 * np.pi, 300, dtype=np.float32):
        yield [np.array([[v]], dtype=np.float32)]


conv = tf.lite.TFLiteConverter.from_keras_model(model)
conv.optimizations = [tf.lite.Optimize.DEFAULT]
conv.representative_dataset = rep
conv.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
conv.inference_input_type = tf.int8
conv.inference_output_type = tf.int8
tfl = conv.convert()
print("model bytes:", len(tfl))

with open("tflm_sine/main/model_data.cc", "w") as f:
    f.write('#include "model_data.h"\n')
    f.write("alignas(16) const unsigned char g_model[] = {%s};\n" % ", ".join(f"0x{b:02x}" for b in tfl))
    f.write(f"const unsigned int g_model_len = {len(tfl)};\n")
with open("tflm_sine/main/model_data.h", "w") as f:
    f.write("#pragma once\nextern const unsigned char g_model[];\nextern const unsigned int g_model_len;\n")

Antes de escrever o firmware, inspecione os operadores do modelo convertido. Abrir o arquivo .tflite em um visualizador de modelos como o Netron ou usar tf.lite.experimental.Analyzer.analyze(model_content=tfl) mostra que este modelo contém apenas operadores FULLY_CONNECTED. Essa lista é exatamente o que você registra no resolver.

Por que quantizar para o TensorFlow Lite Micro

Microcontroladores são muito mais rápidos em aritmética de inteiros do que em ponto flutuante, e muitos nem têm unidade de ponto flutuante. A quantização int8 completa armazena cada peso em um byte em vez de quatro e executa todas as camadas com kernels inteiros, o que reduz o uso de flash, reduz a tensor arena e permite que bibliotecas otimizadas como o CMSIS-NN ou os kernels da Espressif façam seu trabalho. O custo é uma pequena perda de precisão e a etapa extra de converter entradas e saídas com o scale e o zero_point de cada tensor, o que o firmware abaixo faz explicitamente.

Flash e RAM: onde o modelo fica

O array do modelo é declarado como const, então o linker o coloca na flash e ele não consome RAM. A tensor arena é o custo em RAM: entradas, saídas, ativações intermediárias e um pouco de controle interno do interpretador. Na família ESP32, a RAM interna é a opção rápida; arenas muito grandes podem ir para a PSRAM em placas que a possuem, com alguma perda de velocidade. Para este modelo de seno, a arena precisa de bem menos do que os 4 KB reservados abaixo, o que você pode confirmar pela linha de log que o firmware imprime.

Passo 2: configure o projeto da placa

Este guia usa uma placa de desenvolvimento ESP32-S3, que também pode ser programada pela Arduino IDE, mas compila com o ESP-IDF porque o componente esp-tflite-micro da Espressif acompanha a API upstream do TFLM e adiciona kernels otimizados. Em outras placas, o código C++ do próximo passo continua o mesmo; só o sistema de build muda.

tflm_sine/
  CMakeLists.txt
  main/
    CMakeLists.txt
    main.cc
    model_data.cc   (gerado)
    model_data.h    (gerado)

# tflm_sine/CMakeLists.txt
cmake_minimum_required(VERSION 3.16)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
project(tflm_sine)

# tflm_sine/main/CMakeLists.txt
idf_component_register(SRCS "main.cc" "model_data.cc" INCLUDE_DIRS ".")

cd tflm_sine
idf.py set-target esp32s3
idf.py add-dependency "espressif/esp-tflite-micro"

Passo 3: execute a inferência com o TensorFlow Lite Micro

O firmware percorre x de 0 a 2π, quantiza cada valor com a escala e o zero point do tensor de entrada, executa o modelo e imprime uma linha JSON por previsão no console serial.

// tflm_sine/main/main.cc
#include <algorithm>
#include <cmath>
#include <cstdio>

#include "esp_log.h"
#include "esp_timer.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "tensorflow/lite/micro/micro_interpreter.h"
#include "tensorflow/lite/micro/micro_mutable_op_resolver.h"
#include "tensorflow/lite/schema/schema_generated.h"

#include "model_data.h"

static const char *TAG = "tflm";
static constexpr int kArenaSize = 4 * 1024;
alignas(16) static uint8_t tensor_arena[kArenaSize];

extern "C" void app_main(void) {
  // 1. Mapeia o modelo
  const tflite::Model *model = tflite::GetModel(g_model);
  if (model->version() != TFLITE_SCHEMA_VERSION) {
    ESP_LOGE(TAG, "model schema %lu != runtime %d", (unsigned long)model->version(), TFLITE_SCHEMA_VERSION);
    return;
  }

  // 2. Registra apenas os operadores que o modelo usa
  static tflite::MicroMutableOpResolver<3> resolver;
  resolver.AddFullyConnected();
  resolver.AddQuantize();
  resolver.AddDequantize();

  // 3. Constrói o interpretador sobre uma arena estática e aloca os tensores
  static tflite::MicroInterpreter interpreter(model, resolver, tensor_arena, kArenaSize);
  if (interpreter.AllocateTensors() != kTfLiteOk) {
    ESP_LOGE(TAG, "AllocateTensors failed, increase kArenaSize");
    return;
  }
  ESP_LOGI(TAG, "arena used %u of %d bytes", (unsigned)interpreter.arena_used_bytes(), kArenaSize);

  TfLiteTensor *input = interpreter.input(0);
  TfLiteTensor *output = interpreter.output(0);
  const float two_pi = 2.0f * (float)M_PI;
  int step = 0;

  while (true) {
    float x = two_pi * (step % 40) / 40.0f;

    // 4. Quantiza a entrada: q = x / scale + zero_point
    int q = (int)lroundf(x / input->params.scale) + input->params.zero_point;
    input->data.int8[0] = (int8_t)std::min(127, std::max(-128, q));

    int64_t t0 = esp_timer_get_time();
    if (interpreter.Invoke() != kTfLiteOk) {
      ESP_LOGE(TAG, "Invoke failed");
      break;
    }
    int us = (int)(esp_timer_get_time() - t0);

    // 5. Desquantiza a saída: y = (q - zero_point) * scale
    float y = (output->data.int8[0] - output->params.zero_point) * output->params.scale;
    printf("{\"x\":%.3f,\"y\":%.3f,\"ref\":%.3f,\"us\":%d}\n", x, y, sinf(x), us);

    step++;
    vTaskDelay(pdMS_TO_TICKS(500));
  }
}

python train_sine.py
cd tflm_sine
idf.py build flash monitor

O monitor mostra linhas como {"x":1.571,"y":0.98,"ref":1.000,"us":...}. O y previsto deve acompanhar de perto o ref, com pequenos erros decorrentes da quantização. Se todas as saídas tiverem o mesmo valor, quase certamente a entrada está sendo gravada sem quantização.

Passo 4: envie as previsões via MQTT

Há duas formas de levar essas previsões a um broker.

Diretamente via Wi-Fi. Em placas com Wi-Fi, como o ESP32-S3, adicione o cliente MQTT do ESP-IDF e publique cada linha JSON em vez de imprimi-la. O tutorial de TinyML no ESP32 contém um firmware completo que faz exatamente isso, incluindo Last Will e lógica de publicação na mudança.

Ponte serial. Muitas placas populares para TFLM, como placas Cortex-M com sensores embarcados, não têm Wi-Fi. Elas podem imprimir JSON pela serial USB, e um computador ou Raspberry Pi o encaminha. Esta ponte funciona com o firmware acima sem alterações:

pip install pyserial "paho-mqtt>=2.0"

# serial_to_mqtt.py  uso: python serial_to_mqtt.py COM5   (ou /dev/ttyACM0)
import json
import sys
import uuid

import paho.mqtt.client as mqtt
import serial

port = sys.argv[1]
run = uuid.uuid4().hex[:8]
topic = f"connectmq/demo/{run}/lab/tflm-board/inference"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"tflm-bridge-{run}")
client.will_set(f"connectmq/demo/{run}/lab/tflm-board/status", "offline", qos=1, retain=True)
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
client.publish(f"connectmq/demo/{run}/lab/tflm-board/status", "online", qos=1, retain=True)
print("publishing to", topic)

with serial.Serial(port, 115200, timeout=2) as ser:
    while True:
        line = ser.readline().decode("utf-8", errors="ignore").strip()
        if not line.startswith("{"):
            continue                      # ignora linhas de boot e de log
        try:
            json.loads(line)              # encaminha apenas JSON válido
        except json.JSONDecodeError:
            continue
        client.publish(topic, line, qos=0)
        print(line)

Feche o idf.py monitor antes, já que apenas um programa pode ocupar a porta serial. Depois, acompanhe o fluxo de qualquer máquina:

mosquitto_sub -h broker.connectmq.com -p 1883 -i tflm-viewer-12 -t 'connectmq/demo/+/lab/tflm-board/#' -v

O servidor MQTT público (gratuito) serve para testes como este, desde que você se lembre de que qualquer pessoa pode ler a área pública. Dispositivos de produção devem usar credenciais privadas e TLS na porta 8883; na ponte, isso significa chamar client.tls_set() e client.username_pw_set() com os valores do seu broker ConnectMQ e usar o client ID emitido.

O padrão de ponte tem um efeito colateral útil: o firmware da placa fica livre de código de rede, certificados e credenciais, então você pode trocar de broker, adicionar TLS ou mudar os topics no gateway sem regravar nada. O trade-off é ter mais um dispositivo para manter funcionando. Para algumas placas de laboratório em um único computador, é ideal; para uma frota em campo, placas com Wi-Fi integrado que publicam diretamente costumam ser mais simples de operar.

Escolhendo o que publicar

Um modelo de regressão como este produz um valor a cada chamada, mas a maioria das aplicações reais se importa com eventos: um valor cruzando um limite, uma classe mudando ou uma confiança caindo. Decida no dispositivo ou na ponte o que conta como evento e publique esses eventos, além de um heartbeat lento. Isso mantém o tráfego bem dentro dos limites de taxa por cliente e simplifica os consumidores a jusante.

Solução de problemas do TensorFlow Lite Micro

1. Didn't find op for builtin opcode: registre o operador ausente e aumente a contagem do template do resolver.
2. AllocateTensors() falha: aumente a arena e, quando funcionar, reduza-a para arena_used_bytes() mais uma margem.
3. Hard fault ou travamento em Invoke(): o array do modelo não está alinhado ou a arena se sobrepõe a outra memória. Mantenha alignas(16) em ambos.
4. Precisão muito pior do que no PC: o conjunto de dados representativo não cobriu a faixa real de entrada, então a quantização cortou valores. Use amostras reais em representative_dataset.
5. Erros de compilação sobre MicroErrorReporter ou AllOpsResolver: você está misturando código de exemplo antigo com uma versão mais nova do TFLM. As versões atuais usam o construtor mostrado acima e o MicroMutableOpResolver.

Para onde ir a seguir

Substitua o modelo de seno por um classificador alimentado por features reais de sensores, como em TinyML no ESP32, ou deixe uma plataforma gerar o pipeline para você com o tutorial de Edge Impulse com MQTT. O repositório oficial do TensorFlow Lite Micro documenta os operadores suportados e o port para novas placas. Para o contexto de arquitetura, o guia de IA e IoT mostra como os modelos em microcontroladores se encaixam ao lado de analytics na nuvem.

Comece a usar o ConnectMQ

Quer as previsões saiam da sua placa via Wi-Fi ou por uma ponte, elas precisam de um destino confiável. O ConnectMQ é um broker MQTT gerenciado com TLS, MQTT 3.1.1 e 5.0 e um cliente MQTT no navegador para conferir as mensagens à medida que chegam. Crie um broker no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito.

Perguntas frequentes

Qual a diferença entre TensorFlow Lite e TensorFlow Lite Micro?

O TensorFlow Lite é voltado a celulares e dispositivos Linux com sistema operacional e memória dinâmica. O TensorFlow Lite Micro é um runtime enxuto para microcontroladores que usa uma arena estática, um conjunto reduzido de operadores e nenhuma dependência de sistema operacional.

Como escolho o tamanho da tensor arena?

Comece com um tamanho generoso, chame AllocateTensors(), depois leia arena_used_bytes() e defina a arena com esse valor mais uma margem de segurança. O requisito depende dos maiores tensores intermediários do modelo.

O TensorFlow Lite Micro suporta modelos em float?

Sim, muitos operadores têm kernels em float, mas modelos int8 são menores e muito mais rápidos na maioria dos microcontroladores. A quantização inteira completa é a escolha normal para produção.

O TensorFlow Lite Micro consegue enviar dados via MQTT sozinho?

Não. O TFLM apenas executa a inferência. Você publica os resultados com um cliente MQTT separado, como o cliente MQTT do ESP-IDF, a PubSubClient ou uma ponte serial em um gateway.