Zamontuj i aktywuj zatwierdzony model

Wykorzystaj zmierzone wyniki, aby znaleźć lepsze współczynniki modelu. W tym przykładzie aplikacja w Pythonie dopasowuje formułę przewidywania wagi, sprawdza ją w oddzielnych rekordach walidacyjnych i zatwierdza, jeśli przewidywania będą trafniejsze.

Aplikacja (zwana supervisorem) przechowuje zatwierdzone współczynniki do późniejszych prognoz. Fitting działa w NumPy, poza grafem przepływu danych. Wbudowana transformacja mapowania następnie ocenia zatwierdzoną formułę.

Diagram dopasowanych cech i zmierzonych wag przesyłany do zewnętrznej aplikacji do dopasowania i zatwierdzenia. Przechowuje zatwierdzone współczynniki. Osobny wykres predykcyjny odczytuje te współczynniki i ocenia wzór dla każdego nowego odczytu.

Jak łączą się elementy

  1. Eksport dopasowań. Graf eksportu przekazuje ukończone dopasowania do aplikacji.
  2. Dopasowuj i zatwierdzaj zewnętrznie. Przykład w Pythonie oblicza kandydata i decyzję przeglądu. Twój przełożony odpowiada za zatwierdzanie i utrwalanie.
  3. Zapisz zatwierdzone współczynniki. Superwizor zapisuje pełny rekord aktywnego modelu do magazynu stanów.

Aby przetworzyć ten rekord, zobacz Odczytywanie współczynników z magazynu stanu.

W przypadku dostarczania przez MQTT zamiast wyszukiwania w magazynie stanowym, zobacz Dostarczaj zatwierdzone aktualizacje modelu przez MQTT.

Prerequisites

Ukończ wspólne przykładowe wymagania wstępne, w tym default profil i factory-mqtt punkt końcowy. Używaj YAML do debugowania oraz Azure CLI lub Bicep do zarządzania środowiskiem produkcyjnym.

Dla wdrożeń CLI ustaw AIO_INSTANCE_NAME i RESOURCE_GROUP na istniejącą instancję i grupę zasobów. W przypadku wdrożeń Bicep można również ustawić CUSTOM_LOCATION_NAME na istniejącą lokalizację niestandardową instancji. Zakładki wdrożeń odnoszą się do istniejącej instancji, profilu i punktu końcowego; nie tworzą ich. Graf eksportu nie wymaga endpointu rejestru, ponieważ nie ma transformacji.

Skrypt dopasowania wymaga także Python 3 i NumPy. Przykład zapisu stanów w magazynie wykorzystuje klientów MQTT mosquitto_pub i mosquitto_sub.

Przygotuj dopasowane rekordy

Przygotuj ukończone dopasowania z wartościami cech używanymi w każdej prognozie. Wzorzec wyrównania i korelacji pokazuje jeden ze sposobów ich wytworzenia.

Dopasuj kandydata z syntetycznych dopasowanych rekordów

Zacznij od modelu v7: weight = 2 * speed + 0.5 * temperature + 175. Każdy dopasowany rekord musi zawierać wartości cech użyte dla danego elementu, jego zmierzoną wagę oraz wersję modelu, która wygenerowała predykcję. Na przykład dane szkoleniowe dla nadzorcy to:

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

Użyj tych syntetycznych dopasowań: cztery do dopasowania modelu i dwa do walidacji. Zachowaj odrębne identyfikatory elementów zestawów.

Używanie Kawałek Szybkość Temperatura Zmierzona waga Predykcja Model v7
Szkolenia part-1 8 18 198 200
Szkolenia part-2 10 18 202 204
Szkolenia part-3 8 22 200 202
Szkolenia part-4 10 22 204 206
Validation part-5 9 20 201 203
Validation part-6 11 24 207 209

Operacja dopasowania minimalizuje sumę kwadratów błędów dla b0 + b1 * speed + b2 * temperature. Te wartości treningowe dają:

  • Współczynnik prędkości: (202 - 198) / (10 - 8) = 2.
  • Współczynnik temperatury: (200 - 198) / (22 - 18) = 0.5.
  • Przechwycenie: 198 - 2 * 8 - 0.5 * 18 = 173.

Kandydat przewiduje 207 i 201 dla dwóch fragmentów walidacyjnych. Jego pierwiastek z błędu średniokwadratowego (RMSE) to 0; RMSE modelu v7 to 2. Wybierz odpowiednią wielkość próby walidacyjnej dla swojej aplikacji.

W przypadku danych wejściowych na żywo wdroż graf wyeksportowany za pomocą jednej z następujących metod. Aplikacja subskrybuje factory/supervisor-input; publikuje ukończone dopasowania do factory/matches. Samodzielne demo Python poniżej korzysta bezpośrednio z tabeli i nie wymaga MQTT.

  1. W doświadczeniu operacyjnym otwórz swoją instancję i wybierz wykres przepływu danych>Utwórz wykres przepływu danych.
  2. Nadaj nazwę wykresowi export-training-matches i wybierz profil default.
  3. Dodaj źródło source z endpointem factory-mqtt i tematem factory/matches.
  4. Dodaj element docelowy destination z endpointem factory-mqtt i tematem factory/supervisor-input.
  5. Połącz source bezpośrednio do destination, bez dodawania transformacji i wybierz Zapisz.

Graf przekazuje dalej każdy zakończony mecz bez zmian. Aplikacja zbiera dopasowania, oddziela dane szkoleniowe i walidacyjne oraz dopasowuje model dla kandydata.

Przykładowy kandydat wygenerowany przez tę aplikację na 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
  }
}

W tym przykładzie zatwierdzenie wymaga skonfigurowanego zasobu line-1, aktywnej wersji 7, wersji kandydackiej 8, skończonych współczynników, zbiorów próbek rozłącznych, niższej walidacji RMSE oraz zmiany wyrazu wolnego nie większej niż 3. Wyraz wolny zmienił się o -2, więc ten kandydat modelu przechodzi. Kandydat z interceptem 180 nie przechodzi ostatniej próby. Odrzuć ponowienie, gdy v8 jest aktywne. Te kontrole są przykładowymi zasadami aplikacji.

Zapisz ten przykład jako fit-model.py i uruchom python3 fit-model.py. Wykorzystuje powyższe sześć rekordów i drukuje decyzję dotyczącą kandydata i decyzję z przeglądu. Twój przełożony zajmuje się MQTT i zapisami w stanie aktywnym; skrypt tylko dopasowuje i weryfikuje.

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))

Zaokrąglanie liczb zmiennoprzecinkowych może wypisać wartości bliskie 173, 2, 0.5 i 0. Stosuj tolerancje numeryczne i zachowuj dokładne zatwierdzone współczynniki. Review oczekuje poprawnie sformowanych kandydatów i wskaźników z zaufanego procesu dopasowania. Przed zapisem zatwierdzonego rekordu nadzorca ponownie sprawdza aktywną wersję i zapisuje identyfikator zatwierdzenia oraz aktualizacji w swoim własnym, trwałym dzienniku. Serializuj operacje sprawdzania i zapisu dla każdego zasobu, aby konkurencyjne aktualizacje nie mogły przejść tego samego sprawdzenia wersji.

Zapisz zatwierdzone współczynniki

Po zatwierdzeniu nadzorca zapisuje jeden kompletny rekord aktywnego modelu pod kluczem factory-active-models. Początkowo zawiera:

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

Po zaakceptowaniu update-42, zamień go na:

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

Aby przygotować demo samodzielnego wyszukiwania, zapisz syntetyczny rekord v8 bezpośrednio. Dla modelu dopasowanego zastąp wszystkie trzy poniższe współczynniki dokładnie zatwierdzonymi wartościami. Aby porównać v7 i v8, najpierw uruchom writera z modelVersion: 7 i intercept: 175, potem przywróć wartości v8 i uruchom go ponownie.

Użyj protokołu state-store SET. Zapisz ten skrypt Pythona jako write-model.py; tworzy plik żądania o wymaganej długości bajtów.

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)

Do zapisu debugującego wygeneruj plik, a następnie wyślij go z wymaganymi właściwościami MQTT v5. Najpierw uruchom subskrybenta w innym terminalu; Dodaj uwierzytelnianie i opcje TLS brokera do obu poleceń MQTT.

Utrzymuj zegar modułu zapisującego w granicach jednej minuty od zegara brokera. Właściwość __ts zawiera znacznik czasu żądania. Jeśli odpowiedź informuje, że znacznik czasu jest zbyt odległy w przyszłości, poprawij niedopasowanie zegara przed ponowną próbą.

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

Pomyślny zapis wyświetla update-42 +OK: dane korelacji żądania, a następnie payload odpowiedzi. Aby uzyskać wymagania dotyczące transportu i autoryzacji, zobacz dokumentację protokołu magazynu stanu. Ręczne SET wymienia rekord bez sprawdzania wersji oczekiwanego modelu. Do tego demo użyj jednego autoryzowanego writera. W swojej aplikacji zachowaj zatwierdzoną historię i aktualizuj identyfikatory poza predyktorem na potrzeby odzyskiwania i wykrywania duplikatów. Aby cofnąć wzór, zatwierdź stare współczynniki jako nową wersję.

Informacje o grafie, który odczytuje przechowywany model i oblicza prognozy, można znaleźć w sekcji Odczytywanie współczynników z magazynu stanów.

Treści powiązane

  • Ostatnia aktualizacja o