Zarządzanie systemem ML: zgodność, ścieżka audytu i odpowiedzialna sztuczna inteligencja w produkcji
Twój model przewidywania rezygnacji osiąga 92% dokładności, obsługa FastAPI jest stabilna, klaster Kubernetes skaluje się automatycznie. Następnie przychodzi e-mail z działu prawnego: „Nasze model kredytowy należy do kategorii wysokiego ryzyka zgodnie z ustawą UE o sztucznej inteligencji. Musimy udowodnić, że nie dyskryminuje ze względu na płeć lub wiek. Gdzie są dzienniki decyzji z ostatnich 6 miesięcy? Kto zatwierdził wdrożenie do produkcji?”
Jeśli nie masz natychmiastowych odpowiedzi na te pytania, Twój projekt ML jest zagrożony. The Rozporządzenie UE w sprawie sztucznej inteligencji (Ustawa o AI), częściowo weszła w życie w sierpniu 2025 r., przy czym ostateczny termin dla systemów wysokiego ryzyka ustalono na 2 sierpnia 2026 r., przekształciło zarządzanie ML z opcjonalnej najlepszej praktyki w obowiązek prawny grożący karami finansowymi do 6%. roczny obrót światowy. Rynek MLOps wart 4,38 miliarda dolarów w 2026 roku, rosnący w tempie 39,8% CAGR również dlatego, że przedsiębiorstwa muszą wyposażyć się w infrastrukturę zarządzania, aby zachować zgodność.
W tym artykule budujemy kompleksowe ramy zarządzania ML typu open source: na podstawie analizy wymogów ustawy AI, do automatycznego generowania wzorców kart, do ścieżki audytu za pomocą MLflow, do wykrywanie uczciwości za pomocą Fairlearn, aż do wyjaśnienia za pomocą SHAP. Wszystko z działającym kodem Pythona i możliwe do natychmiastowego zastosowania, nawet przy ograniczonym budżecie.
Czego się nauczysz
- Wymogi praktyczne wynikające z unijnej ustawy o sztucznej inteligencji dotyczące systemów prania pieniędzy wysokiego ryzyka (termin: sierpień 2026 r.)
- Generuj automatyczne karty modeli ze znormalizowaną dokumentacją
- Wdrażaj kompleksowe ścieżki audytu za pomocą MLflow i logów strukturalnych
- Mierz i łagodź stronniczość dzięki Fairlearn (parytet demograficzny, wyrównane szanse)
- Wyjaśnialność za pomocą SHAP: globalne znaczenie funkcji i wyjaśnienia lokalne
- Rejestr zarządzania modelami: przejścia między etapami ze śledzonymi zatwierdzeniami
- Ramy oceny ryzyka umożliwiające klasyfikację modeli sztucznej inteligencji
- Lista kontrolna odpowiedzialnej sztucznej inteligencji dla MŚP z budżetem poniżej 5000 EUR/rok
Ustawa UE o sztucznej inteligencji: jakie zmiany w modelach uczenia maszynowego
Ustawa AI klasyfikuje systemy AI na cztery poziomy ryzyka wraz ze wzrostem obowiązków. Przed wdrożyć dowolne ramy zarządzania, musisz zrozumieć, gdzie pasuje do nich Twój model taksonomia. Klasyfikacja nie zależy od zastosowanej technologii (Twój XGBoost nie jest z natury mniej lub bardziej ryzykowne niż transformator), ale zzamierzone zastosowanie i domena zastosowania.
4 kategorie ryzyka ustawy o sztucznej inteligencji
Ryzyko niedopuszczalne (zabronione): systemy punktacji społecznej, manipulacja podprogowa,
wykorzystanie podatności. Brak możliwości wdrożenia.
Wysokie ryzyko (rygorystyczne wymagania): systemy kredytowe, zatrudniania, edukacji,
infrastruktura krytyczna, wyroby medyczne, migracje. Termin realizacji: 2 sierpnia 2026 r.
Ograniczone ryzyko (obowiązkowa przejrzystość): chatboty, deepfakes, systemy rekomendacyjne.
Obowiązek informowania użytkownika, który wchodzi w interakcję z AI.
Minimalne ryzyko (dobrowolne): filtry antyspamowe, gry AI, standardowe rekomendacje.
Brak szczególnych obowiązków regulacyjnych.
Dla systemów przy wysokim ryzykuustawa o sztucznej inteligencji nakłada szereg obowiązków technicznych i kroki organizacyjne, które zespół ML musi wdrożyć w swoim rurociągu MLOps. Główne wymagania, obowiązuje od sierpnia 2026 r., dotyczy:
- System Zarządzania Ryzykiem: ciągły proces identyfikowania, analizowania i łagodzenia m.in ryzyka w całym cyklu życia modelu.
- Zarządzanie danymi: zbiory danych szkoleniowych, walidacyjnych i testowych muszą być udokumentowane, reprezentatywne, wolne od uprzedzeń i odpowiednie do określonego celu.
- Dokumentacja techniczna: wystarczającą dokumentację techniczną, aby wykazać zgodność z organami nadzorczymi. Obejmuje architekturę, procedury szkoleniowe, metryki wydajność i znane ograniczenia.
- Automatyczne prowadzenie rejestrów: system musi automatycznie rejestrować zdarzenia istotne podczas operacji, z niezmiennymi dziennikami do celów audytu.
- Przejrzystość i informacje o użytkowniku: użytkownicy muszą wiedzieć, kim są interakcję z systemem AI i otrzymywanie zrozumiałych informacji o jego możliwościach i ograniczeniach.
- Nadzór człowieka: mechanizmy techniczne umożliwiające nadzór i interwencję człowieka i unieważniania zautomatyzowanych decyzji.
- Dokładność, solidność, cyberbezpieczeństwo: udokumentowane wskaźniki wydajności, Test odporności na zmiany w dystrybucji i kontradyktoryjne dane wejściowe.
Sankcje na mocy ustawy o sztucznej inteligencji: nie są teoretyczne
Kary za nieprzestrzeganie ustawy o AI są dotkliwe: do 30 milionów euro, czyli 6%. roczny obrót światowy za naruszenia dotyczące systemów o niedopuszczalnym ryzyku; dopóki 20 milionów euro lub 4% za inne zobowiązania; dopóki 10 milionów euro lub 2% za nieprawidłowe informacje władzom. Pierwszy krytyczny termin dla systemów wysokiego ryzyka i 2 sierpnia 2026 r: mniej niż 6 miesięcy od daty tego artykułu. Jeśli Twój model działa w branży infrastruktury kredytowej, wynajmu lub infrastruktury krytycznej, plan zapewnienia zgodności musi rozpocząć się już dziś.
Karty modeli: znormalizowana dokumentacja modeli
Le karta modelu, wprowadzony przez Google w 2019 r. i obecnie przyjęty jako standard branżowy, to ustrukturyzowane dokumenty opisujące model ML: cel, wydajność dla różnych podgrup dane demograficzne, znane ograniczenia, zamierzone i zalecane zastosowanie. Ustawa o sztucznej inteligencji domyślnie wymaga ich w część „dokumentacja techniczna”. Generuj je ręcznie i są podatne na błędy: poniższy kod automatyzuje tworzenie na podstawie metadanych szkoleniowych i wyników ewaluacji.
# model_card_generator.py
# Generatore automatico di Model Card conforme AI Act
# Compatibile con MLflow per tracciamento artefatti
import json
import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional
import mlflow
import pandas as pd
import numpy as np
from sklearn.metrics import (
accuracy_score, precision_score, recall_score,
f1_score, roc_auc_score, confusion_matrix
)
@dataclass
class ModelCardMetrics:
"""Metriche di performance del modello."""
accuracy: float
precision: float
recall: float
f1: float
auc_roc: Optional[float] = None
sample_size: int = 0
evaluation_date: str = ""
@dataclass
class SubgroupMetrics:
"""Metriche per sottogruppo demografico (richieste EU AI Act)."""
group_name: str
group_value: str
accuracy: float
precision: float
recall: float
false_positive_rate: float
sample_size: int
@dataclass
class ModelCard:
"""Model Card standardizzata conforme alle linee guida EU AI Act."""
# Identificazione modello
model_name: str
model_version: str
model_type: str
creation_date: str
last_updated: str
# Descrizione
intended_use: str
out_of_scope_uses: str
primary_intended_users: str
# Dati di training
training_data_description: str
training_data_size: int
feature_names: list
target_variable: str
sensitive_features: list
# Performance complessiva
overall_metrics: ModelCardMetrics = field(default_factory=ModelCardMetrics)
# Performance per sottogruppi (fairness)
subgroup_metrics: list = field(default_factory=list)
# Limitazioni note
limitations: list = field(default_factory=list)
# Avvertenze etiche
ethical_considerations: str = ""
# Compliance
risk_category: str = "" # "high-risk", "limited-risk", "minimal-risk"
regulatory_scope: list = field(default_factory=list)
# Approvazioni e contatti
model_owner: str = ""
approved_by: str = ""
approval_date: str = ""
contact: str = ""
def to_json(self) -> str:
return json.dumps(asdict(self), indent=2, ensure_ascii=False)
def to_markdown(self) -> str:
"""Genera Model Card in formato Markdown per documentazione."""
md = f"""# Model Card: {self.model_name} v{self.model_version}
**Tipo:** {self.model_type}
**Data creazione:** {self.creation_date}
**Ultimo aggiornamento:** {self.last_updated}
**Owner:** {self.model_owner}
**Approvato da:** {self.approved_by} ({self.approval_date})
**Categoria di rischio AI Act:** {self.risk_category}
## Uso Previsto
{self.intended_use}
## Uso NON Previsto
{self.out_of_scope_uses}
## Utenti Primari
{self.primary_intended_users}
## Dati di Training
- Descrizione: {self.training_data_description}
- Dimensione: {self.training_data_size:,} campioni
- Features: {', '.join(self.feature_names)}
- Target: {self.target_variable}
- Features sensibili (per fairness): {', '.join(self.sensitive_features)}
## Performance Complessiva
| Metrica | Valore |
|---------|--------|
| Accuracy | {self.overall_metrics.accuracy:.4f} |
| Precision | {self.overall_metrics.precision:.4f} |
| Recall | {self.overall_metrics.recall:.4f} |
| F1 Score | {self.overall_metrics.f1:.4f} |
| AUC-ROC | {self.overall_metrics.auc_roc:.4f} |
| Campioni valutazione | {self.overall_metrics.sample_size:,} |
## Performance per Sottogruppi (Fairness Analysis)
"""
for sg in self.subgroup_metrics:
md += f"""
### {sg['group_name']} = {sg['group_value']} (n={sg['sample_size']})
- Accuracy: {sg['accuracy']:.4f}
- Precision: {sg['precision']:.4f}
- Recall: {sg['recall']:.4f}
- False Positive Rate: {sg['false_positive_rate']:.4f}
"""
md += f"""
## Limitazioni Note
"""
for lim in self.limitations:
md += f"- {lim}\n"
md += f"""
## Considerazioni Etiche
{self.ethical_considerations}
## Contatto
{self.contact}
"""
return md
def generate_model_card(
model_name: str,
model_version: str,
model,
X_test: pd.DataFrame,
y_test: pd.Series,
sensitive_cols: list,
intended_use: str,
config: dict
) -> ModelCard:
"""
Genera una Model Card completa a partire dal modello e dai dati di test.
Args:
model: Modello scikit-learn addestrato
X_test: Feature set di test
y_test: Label di test
sensitive_cols: Colonne sensibili per analisi fairness
intended_use: Descrizione dell'uso previsto
config: Configurazione aggiuntiva (owner, contact, etc.)
"""
y_pred = model.predict(X_test)
y_proba = model.predict_proba(X_test)[:, 1] if hasattr(model, 'predict_proba') else None
# Metriche globali
overall = ModelCardMetrics(
accuracy=accuracy_score(y_test, y_pred),
precision=precision_score(y_test, y_pred, zero_division=0),
recall=recall_score(y_test, y_pred, zero_division=0),
f1=f1_score(y_test, y_pred, zero_division=0),
auc_roc=roc_auc_score(y_test, y_proba) if y_proba is not None else None,
sample_size=len(y_test),
evaluation_date=datetime.datetime.utcnow().isoformat()
)
# Metriche per sottogruppi (RICHIESTO EU AI Act per High-Risk)
subgroup_list = []
for col in sensitive_cols:
if col not in X_test.columns:
continue
for val in X_test[col].unique():
mask = X_test[col] == val
if mask.sum() < 30: # Skip gruppi troppo piccoli
continue
y_sub_true = y_test[mask]
y_sub_pred = y_pred[mask]
cm = confusion_matrix(y_sub_true, y_sub_pred, labels=[0, 1])
tn, fp, fn, tp = cm.ravel() if cm.size == 4 else (0, 0, 0, cm[0][0])
fpr = fp / (fp + tn) if (fp + tn) > 0 else 0.0
subgroup_list.append({
"group_name": col,
"group_value": str(val),
"accuracy": accuracy_score(y_sub_true, y_sub_pred),
"precision": precision_score(y_sub_true, y_sub_pred, zero_division=0),
"recall": recall_score(y_sub_true, y_sub_pred, zero_division=0),
"false_positive_rate": fpr,
"sample_size": int(mask.sum())
})
card = ModelCard(
model_name=model_name,
model_version=model_version,
model_type=type(model).__name__,
creation_date=config.get("creation_date", datetime.date.today().isoformat()),
last_updated=datetime.date.today().isoformat(),
intended_use=intended_use,
out_of_scope_uses=config.get("out_of_scope_uses", "Non specificato"),
primary_intended_users=config.get("primary_users", "Team Data Science"),
training_data_description=config.get("data_description", ""),
training_data_size=config.get("training_size", 0),
feature_names=list(X_test.columns),
target_variable=config.get("target", "label"),
sensitive_features=sensitive_cols,
overall_metrics=overall,
subgroup_metrics=subgroup_list,
limitations=config.get("limitations", []),
ethical_considerations=config.get("ethical_considerations", ""),
risk_category=config.get("risk_category", "minimal-risk"),
regulatory_scope=config.get("regulatory_scope", []),
model_owner=config.get("owner", ""),
approved_by=config.get("approved_by", ""),
approval_date=config.get("approval_date", ""),
contact=config.get("contact", "")
)
return card
def log_model_card_to_mlflow(card: ModelCard, run_id: str):
"""Registra la model card come artefatto MLflow per tracciabilita."""
with mlflow.start_run(run_id=run_id):
# Log come JSON
json_path = f"/tmp/model_card_{card.model_name}_v{card.model_version}.json"
with open(json_path, "w") as f:
f.write(card.to_json())
mlflow.log_artifact(json_path, artifact_path="governance")
# Log come Markdown
md_path = f"/tmp/model_card_{card.model_name}_v{card.model_version}.md"
with open(md_path, "w") as f:
f.write(card.to_markdown())
mlflow.log_artifact(md_path, artifact_path="governance")
# Log metriche di fairness come tag MLflow per ricerca rapida
mlflow.set_tag("governance.risk_category", card.risk_category)
mlflow.set_tag("governance.model_owner", card.model_owner)
mlflow.set_tag("governance.approved_by", card.approved_by)
mlflow.set_tag("governance.has_subgroup_analysis", str(len(card.subgroup_metrics) > 0))
print(f"Model Card registrata in MLflow run {run_id} sotto artifacts/governance/")Wykrywanie uczciwości i uprzedzeń za pomocą Fairlearn
Uczciwość w uczeniu się nie jest pojedynczym pojęciem: istnieje kilka definicji matematycznych niekompatybilne ze sobą. Ustawa o sztucznej inteligencji nie określa, jakich wskaźników należy używać, ale wymaga takich systemów grupy wysokiego ryzyka są oceniane i dokumentowane w porównaniu z grupami chronionymi. Dwa najbardziej metryki wspólne i uzupełniające się są Parytet demograficzny (równość demograficzna) mi theWyrównane szanse (równość szans pod warunkiem prawdziwych etykiet).
La Parytet demograficzny wymaga prawdopodobieństwa uzyskania pozytywnego wyniku jest taki sam dla wszystkich grup: P(Y_pred=1 | A=0) = P(Y_pred=1 | A=1). I najbardziej intuicyjny wskaźnik ale może ukryć różnice w rzeczywistej wydajności, jeśli grupy mają różne stawki za etykiety. THE'Wyrównane szanse i bardziej rygorystyczne: wymaga stosowania zarówno TPR, jak i FPR jednakowe we wszystkich grupach, zapewniając, że model popełnia błędy z tą samą częstotliwością niezależnie od przynależności do wrażliwej grupy.
# fairness_checker.py
# Analisi completa di fairness con Fairlearn
# pip install fairlearn scikit-learn pandas
import pandas as pd
import numpy as np
from fairlearn.metrics import (
MetricFrame,
demographic_parity_difference,
demographic_parity_ratio,
equalized_odds_difference,
equalized_odds_ratio,
false_positive_rate,
false_negative_rate,
selection_rate
)
from fairlearn.reductions import ExponentiatedGradient, DemographicParity, EqualizedOdds
from sklearn.metrics import accuracy_score, f1_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import mlflow
def run_fairness_analysis(
y_true: pd.Series,
y_pred: np.ndarray,
sensitive_feature: pd.Series,
y_proba: np.ndarray = None,
threshold: float = 0.1
) -> dict:
"""
Analisi completa di fairness per un modello.
Args:
y_true: Etichette vere
y_pred: Predizioni del modello
sensitive_feature: Feature sensibile (es. genere, eta_group)
y_proba: Probabilità predette (opzionale)
threshold: Soglia di accettabilita per le differenze di fairness
Returns:
dict con metriche di fairness e flag di conformità
"""
results = {}
# ---- 1. Demographic Parity ----
dp_diff = demographic_parity_difference(
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_feature
)
dp_ratio = demographic_parity_ratio(
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_feature
)
results["demographic_parity_difference"] = float(dp_diff)
results["demographic_parity_ratio"] = float(dp_ratio)
results["demographic_parity_pass"] = abs(dp_diff) <= threshold
# ---- 2. Equalized Odds ----
eo_diff = equalized_odds_difference(
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_feature
)
eo_ratio = equalized_odds_ratio(
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_feature
)
results["equalized_odds_difference"] = float(eo_diff)
results["equalized_odds_ratio"] = float(eo_ratio)
results["equalized_odds_pass"] = abs(eo_diff) <= threshold
# ---- 3. MetricFrame: metriche disaggregate per gruppo ----
metrics_dict = {
"accuracy": accuracy_score,
"f1": lambda y_t, y_p: f1_score(y_t, y_p, zero_division=0),
"false_positive_rate": false_positive_rate,
"false_negative_rate": false_negative_rate,
"selection_rate": selection_rate,
}
metric_frame = MetricFrame(
metrics=metrics_dict,
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_feature
)
results["by_group"] = metric_frame.by_group.to_dict()
results["overall"] = metric_frame.overall.to_dict()
results["difference"] = metric_frame.difference().to_dict()
results["ratio"] = metric_frame.ratio().to_dict()
# ---- 4. Flag conformità complessiva ----
results["is_fair"] = (
results["demographic_parity_pass"] and
results["equalized_odds_pass"]
)
# Messaggi esplicativi
messages = []
if not results["demographic_parity_pass"]:
messages.append(
f"ATTENZIONE: Demographic Parity Difference = {dp_diff:.4f} "
f"(soglia: {threshold}). Il gruppo svantaggiato riceve outcome positivi "
f"meno frequentemente del {abs(dp_diff)*100:.1f}%."
)
if not results["equalized_odds_pass"]:
messages.append(
f"ATTENZIONE: Equalized Odds Difference = {eo_diff:.4f} "
f"(soglia: {threshold}). I tassi di errore variano significativamente tra gruppi."
)
if results["is_fair"]:
messages.append("OK: Tutte le metriche di fairness rientrano nella soglia accettabile.")
results["messages"] = messages
return results
def mitigate_bias_with_reductions(
estimator,
X_train: pd.DataFrame,
y_train: pd.Series,
sensitive_train: pd.Series,
constraint: str = "demographic_parity"
) -> object:
"""
Mitigazione del bias tramite Exponentiated Gradient (Fairlearn).
Restituisce un modello fairness-aware.
Args:
constraint: "demographic_parity" o "equalized_odds"
"""
if constraint == "demographic_parity":
fairness_constraint = DemographicParity(difference_bound=0.05)
else:
fairness_constraint = EqualizedOdds(difference_bound=0.05)
mitigator = ExponentiatedGradient(
estimator=estimator,
constraints=fairness_constraint,
max_iter=50,
nu=1e-6
)
mitigator.fit(
X_train,
y_train,
sensitive_features=sensitive_train
)
print(f"Mitigazione completata con constraint: {constraint}")
print(f"N. classificatori nell'ensemble: {len(mitigator.predictors_)}")
return mitigator
def log_fairness_to_mlflow(fairness_results: dict, run_id: str):
"""Registra i risultati di fairness in MLflow per audit trail."""
with mlflow.start_run(run_id=run_id):
mlflow.log_metrics({
"fairness.demographic_parity_diff": fairness_results["demographic_parity_difference"],
"fairness.demographic_parity_ratio": fairness_results["demographic_parity_ratio"],
"fairness.equalized_odds_diff": fairness_results["equalized_odds_difference"],
"fairness.equalized_odds_ratio": fairness_results["equalized_odds_ratio"],
})
mlflow.set_tag("fairness.is_fair", str(fairness_results["is_fair"]))
# Log report completo come artefatto
import json
report_path = "/tmp/fairness_report.json"
with open(report_path, "w") as f:
# Serializza numpy floats
serializable = {
k: v for k, v in fairness_results.items()
if isinstance(v, (dict, list, bool, str, float, int))
}
json.dump(serializable, f, indent=2, default=float)
mlflow.log_artifact(report_path, artifact_path="governance/fairness")
print("Fairness report loggato in MLflow.")Parytet demograficzny a wyrównane szanse: którego użyć?
- Parytet demograficzny: idealny, gdy nie oczekuje się żadnej różnicy w rozkładach między grupami (np. zatwierdzenie kredytu: nie może być różnicy w stawkach między mężczyznami i kobietami). niezależnie od rzeczywistej wypłacalności). May penalize overall accuracy.
- Wyrównane szanse: idealne, gdy rzeczywiste etykiety mogą różnić się w zależności od grupy (np. badania lekarskie: różne wskaźniki zachorowań w zależności od wieku). Gwarantuje fałszywe alarmy i fałszywe negatywy są rozdzielane równomiernie.
- Ostrzeżenie (twierdzenie o niemożliwości): parytet demograficzny i wyrównane szanse nie mogą być spełnione jednocześnie, chyba że stawki podstawowe są identyczne pomiędzy grupami. Wybierz metrykę odpowiednią do kontekstu i udokumentuj ją na karcie modelu.
Wyjaśnialność za pomocą SHAP: przejrzystość lokalna i globalna
Wyjaśnialność to nie tylko wymóg regulacyjny: to praktyczny wymóg dotyczący debugowania modeli, zdobyć zaufanie interesariuszy i wykryć ukryte uprzedzenia. KSZTAŁT (Dodatek Shapleya exPlanations) stał się de facto standardem wyjaśnialności w ML, ponieważ oferuje właściwości solidna matematyka: spójność, lokalna dokładność i obsługa modeli opartych na drzewach (z wariantem TreeSHAP, O(TLD^2) zamiast O(TL2^M) w wersji ogólnej).
SHAP oblicza marginalny udział każdej cechy w prognozie w oparciu o teorię gier spółdzielnia (wartości Shapleya). Wynikiem jest wartość SHAP dla każdej cechy każdej próbki: wartości dodatnie zwiększają prognozę, wartości ujemne ją zmniejszają. Suma wartości SHAP wszystkich cech i równa różnicy między przewidywaniem modelu a przewidywaniem średnim (wartość bazowa).
# explainability_shap.py
# Explainability completa con SHAP per modelli ML in produzione
# pip install shap matplotlib pandas scikit-learn
import shap
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use('Agg') # Backend non-interattivo per ambienti server
import matplotlib.pyplot as plt
import mlflow
import json
from pathlib import Path
class MLExplainer:
"""
Classe per gestire l'explainability SHAP in produzione.
Supporta TreeSHAP (XGBoost, LightGBM, RandomForest) e KernelSHAP (any model).
"""
def __init__(self, model, model_type: str = "tree", background_data=None):
"""
Args:
model: Modello addestrato
model_type: "tree" per modelli tree-based, "linear" per lineari, "kernel" per altri
background_data: Dati di background per KernelSHAP (sample del training set)
"""
self.model = model
self.model_type = model_type
if model_type == "tree":
# TreeSHAP: molto più veloce, non richiede background data
self.explainer = shap.TreeExplainer(model)
elif model_type == "linear":
self.explainer = shap.LinearExplainer(model, background_data)
else:
# KernelSHAP: universale ma lento, usa un sample piccolo come background
if background_data is None:
raise ValueError("KernelSHAP richiede background_data (sample del training set)")
bg_summary = shap.kmeans(background_data, 50) # Riduce a 50 cluster
self.explainer = shap.KernelExplainer(model.predict_proba, bg_summary)
self.shap_values = None
self.feature_names = None
def compute_shap_values(self, X: pd.DataFrame) -> np.ndarray:
"""Calcola i valori SHAP per un dataset."""
self.feature_names = list(X.columns)
shap_output = self.explainer(X)
# Per classificazione binaria, prendi la classe positiva
if hasattr(shap_output, 'values'):
values = shap_output.values
if len(values.shape) == 3: # [samples, features, classes]
self.shap_values = values[:, :, 1] # Classe positiva
else:
self.shap_values = values
else:
self.shap_values = shap_output
return self.shap_values
def global_importance(self, top_n: int = 15) -> pd.DataFrame:
"""
Feature importance globale: media dei valori SHAP assoluti.
Questa e la visione che interessa ai regolatori (AI Act, XAI requirement).
"""
if self.shap_values is None:
raise RuntimeError("Esegui compute_shap_values() prima.")
mean_abs_shap = np.abs(self.shap_values).mean(axis=0)
importance_df = pd.DataFrame({
"feature": self.feature_names,
"mean_abs_shap": mean_abs_shap
}).sort_values("mean_abs_shap", ascending=False)
return importance_df.head(top_n)
def explain_single_prediction(
self, sample: pd.Series, threshold: float = 0.5
) -> dict:
"""
Spiegazione locale per una singola predizione.
Utile per audit di decisioni individuali (obbligatorio AI Act per High-Risk).
"""
sample_df = pd.DataFrame([sample])
single_shap = self.explainer(sample_df)
if hasattr(single_shap, 'values'):
values = single_shap.values[0]
if len(values.shape) == 2: # [features, classes]
values = values[:, 1]
base_value = float(single_shap.base_values[0]) if len(single_shap.base_values.shape) > 1 \
else float(single_shap.base_values[0])
else:
values = single_shap[0]
base_value = float(self.explainer.expected_value)
# Predizione del modello
pred_proba = self.model.predict_proba(sample_df)[0, 1]
pred_class = int(pred_proba >= threshold)
# Feature contributions ordinate per importanza
contributions = sorted(
zip(self.feature_names, values, sample.values),
key=lambda x: abs(x[1]),
reverse=True
)
return {
"prediction_probability": float(pred_proba),
"prediction_class": pred_class,
"base_value": base_value,
"top_contributing_features": [
{
"feature": feat,
"shap_value": float(shap_val),
"feature_value": float(feat_val) if isinstance(feat_val, (int, float, np.number)) else str(feat_val),
"direction": "positive" if shap_val > 0 else "negative"
}
for feat, shap_val, feat_val in contributions[:10]
]
}
def save_summary_plot(self, X: pd.DataFrame, output_path: str):
"""Salva summary plot SHAP per documentazione/governance."""
if self.shap_values is None:
self.compute_shap_values(X)
plt.figure(figsize=(10, 8))
shap.summary_plot(
self.shap_values,
X,
plot_type="bar",
show=False,
max_display=15
)
plt.tight_layout()
plt.savefig(output_path, dpi=150, bbox_inches='tight')
plt.close()
print(f"Summary plot salvato in: {output_path}")
def log_to_mlflow(self, X_sample: pd.DataFrame, run_id: str):
"""Registra explainability artifacts in MLflow per audit trail."""
if self.shap_values is None:
self.compute_shap_values(X_sample)
with mlflow.start_run(run_id=run_id):
# Global importance come JSON
importance = self.global_importance()
importance_path = "/tmp/shap_importance.json"
importance.to_json(importance_path, orient="records", indent=2)
mlflow.log_artifact(importance_path, artifact_path="governance/explainability")
# Summary plot
plot_path = "/tmp/shap_summary_plot.png"
self.save_summary_plot(X_sample, plot_path)
mlflow.log_artifact(plot_path, artifact_path="governance/explainability")
# Top feature come metric (per dashboard governance)
for i, row in importance.head(5).iterrows():
feat_name = row["feature"].replace(" ", "_").replace("-", "_")
mlflow.log_metric(
f"shap.top_feature_importance.{feat_name}",
float(row["mean_abs_shap"])
)
print(f"Explainability artifacts registrati in MLflow run {run_id}")Ścieżka audytu: Niezmienne dzienniki zgodności
Ustawa o sztucznej inteligencji wymaga, aby systemy wysokiego ryzyka automatycznie rejestrowały odpowiednie zdarzenia podczas ich działania, z rejestrami wystarczająco szczegółowymi, aby umożliwić to władzom sprawdzić zgodność. Te logi muszą być niezmienne, oznaczone datą i identyfikowalne. Solidny system ścieżki audytu obejmuje trzy poziomy: Dziennik prognoz dzienniki indywidualne, dziennik przejścia etapu modelu i dziennik zdarzeń związanych z zarządzaniem.
# audit_logger.py
# Sistema di audit trail conforme AI Act per modelli in produzione
# Usa struttura di log immodificabile con hash SHA-256 per integrità
import hashlib
import json
import logging
import datetime
import uuid
from typing import Optional
import structlog # pip install structlog
import mlflow
# Configurazione structlog per log strutturati in JSON
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.add_log_level,
structlog.processors.JSONRenderer()
]
)
audit_log = structlog.get_logger("audit")
class MLAuditLogger:
"""
Sistema di audit trail per modelli ML in produzione.
Ogni log entry include un hash SHA-256 del contenuto per rilevare manomissioni.
"""
def __init__(self, model_name: str, model_version: str, log_file: str = None):
self.model_name = model_name
self.model_version = model_version
self.log_file = log_file or f"audit_{model_name}_v{model_version}.jsonl"
# Logger Python standard per file locale (append-only)
self.file_logger = logging.getLogger(f"audit.{model_name}")
if not self.file_logger.handlers:
handler = logging.FileHandler(self.log_file, mode='a')
handler.setFormatter(logging.Formatter('%(message)s'))
self.file_logger.addHandler(handler)
self.file_logger.setLevel(logging.INFO)
def _compute_hash(self, entry: dict) -> str:
"""Calcola hash SHA-256 dell'entry per integrità."""
content = json.dumps(entry, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(content.encode()).hexdigest()
def log_prediction(
self,
request_id: str,
input_features: dict,
prediction: float,
prediction_class: int,
shap_explanation: Optional[dict] = None,
user_id: Optional[str] = None,
session_id: Optional[str] = None
):
"""
Log di una singola predizione.
Per AI Act High-Risk: OGNI decisione deve essere registrata.
"""
entry = {
"event_type": "PREDICTION",
"event_id": str(uuid.uuid4()),
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"model_name": self.model_name,
"model_version": self.model_version,
"request_id": request_id,
"user_id": user_id or "anonymous",
"session_id": session_id,
"input": input_features,
"output": {
"prediction_probability": float(prediction),
"prediction_class": int(prediction_class),
"decision": "POSITIVE" if prediction_class == 1 else "NEGATIVE"
},
"explanation": shap_explanation, # Top SHAP features per spiegazione
}
entry["content_hash"] = self._compute_hash(
{k: v for k, v in entry.items() if k != "content_hash"}
)
self.file_logger.info(json.dumps(entry, ensure_ascii=False))
audit_log.info("prediction_logged", request_id=request_id, decision=entry["output"]["decision"])
return entry["event_id"]
def log_model_stage_transition(
self,
from_stage: str,
to_stage: str,
approved_by: str,
justification: str,
metrics: Optional[dict] = None
):
"""
Log di una transizione di stage del modello (Staging -> Production).
Crea un audit trail delle approvazioni richiesto dal processo di governance.
"""
entry = {
"event_type": "STAGE_TRANSITION",
"event_id": str(uuid.uuid4()),
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"model_name": self.model_name,
"model_version": self.model_version,
"from_stage": from_stage,
"to_stage": to_stage,
"approved_by": approved_by,
"justification": justification,
"performance_metrics": metrics or {},
}
entry["content_hash"] = self._compute_hash(
{k: v for k, v in entry.items() if k != "content_hash"}
)
self.file_logger.info(json.dumps(entry, ensure_ascii=False))
audit_log.info(
"stage_transition",
model=self.model_name,
from_stage=from_stage,
to_stage=to_stage,
approved_by=approved_by
)
# Registra anche in MLflow Model Registry per tracciabilita centralizzata
client = mlflow.tracking.MlflowClient()
try:
client.set_registered_model_alias(
name=self.model_name,
alias=to_stage.lower(),
version=self.model_version
)
client.set_model_version_tag(
name=self.model_name,
version=self.model_version,
key="governance.approved_by",
value=approved_by
)
client.set_model_version_tag(
name=self.model_name,
version=self.model_version,
key="governance.approval_date",
value=datetime.date.today().isoformat()
)
except Exception as e:
audit_log.warning("mlflow_tag_failed", error=str(e))
def log_governance_event(
self,
event_subtype: str,
details: dict,
actor: str
):
"""
Log generico per eventi di governance: fairness check, model card update,
bias detection alert, human override, etc.
"""
entry = {
"event_type": "GOVERNANCE",
"event_subtype": event_subtype,
"event_id": str(uuid.uuid4()),
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"model_name": self.model_name,
"model_version": self.model_version,
"actor": actor,
"details": details,
}
entry["content_hash"] = self._compute_hash(
{k: v for k, v in entry.items() if k != "content_hash"}
)
self.file_logger.info(json.dumps(entry, ensure_ascii=False))
audit_log.info("governance_event", subtype=event_subtype, actor=actor)
def verify_log_integrity(self) -> dict:
"""
Verifica l'integrita del file di log ricalcolando gli hash.
Rileva eventuali manomissioni dei log.
"""
results = {"total": 0, "valid": 0, "tampered": [], "errors": []}
try:
with open(self.log_file, 'r') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
results["total"] += 1
try:
entry = json.loads(line)
stored_hash = entry.pop("content_hash", None)
recomputed = self._compute_hash(entry)
if stored_hash == recomputed:
results["valid"] += 1
else:
results["tampered"].append({
"line": line_num,
"event_id": entry.get("event_id"),
"timestamp": entry.get("timestamp")
})
except json.JSONDecodeError as e:
results["errors"].append({"line": line_num, "error": str(e)})
except FileNotFoundError:
results["errors"].append({"error": f"Log file non trovato: {self.log_file}"})
integrity_ok = len(results["tampered"]) == 0 and len(results["errors"]) == 0
results["integrity_ok"] = integrity_ok
if not integrity_ok:
audit_log.error(
"log_integrity_violated",
tampered_count=len(results["tampered"]),
error_count=len(results["errors"])
)
return resultsDziennik audytu: przechowywanie i ochrona
Ustawa o AI nie określa dokładnego okresu przechowywania logów z systemów wysokiego ryzyka, ale wytyczne wskazują, że należy je zachować przynajmniej przez cały okres użytkowania systemu, zazwyczaj 3–10 lat. Przechowuj logi w niezmiennej pamięci (np. S3 z Blokada obiektu, GCS z polityką przechowywania lub baza danych tylko do dołączania). Nigdy nie korzystaj z bazy danych standardowa relacja relacyjna jako pojedynczy magazyn ścieżki audytu: wiersze można usuwać lub zmodyfikowany. Zaszyfrowana struktura SHA-256 w powyższym kodzie umożliwia wykrycie sabotażu a posteriori, ale nie zapobiega to fizycznemu usunięciu plików.
Modelowe zarządzanie rejestrem za pomocą MLflow
Il Rejestr modelu MLflow oraz centralny element zarządzania: centralizacja wszystkie produkowane modele wraz z ich cyklem życia, od wersji eksperymentalnych po modele w fazie produkcyjnej, z pełną historią przejść, komentarzami i znacznikami zarządzania. Dzięki MLflow 3.0, wprowadzony w 2025 r. rejestr natywnie obsługuje aliasy modeli (produkcja, inscenizacja, mistrz, challenger), aby zastąpić przestarzałe etapy i integruje się z katalogiem Unity firmy Databricks dla zarządzanie między obszarami roboczymi.
# model_registry_governance.py
# Workflow di governance per MLflow Model Registry
# Include: registrazione, review, promozione con approvazione, deprecazione
import mlflow
from mlflow.tracking import MlflowClient
from mlflow.exceptions import MlflowException
import datetime
from typing import Optional
from dataclasses import dataclass
MLFLOW_TRACKING_URI = "http://localhost:5000"
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
client = MlflowClient()
@dataclass
class GovernanceConfig:
"""Configurazione di governance per un modello."""
model_name: str
required_min_accuracy: float = 0.80
required_fairness_threshold: float = 0.10
required_approvers: list = None
risk_category: str = "minimal-risk"
def __post_init__(self):
if self.required_approvers is None:
self.required_approvers = ["ml-lead", "compliance-officer"]
def register_model_with_governance(
run_id: str,
model_path: str,
config: GovernanceConfig,
model_card_path: str,
fairness_report_path: str,
) -> str:
"""
Registra un modello nel Model Registry con tutti i tag di governance.
Returns:
version: Versione registrata del modello
"""
# Registra il modello
model_uri = f"runs:/{run_id}/{model_path}"
result = mlflow.register_model(
model_uri=model_uri,
name=config.model_name
)
version = result.version
# Tag obbligatori per governance
governance_tags = {
"governance.risk_category": config.risk_category,
"governance.registration_date": datetime.date.today().isoformat(),
"governance.registration_author": "automated-pipeline",
"governance.status": "PENDING_REVIEW",
"governance.model_card_logged": "true",
"governance.fairness_checked": "true",
"governance.explainability_logged": "true",
"governance.required_approvers": ",".join(config.required_approvers),
"source.run_id": run_id,
}
for key, value in governance_tags.items():
client.set_model_version_tag(
name=config.model_name,
version=version,
key=key,
value=value
)
# Alias iniziale: "candidate" (in attesa di review)
client.set_registered_model_alias(
name=config.model_name,
alias="candidate",
version=version
)
print(f"Modello registrato: {config.model_name} v{version} - Status: PENDING_REVIEW")
return version
def approve_for_staging(
model_name: str,
version: str,
approver: str,
justification: str,
performance_metrics: dict,
config: GovernanceConfig
) -> bool:
"""
Promuove un modello in Staging dopo review dei gate di qualità.
Verifica automaticamente accuracy e fairness prima dell'approvazione.
"""
# ---- Gate 1: Accuracy minima ----
accuracy = performance_metrics.get("accuracy", 0)
if accuracy < config.required_min_accuracy:
print(f"GATE FALLITO: accuracy {accuracy:.4f} < soglia {config.required_min_accuracy}")
client.set_model_version_tag(
name=model_name, version=version,
key="governance.rejection_reason",
value=f"Accuracy insufficiente: {accuracy:.4f}"
)
return False
# ---- Gate 2: Fairness ----
dp_diff = performance_metrics.get("demographic_parity_difference", 999)
if abs(dp_diff) > config.required_fairness_threshold:
print(f"GATE FALLITO: Fairness violation - DP diff {dp_diff:.4f}")
client.set_model_version_tag(
name=model_name, version=version,
key="governance.rejection_reason",
value=f"Fairness violation: DP diff {dp_diff:.4f}"
)
return False
# ---- Gate 3: Approvazione umana ----
# In produzione, questo step richiede un sistema di ticketing/approval workflow
# Per ora, verifica che l'approver sia autorizzato
if approver not in config.required_approvers:
print(f"GATE FALLITO: Approver non autorizzato: {approver}")
return False
# Tutti i gate superati: promuovi a Staging
approval_tags = {
"governance.status": "STAGING",
"governance.approved_by": approver,
"governance.approval_date": datetime.datetime.utcnow().isoformat(),
"governance.justification": justification,
f"governance.metrics.accuracy": str(accuracy),
f"governance.metrics.dp_diff": str(dp_diff),
}
for key, value in approval_tags.items():
client.set_model_version_tag(
name=model_name, version=version,
key=key, value=value
)
# Aggiorna alias
client.set_registered_model_alias(
name=model_name,
alias="staging",
version=version
)
print(f"Modello {model_name} v{version} promosso in STAGING da {approver}")
return True
def promote_to_production(
model_name: str,
version: str,
approver: str,
champion_strategy: str = "blue-green"
) -> bool:
"""
Promuove il modello in Production.
Implementa champion/challenger: mantiene la versione precedente come challenger.
"""
# Trova versione production corrente (champion)
try:
current_champion = client.get_model_version_by_alias(model_name, "production")
old_champion_version = current_champion.version
# Marca il vecchio champion come challenger
client.set_registered_model_alias(
name=model_name,
alias="challenger",
version=old_champion_version
)
client.set_model_version_tag(
name=model_name, version=old_champion_version,
key="governance.status", value="CHALLENGER"
)
except MlflowException:
# Nessun champion precedente (primo deploy)
old_champion_version = None
# Promuovi nuova versione a champion
production_tags = {
"governance.status": "PRODUCTION",
"governance.production_date": datetime.datetime.utcnow().isoformat(),
"governance.production_approver": approver,
"governance.deployment_strategy": champion_strategy,
"governance.previous_version": str(old_champion_version) if old_champion_version else "none",
}
for key, value in production_tags.items():
client.set_model_version_tag(
name=model_name, version=version,
key=key, value=value
)
client.set_registered_model_alias(
name=model_name,
alias="production",
version=version
)
print(f"Modello {model_name} v{version} promosso in PRODUCTION (champion)")
if old_champion_version:
print(f"Versione precedente v{old_champion_version} mantenuta come CHALLENGER")
return True
def get_governance_report(model_name: str) -> dict:
"""
Genera un report di governance completo per un modello registrato.
Utile per audit esterni e report di compliance.
"""
registered_model = client.get_registered_model(model_name)
versions = client.search_model_versions(f"name='{model_name}'")
report = {
"model_name": model_name,
"report_date": datetime.date.today().isoformat(),
"total_versions": len(versions),
"versions_detail": []
}
for v in versions:
tags = v.tags
report["versions_detail"].append({
"version": v.version,
"status": tags.get("governance.status", "UNKNOWN"),
"risk_category": tags.get("governance.risk_category", "UNKNOWN"),
"registration_date": tags.get("governance.registration_date", ""),
"approved_by": tags.get("governance.approved_by", ""),
"approval_date": tags.get("governance.approval_date", ""),
"fairness_checked": tags.get("governance.fairness_checked", "false"),
"model_card_logged": tags.get("governance.model_card_logged", "false"),
})
return reportOdpowiedzialne ramy AI: lista kontrolna dla projektów ML
Ramy operacyjne odpowiedzialnej sztucznej inteligencji nie muszą być dokumentem teoretycznym: muszą to być: konkretna lista kontrolna zintegrowana z procesem rozwoju. Poniższe ramy opierają się na zasadach ustawy AI, wytyczne NIST AI RMF (Risk Management Framework) i najlepsze praktyki spółki z najbardziej zaawansowanych zespołów ML. Zaprojektowany tak, aby był wykonywalny za pomocą narzędzi typu open source przy zerowym koszcie.
# responsible_ai_framework.py
# Framework operativo per Responsible AI
# Integra tutti i componenti precedenti in una pipeline CI/CD-ready
import mlflow
import pandas as pd
import numpy as np
from typing import Optional
from pathlib import Path
import json
import datetime
# Import dei moduli sviluppati in questo articolo
# from model_card_generator import generate_model_card, log_model_card_to_mlflow
# from fairness_checker import run_fairness_analysis, log_fairness_to_mlflow
# from explainability_shap import MLExplainer
# from audit_logger import MLAuditLogger
# from model_registry_governance import GovernanceConfig, register_model_with_governance
class ResponsibleAIPipeline:
"""
Pipeline completa di Responsible AI per training e deployment di modelli ML.
Integra: model card, fairness check, explainability, audit trail, governance.
"""
def __init__(
self,
model_name: str,
model_version: str,
risk_category: str,
sensitive_features: list,
output_dir: str = "./governance_output"
):
self.model_name = model_name
self.model_version = model_version
self.risk_category = risk_category
self.sensitive_features = sensitive_features
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.audit_logger = MLAuditLogger(
model_name=model_name,
model_version=model_version,
log_file=str(self.output_dir / f"audit_{model_name}.jsonl")
)
self.governance_report = {
"model_name": model_name,
"version": model_version,
"risk_category": risk_category,
"evaluation_date": datetime.date.today().isoformat(),
"checks": {}
}
def run_full_governance_check(
self,
model,
X_test: pd.DataFrame,
y_test: pd.Series,
run_id: str,
config: dict
) -> dict:
"""
Esegue tutti i controlli di governance in sequenza.
Restituisce report completo con esito di ogni check.
"""
print(f"\n=== RESPONSIBLE AI GOVERNANCE CHECK ===")
print(f"Modello: {self.model_name} v{self.model_version}")
print(f"Risk Category: {self.risk_category}\n")
# ---- 1. Model Card ----
print("[1/5] Generazione Model Card...")
try:
card = generate_model_card(
model_name=self.model_name,
model_version=self.model_version,
model=model,
X_test=X_test,
y_test=y_test,
sensitive_cols=self.sensitive_features,
intended_use=config.get("intended_use", ""),
config=config
)
log_model_card_to_mlflow(card, run_id)
self.governance_report["checks"]["model_card"] = "PASS"
print(" OK: Model Card generata e loggata")
except Exception as e:
self.governance_report["checks"]["model_card"] = f"FAIL: {str(e)}"
print(f" FAIL: {e}")
# ---- 2. Fairness Analysis ----
print("[2/5] Fairness Analysis...")
fairness_results_all = {}
fairness_pass = True
y_pred = model.predict(X_test)
for feat in self.sensitive_features:
if feat not in X_test.columns:
continue
try:
results = run_fairness_analysis(
y_true=y_test,
y_pred=y_pred,
sensitive_feature=X_test[feat],
threshold=config.get("fairness_threshold", 0.10)
)
fairness_results_all[feat] = results
log_fairness_to_mlflow(results, run_id)
if not results["is_fair"]:
fairness_pass = False
for msg in results["messages"]:
print(f" {msg}")
else:
print(f" OK: Fairness per '{feat}' nei limiti accettabili")
except Exception as e:
print(f" WARNING: Fairness check per '{feat}' fallito: {e}")
self.governance_report["checks"]["fairness"] = "PASS" if fairness_pass else "FAIL"
# ---- 3. Explainability ----
print("[3/5] Explainability (SHAP)...")
try:
explainer = MLExplainer(
model=model,
model_type=config.get("model_type", "tree")
)
# Usa un sample per efficienza
sample_size = min(500, len(X_test))
X_sample = X_test.sample(n=sample_size, random_state=42)
explainer.compute_shap_values(X_sample)
explainer.log_to_mlflow(X_sample, run_id)
top_features = explainer.global_importance(top_n=5)
print(f" OK: Top 5 features: {', '.join(top_features['feature'].tolist())}")
self.governance_report["checks"]["explainability"] = "PASS"
except Exception as e:
self.governance_report["checks"]["explainability"] = f"FAIL: {str(e)}"
print(f" FAIL: {e}")
# ---- 4. Risk Assessment ----
print("[4/5] Risk Assessment...")
risk_assessment = self._run_risk_assessment(config)
self.governance_report["risk_assessment"] = risk_assessment
if risk_assessment["risk_score"] > 0.7:
print(f" ATTENZIONE: Risk score alto ({risk_assessment['risk_score']:.2f})")
self.governance_report["checks"]["risk_assessment"] = "HIGH_RISK"
else:
print(f" OK: Risk score accettabile ({risk_assessment['risk_score']:.2f})")
self.governance_report["checks"]["risk_assessment"] = "PASS"
# ---- 5. Audit Log dell'evento di governance ----
print("[5/5] Audit Trail...")
self.audit_logger.log_governance_event(
event_subtype="FULL_GOVERNANCE_CHECK",
details={
"checks": self.governance_report["checks"],
"risk_score": risk_assessment["risk_score"],
"fairness_pass": fairness_pass,
"run_id": run_id
},
actor="automated-pipeline"
)
self.governance_report["checks"]["audit_trail"] = "PASS"
print(" OK: Audit trail registrato")
# Report finale
all_pass = all(
v == "PASS" for v in self.governance_report["checks"].values()
)
self.governance_report["overall_status"] = "APPROVED" if all_pass else "REQUIRES_REVIEW"
print(f"\n=== RISULTATO: {self.governance_report['overall_status']} ===\n")
# Salva report
report_path = self.output_dir / f"governance_report_{self.model_name}_v{self.model_version}.json"
with open(report_path, "w") as f:
json.dump(self.governance_report, f, indent=2, ensure_ascii=False)
return self.governance_report
def _run_risk_assessment(self, config: dict) -> dict:
"""
Calcola un risk score composito basato sul contesto del modello.
Segue la tassonomia EU AI Act.
"""
score = 0.0
factors = []
# Categoria rischio
if self.risk_category == "high-risk":
score += 0.4
factors.append("High-risk AI Act category: +0.4")
elif self.risk_category == "limited-risk":
score += 0.2
factors.append("Limited-risk AI Act category: +0.2")
# Features sensibili
if len(self.sensitive_features) > 0:
score += 0.1 * min(len(self.sensitive_features), 3)
factors.append(f"{len(self.sensitive_features)} sensitive features: +{0.1 * min(len(self.sensitive_features), 3)}")
# Uso in decisioni automatiche
if config.get("automated_decisions", False):
score += 0.2
factors.append("Decisioni automatiche senza revisione umana: +0.2")
# Dati personali
if config.get("processes_personal_data", False):
score += 0.1
factors.append("Elabora dati personali: +0.1")
return {
"risk_score": min(score, 1.0),
"risk_level": "HIGH" if score > 0.6 else ("MEDIUM" if score > 0.3 else "LOW"),
"factors": factors
}Najlepsze praktyki i antywzorce w zakresie zarządzania systemem uczenia maszynowego
Po zapoznaniu się z narzędziami technicznymi konieczne jest zrozumienie, w jaki sposób włączyć zarządzanie do codzienne życie zespołu ML bez tworzenia niezrównoważonych kosztów ogólnych. Skuteczne zarządzanie musi być w pełni zautomatyzowany i zintegrowany z potokiem CI/CD, a nie ręczny proces dodawany po fakcie.
Najlepsze praktyki w zakresie zarządzania systemem uczenia maszynowego
- Zarządzanie jako kod: integruje wszystkie kontrole (karta wzorcowa, rzetelność, wyjaśnialność) w rurociągu CI/CD jako kroki automatyczne. Jeśli sprawdzenie poprawności nie powiedzie się, wdrożenie nie nastąpi. Zero ręcznych decyzji dla bramek technicznych, nadzór człowieka tylko dla bramek organizacyjnych.
- Scentralizuj w rejestrze modeli: Rejestr modelu MLflow jest jedynym źródłem prawdy. Każda wersja musi posiadać ustandaryzowane tagi: kategoria_ryzyka, zatwierdzona przez, uczciwość_sprawdzona. Zabroń wdrażania dowolnej wersji bez wymaganych tagów.
- Dokumentuj decyzje, a nie tylko wyniki: zapisz także kiedy decydujesz się zaakceptować błąd resztkowy z uzasadnieniem (np. „Zaakceptowano różnicę DP 0,12 ponieważ przekracza próg tylko dla podgrupy X z n=45, nieistotne statystycznie”).
- Mistrz/Challenger zawsze: nigdy nie usuwaj poprzedniej wersji pliku szablon z rejestru podczas promowania nowego. Zachowaj to jako wyzwanie do wycofania się natychmiastowe w przypadku problemów z rzetelnością lub wydajnością produkcji.
- Przegląd okresowy pod kątem odchyleń od sprawiedliwości: nastawienie modelu może się zmienić w miarę upływu czasu, jeśli zmieni się rozkład danych wejściowych. Powtarzaj analizę uczciwości co miesiąc na danych produkcyjnych, a nie tylko przy pierwszym wdrożeniu.
- Człowiek w pętli w przypadku wysokiego ryzyka: w przypadku systemów wysokiego ryzyka AI Act należy wdrożyć mechanizm flagowania, który wysyła decyzje obarczone wysokim ryzykiem (np. prawdopodobieństwo od 0,45 do 0.55) operatorowi do ręcznego przeglądu przed przekazaniem ich użytkownikowi końcowemu.
Anty-wzorce, których należy unikać
- Kontrola rzetelności tylko podczas szkolenia: Uczciwość w produkcji zmienia się z biegiem czasu. Model, który jest uczciwy w szkoleniu, może stać się nieuczciwy, jeśli dane produkcyjne ulegną zmianie (dryf danych). Stale monitoruj uczciwość za pomocą narzędzi Evidently lub niestandardowych.
- Karta modelu statycznego: karta modelu napisana przy pierwszym wdrożeniu i już przestarzała druga aktualizacja. Automatyzuj generowanie przy każdym nowym szkoleniu: nie może to być: Dokument programu Word zaktualizowany ręcznie.
- SHAP na całym zestawie treningowym: TreeSHAP na zbiorach danych obejmujących miliony próbek może zająć wiele godzin. Zawsze używaj reprezentatywnej próbki (500–2000 próbek). globalne poglądy. W przypadku lokalnych wyjaśnień w produkcji należy obliczyć SHAP tylko dla przewidywania wymagające wyjaśnienia (wysokie ryzyko lub bliskie progu decyzji).
- Dziennik audytu w bazie transakcyjnej: standardowa baza danych SQL umożliwia AKTUALIZACJĘ i USUŃ. W przypadku dzienników kontrolnych AI Act nie używaj edytowalnych tabel SQL: używaj plików przeznaczonych wyłącznie do dodawania w pamięci niezmiennej z hashem integralności.
- Zatwierdzenia przez e-mail/czat: należy śledzić zatwierdzenia ze strony władz w systemie (tagi MLflow, bilety Jira itp.), a nie w wątkach Slack czy mailach. W audycie regulacyjnych „Marco napisał w aplikacji Teams, że wszystko w porządku” nie jest wystarczającą dokumentacją.
Budżet dla MŚP: Zarządzanie ML z kwotą mniejszą niż 5000 EUR/rok
Ład ML nie wymaga budżetów przedsiębiorstwa. Cały framework opisany w tym artykule i w 100% open source:
- Przepływ ML: darmowy, hostowany na jednym serwerze (wystarczą 2 vCPU, 4GB RAM). dla małych zespołów). Koszt: ~15-20 EUR/miesiąc w chmurze.
- Fairlearn + SHAP + scikit-learn: Otwarte, bezpłatne biblioteki Pythona licencjonowanie.
- Dzienniki audytu: JSONL w pamięci obiektowej zgodnej z S3 (samodzielnie hostowany MinIO lub chmura). 100 GB logów: około 2-3 EUR/miesiąc.
- Prometeusz + Grafana do monitorowania wskaźników uczciwości w produkcji: bezpłatny, do zainstalowania na K3 lub Docker Compose.
Razem: mniej niż 300 EUR/rok kompleksowego, zgodnego z przepisami systemu zarządzania zgodnie z wymogami ustawy o sztucznej inteligencji, z automatycznymi kartami wzorcowymi, sprawdzaniem rzetelności, wyjaśnialnością SHAP i niezmienna ścieżka audytu.
Wnioski
Zarządzanie systemem uczenia się w 2026 r. nie będzie już opcjonalne dla podmiotów opracowujących systemy sztucznej inteligencji w UE: w ustawie o sztucznej inteligencji ustanowiono konkretne wymagania wraz z surowymi sankcjami. Ale patrząc poza przestrzeganie zasad, ramy dobrego zarządzania przynoszą konkretne korzyści: solidniejsze modele dzięki sprawiedliwości analiza, szybsze debugowanie dzięki SHAP (31% szybsze według danych branżowych), większe zaufanie interesariuszy i bezpieczne wycofanie w przypadku problemów.
Podejście opisane w tym artykule, całkowicie otwarte i zautomatyzowane w CI/CD, umożliwia zespołom dowolnej wielkości wdrożenie zarządzania uczeniem się maszynowym bez nadmiernych kosztów ogólnych. Klucz i automatyzacja: generowane kodem karty modeli, kontrole rzetelności jako bramki CI, audyty szlak jako obsługujący wózek boczny, SHAP obliczany przy każdym rozmieszczeniu. Zarządzanie staje się częścią proces inżynieryjny, a nie dokument do wypełnienia przed audytem.
W kolejnym i ostatnim artykule z serii pt Studium przypadku Przewidywanie rezygnacji w produkcji (ID 315), zintegrujemy wszystkie komponenty serii: śledzenie eksperymentalne MLflow, DVC do wersjonowania, Obsługa FastAPI, wdrożenie Kubernetes, monitorowanie za pomocą Prometheusa i ramy zarządzania opisane w tym artykule, w jednym działającym, kompleksowym potoku.
Powiązane artykuły z tej serii MLOps
- MLOps: od eksperymentu do produkcji - Podstawy i pełny cykl życia
- Potok ML z CI/CD: GitHub Actions + Docker - Zintegruj bramy zarządzania z CI
- Śledzenie eksperymentów za pomocą MLflow - Rejestr modelowy jako podstawa zarządzania
- Wykrywanie dryfu modelu i automatyczne ponowne uczenie - Dryfowanie i monitorowanie uczciwości
- Skalowanie ML na Kubernetesie - Wdróż ramy zarządzania na K8
- Testowanie A/B modeli ML - Zarządzanie mistrzem/pretendentem
Seria krzyżowa
- Zaawansowana seria głębokiego uczenia się - Zarządzanie modelami głębokiego uczenia się i LLM
- Seria biznesowa dotycząca danych i sztucznej inteligencji - Zarządzanie sztuczną inteligencją w kontekście biznesowym