Installez et activez un modèle approuvé

Utilisez des résultats mesurés pour trouver de meilleurs coefficients de modèle. Dans cet exemple, une application Python ajuste une formule de prédiction pondérative, la teste sur des enregistrements de validation distincts, et l’approuve si les prédictions s’améliorent.

L’application (appelée superviseur) stocke les coefficients approuvés en vue de prédictions ultérieures. L’ajustement s’exécute en NumPy, en dehors du graphique de flux de données. Une transformation de mappage intégrée évalue alors la formule approuvée.

Diagramme des caractéristiques appariées et des poids mesurés envoyé à une application externe pour ajustement et approbation. Il stocke les coefficients approuvés. Un graphique de prédiction séparé lit ces coefficients et évalue la formule pour chaque nouvelle lecture.

Comment les pièces se connectent

  1. Exportez les correspondances. Le graphe d’export relaie les correspondances terminées à l’application.
  2. Ajustez et approuvez en externe. L’exemple Python calcule un résultat candidat et une décision de révision. Votre superviseur gère l’approbation et la persistance.
  3. Stockez les coefficients approuvés Le superviseur écrit l’enregistrement complet du modèle actif dans le magasin d’état.

Pour consommer cet enregistrement, voir Lire les coefficients depuis le magasin d’état.

Pour la livraison via MQTT plutôt que la recherche sur le magasin d’état, voir Livrer les mises à jour de modèles approuvées via MQTT.

Prerequisites

Compléter les prérequis de l’exemple partagé, y compris le factory-mqtt profil et le default point de terminaison. Utilisez YAML pour le débogage et Azure CLI ou Bicep pour la gestion de production.

Pour les déploiements de CLI, définissez AIO_INSTANCE_NAME et RESOURCE_GROUP sur votre instance existante et votre groupe de ressources existant. Pour les déploiements Bicep, il faut également définir CUSTOM_LOCATION_NAME sur l'emplacement personnalisé existant de l'instance. Les onglets de déploiement font référence à l’instance, au profil et au point de terminaison existants ; ils ne les créent pas. Le graphe d’export ne nécessite pas de point de terminaison du registre car il n’a pas de transformations.

Le script d’ajustement nécessite aussi Python 3 et NumPy. L’exemple d’écriture state-store utilise les clients MQTT mosquitto_pub et mosquitto_sub.

Préparer les enregistrements correspondants

Préparez les correspondances terminées avec les valeurs de caractéristiques utilisées par chaque prédiction. Le schéma d’alignement et de corrélation montre une manière de les produire.

Ajuster un candidat à partir d’enregistrements synthétiques correspondants

Commencez par le modèle v7: weight = 2 * speed + 0.5 * temperature + 175. Chaque enregistrement apparié doit inclure les valeurs de caractéristiques utilisées pour cette pièce, son poids mesuré, et la version du modèle ayant produit la prédiction. Par exemple, une donnée d’entrée d’entraînement au superviseur est la suivante :

{"assetId":"line-1","pieceId":"part-1","modelVersion":7,"speed":8,"temperature":18,"predictedWeight":200,"measuredWeight":198,"residual":-2}

Utilisez ces correspondances synthétiques : quatre pour l’ajustement et deux pour la validation. Gardez les identifiants des pièces des ensembles distincts.

Utiliser Élément Vitesse Température Poids mesuré Prédiction du modèle v7
Formation part-1 8 18 198 200
Formation part-2 10 18 202 204
Formation part-3 8 22 200 202
Formation part-4 10 22 204 206
Validation part-5 9 20 201 203
Validation part-6 11 24 207 209

L’opération d’ajustement minimise la somme des carrés des erreurs pour b0 + b1 * speed + b2 * temperature. Ces valeurs d’entraînement donnent :

  • Coefficient de vitesse : (202 - 198) / (10 - 8) = 2.
  • Coefficient de température : (200 - 198) / (22 - 18) = 0.5.
  • Intercept: 198 - 2 * 8 - 0.5 * 18 = 173.

Le candidat prédit 201 et 207 sur les deux parties de validation. Son erreur quadratique moyenne (RMSE) est 0; la RMSE du modèle v7 est 2. Choisissez une taille d’échantillon de validation adéquate pour votre application.

Pour l’entrée en temps réel, déployez le graphe d’exportation en utilisant l’une des méthodes suivantes. L’application s’abonne à factory/supervisor-input; et publie les correspondances finalisées à factory/matches. La démo Python autonome ci-dessous utilise directement la table et ne nécessite pas de MQTT.

  1. Dans l’expérience d’opérations, ouvrez votre instance et sélectionnez Graphique de Data Flow>Créer un graphique de Data Flow.
  2. Nommez le graphique export-training-matches et sélectionnez le default profil.
  3. Ajouter la source source avec le point de terminaison factory-mqtt et le topic factory/matches.
  4. Ajoutez la destination destination avec le point de terminaison factory-mqtt et la rubrique factory/supervisor-input.
  5. Connectez source directement à destination, sans ajouter de transformation, et sélectionnez Enregistrer.

Le graphe transmet chaque correspondance trouvée sans changement. L’application collecte les correspondances, sépare les enregistrements d’apprentissage et de validation, puis ajuste un modèle candidat.

Un exemple de candidat émis par cette application sur factory/model-proposals:

{
  "kind": "modelProposal",
  "assetId": "line-1",
  "updateId": "update-42",
  "expectedModelVersion": 7,
  "modelVersion": 8,
  "coefficients": {
    "speedCoefficient": 2,
    "temperatureCoefficient": 0.5,
    "intercept": 173
  },
  "validation": {
    "trainingPieceIds": ["part-1", "part-2", "part-3", "part-4"],
    "validationPieceIds": ["part-5", "part-6"],
    "activeRmse": 2,
    "candidateRmse": 0
  }
}

Pour cet exemple, l’approbation nécessite l’actif configuré line-1, version active 7, version candidate 8, coefficients finis, ensembles d’échantillons disjoints, RMSE de validation plus faible, et une variation de l’ordonnée à l’origine ne dépassant pas 3. L’ordonnée à l’origine a varié de -2, ce modèle candidat est donc validé. Un candidat avec intercept 180 échoue au dernier test. Rejeter une nouvelle tentative après l’activation de v8. Ces vérifications sont des règles de stratégie données à titre d’exemple pour l’application.

Enregistrez cet exemple comme fit-model.py et exécutez python3 fit-model.py. Elle utilise les six enregistrements ci-dessus et affiche un modèle candidat ainsi qu’une décision de validation. Votre superviseur gère MQTT et les écritures en état actif ; le script ne fait qu’ajuster et réviser.

import json
import numpy as np

# Each row is (piece ID, speed, temperature, measured weight).
training = [
    ("part-1", 8, 18, 198),
    ("part-2", 10, 18, 202),
    ("part-3", 8, 22, 200),
    ("part-4", 10, 22, 204),
]
validation = [("part-5", 9, 20, 201), ("part-6", 11, 24, 207)]
if {r[0] for r in training} & {r[0] for r in validation}:
    raise ValueError("Training and validation pieces must be distinct.")

x = np.array([[1, speed, temperature] for _, speed, temperature, _ in training], dtype=float)
y = np.array([weight for _, _, _, weight in training], dtype=float)
candidate, _, rank, _ = np.linalg.lstsq(x, y, rcond=None)
if rank != 3 or not np.isfinite(candidate).all():
    raise ValueError("The fit needs three independent, finite parameters.")

vx = np.array([[1, speed, temperature] for _, speed, temperature, _ in validation], dtype=float)
vy = np.array([weight for _, _, _, weight in validation], dtype=float)
active = np.array([175, 2, 0.5])  # Intercept, speed, temperature for v7.
active_rmse = float(np.sqrt(np.mean((vx @ active - vy) ** 2)))
candidate_rmse = float(np.sqrt(np.mean((vx @ candidate - vy) ** 2)))
proposal = {
    "kind": "modelProposal",
    "assetId": "line-1",
    "updateId": "update-42",
    "expectedModelVersion": 7,
    "modelVersion": 8,
    "coefficients": {
        "intercept": float(candidate[0]),
        "speedCoefficient": float(candidate[1]),
        "temperatureCoefficient": float(candidate[2]),
    },
    "validation": {
        "trainingPieceIds": [row[0] for row in training],
        "validationPieceIds": [row[0] for row in validation],
        "activeRmse": active_rmse,
        "candidateRmse": candidate_rmse,
    },
}


def review(update, active_model, committed_update_ids):
    """Example policy for candidates produced by this trusted fitting process."""
    coefficients = update["coefficients"]
    metrics = update["validation"]
    if update["updateId"] in committed_update_ids:
        return {"approved": False, "reason": "duplicate update"}
    if update["assetId"] != active_model["assetId"]:
        return {"approved": False, "reason": "wrong asset"}
    if (update["expectedModelVersion"] != active_model["modelVersion"]
            or update["modelVersion"] != active_model["modelVersion"] + 1):
        return {"approved": False, "reason": "stale or unexpected version"}
    if (set(metrics["trainingPieceIds"]) & set(metrics["validationPieceIds"])
            or not metrics["trainingPieceIds"] or not metrics["validationPieceIds"]):
        return {"approved": False, "reason": "invalid validation split"}
    numbers = [*coefficients.values(), metrics["activeRmse"], metrics["candidateRmse"]]
    if not np.isfinite(numbers).all():
        return {"approved": False, "reason": "non-finite coefficient or metric"}
    if metrics["candidateRmse"] < 0 or metrics["activeRmse"] < 0:
        return {"approved": False, "reason": "invalid RMSE"}
    if metrics["candidateRmse"] >= metrics["activeRmse"]:
        return {"approved": False, "reason": "no validation improvement"}
    if abs(coefficients["intercept"] - active_model["intercept"]) > 3:
        return {"approved": False, "reason": "intercept change exceeds policy"}
    return {"approved": True, "reason": "candidate passes example policy"}


active_record = {"assetId": "line-1", "modelVersion": 7, "intercept": 175}
decision = review(proposal, active_record, set())
print(json.dumps({"candidate": proposal, "review": decision}, allow_nan=False))

L’ajustement en virgule flottante peut afficher des valeurs proches de 173, 2, 0.5, et 0. Utilisez des tolérances numériques et maintenez les coefficients exacts approuvés. La revue attend des candidats syntaxiquement valides et des indicateurs issus du processus d’ajustement fiable. Avant d’écrire l’enregistrement approuvé, le superviseur vérifie à nouveau la version active et enregistre l’ID d’approbation et de mise à jour dans son propre journal persistant. Sérialisez la revue et les écritures par ressource pour que les mises à jour concurrentes ne puissent pas toutes deux passer le même contrôle de version.

Stocker les coefficients approuvés

Après approbation, le superviseur écrit un enregistrement complet du modèle actif sous la clé factory-active-models. Au départ, il contient :

{"assetId":"line-1","modelVersion":7,"speedCoefficient":2,"temperatureCoefficient":0.5,"intercept":175}

Après avoir accepté update-42, remplacez-le par :

{"assetId":"line-1","modelVersion":8,"speedCoefficient":2,"temperatureCoefficient":0.5,"intercept":173}

Pour une démo de lookup autonome, écrivez directement l’enregistrement synthétique v8. Pour un modèle ajusté, remplacez les trois coefficients ci-dessous par les valeurs exactes approuvées. Pour comparer v7 et v8, lancez d’abord Writer avec modelVersion: 7 et intercept: 175, puis restaurez les valeurs v8 et relancez-le.

Utilisez le protocole state-store SET . Enregistrez ce script Python sous write-model.py; il crée le fichier de requête avec les longueurs d’octets requises.

import json
from pathlib import Path

record = {
    "assetId": "line-1",
    "modelVersion": 8,
    "speedCoefficient": 2,
    "temperatureCoefficient": 0.5,
    "intercept": 173,
}
parts = [b"SET", b"factory-active-models", json.dumps(record, separators=(",", ":")).encode("utf-8")]
request = b"*3\r\n" + b"".join(
    b"$" + str(len(part)).encode("ascii") + b"\r\n" + part + b"\r\n"
    for part in parts
)
Path("set-model.resp").write_bytes(request)

Pour une écriture de débogage, générez le fichier, puis envoyez-le avec les propriétés requises de MQTT v5. Exécutez d’abord l’abonné dans un autre terminal ; ajoutez les options d’authentification et TLS de votre courtier aux deux commandes MQTT.

Gardez l’horloge du writer à moins d’une minute de celle du broker. La __ts propriété porte l’horodatage de la demande. Si la réponse indique que l’horodatage est trop avancé dans le futur, corrigez le décalage d’horloge avant de réessayer.

mosquitto_sub -h <BROKER_HOST> -p <BROKER_PORT> -V mqttv5 -q 1 \
  -F '%D %p' \
  -t 'clients/model-writer/services/statestore/_any_/command/invoke/response'

python3 write-model.py
mosquitto_pub -h <BROKER_HOST> -p <BROKER_PORT> -V mqttv5 -q 1 \
  -t 'statestore/v1/FA9AE35F-2F64-47CD-9BFF-08E2B32A0FE8/command/invoke' \
  -D publish response-topic 'clients/model-writer/services/statestore/_any_/command/invoke/response' \
  -D publish correlation-data 'update-42' \
  -D publish user-property __ts "$(date +%s%3N):0:model-writer" \
  -f set-model.resp

Une écriture réussie affiche update-42 +OK: les données de corrélation de la requête suivies de la charge utile de la réponse. Pour les besoins de transport et d’autorisation, voir la référence du protocole State Store. Le manuel SET remplace l’enregistrement sans vérification de la version du modèle attendu. Utilisez un seul rédacteur agréé pour cette démonstration. Dans votre application, conservez l’historique approuvé et mettez à jour les identifiants en dehors du prédicteur pour la récupération et les vérifications en double. Pour rétablir la version précédente de la formule, approuvez les anciens coefficients comme une nouvelle version.

Pour le graphe qui lit le modèle stocké et calcule les prédictions, voir Lire les coefficients depuis le magasin d’état.

Contenu connexe

  • Dernière mise à jour :