ML Governance: Compliance, Audit Trail a zodpovědná AI ve výrobě
Váš model předpovědi odchodu dosahuje 92% přesnosti, poskytování FastAPI je stabilní, cluster Kubernetes se automaticky škáluje. Pak přijde email z právního oddělení: „Naše úvěrový model spadá do kategorie High-Risk podle zákona o umělé inteligenci EU. Musíme dokázat, že ne diskriminuje podle pohlaví nebo věku. Kde jsou protokoly rozhodnutí za posledních 6 měsíců? Kdo schválil nasazení ve výrobě?"
Pokud na tyto otázky nemáte okamžité odpovědi, je váš projekt ML ohrožen. The Nařízení EU o umělé inteligenci (zákon o AI), částečně vstoupil v platnost v srpnu 2025 s konečným termínem pro vysoce rizikové systémy stanoveným na 2. srpna 2026, transformovala správu praní peněz z nepovinného nejlepšího postupu na zákonnou povinnost s pokutami až do výše 6 %. ročního celosvětového obratu. Trh MLOps, v hodnotě 4,38 miliardy dolarů v roce 2026, roste o 39,8 % CAGR také proto, že společnosti se musí vybavit infrastrukturou řízení, aby zůstaly v souladu.
V tomto článku budujeme komplexní open-source rámec pro správu ML: z analýzy požadavky zákona o AI, na automatické generování vzorových karet, na audit trail s MLflow, na detekce spravedlnosti pomocí Fairlearnu až po vysvětlitelnost pomocí SHAP. Vše s funkčním kódem Python a okamžitě použitelné i s omezeným rozpočtem.
Co se naučíte
- Praktické požadavky zákona EU o umělé inteligenci pro vysoce rizikové systémy ML (termín v srpnu 2026)
- Generujte automatické modelové karty se standardizovanou dokumentací
- Implementujte komplexní auditní záznamy pomocí MLflow a strukturovaných protokolů
- Změřte a zmírněte zkreslení pomocí Fairlearnu (demografická parita, vyrovnané šance)
- Vysvětlitelnost pomocí SHAP: globální význam funkcí a místní vysvětlení
- Registr řízení modelu: přechody do fáze se sledovanými schváleními
- Rámec hodnocení rizik pro klasifikaci vašich modelů umělé inteligence
- Kontrolní seznam odpovědné umělé inteligence pro malé a střední podniky s rozpočtem nižším než 5 000 EUR/rok
EU AI Act: Jaké změny pro vaše modely ML
Zákon o umělé inteligenci klasifikuje systémy umělé inteligence do čtyř úrovní rizika s rostoucími povinnostmi. Předtím implementovat jakýkoli rámec řízení, musíte pochopit, kam do něj váš model zapadá taxonomie. Klasifikace nezávisí na použité technologii (vaše XGBoost není vnitřně více či méně riskantní než transformátor), ale odzamýšlené použití a aplikační doménu.
4 rizikové kategorie zákona o AI
Nepřijatelné riziko (zakázané): sociální skórovací systémy, podprahová manipulace,
zneužití zranitelnosti. Není možné nasazení.
Vysoké riziko (přísné požadavky): systémy pro úvěry, najímání, vzdělávání,
kritické infrastruktury, zdravotnická zařízení, migrace. Termín plnění: 2. srpna 2026.
Omezené riziko (povinná transparentnost): chatboti, deepfakes, systémy doporučení.
Povinnost informovat uživatele, který interaguje s AI.
Minimální riziko (dobrovolné): spamové filtry, AI hry, standardní doporučení.
Žádné zvláštní regulační povinnosti.
Pro systémy ve vysokém riziku, zákon o AI ukládá řadu technických povinností a organizační kroky, které tým ML potřebuje implementovat do svého kanálu MLOps. Hlavní požadavky, v platnosti od srpna 2026 se týkají:
- Systém řízení rizik: nepřetržitý proces identifikace, analýzy a zmírnění i rizika v průběhu celého životního cyklu modelu.
- Správa dat: tréninkové, validační a testovací datové soubory musí být zdokumentovány, reprezentativní, nezaujaté a vhodné pro uvedený účel.
- Technická dokumentace: dostatečnou technickou dokumentaci k prokázání soulad s dozorovými orgány. Zahrnuje architekturu, tréninkové procedury, metriky výkon a známá omezení.
- Automatické uchovávání záznamů: systém musí automaticky zaznamenávat události relevantní během operací, s neměnnými protokoly pro audit.
- Transparentnost a informace o uživateli: uživatelé potřebují vědět, kdo jsou interakce se systémem AI a získávání srozumitelných informací o jeho schopnostech a omezeních.
- Lidský dohled: technické mechanismy umožňující lidský dohled, zásah a prvořadá automatizovaná rozhodnutí.
- Přesnost, robustnost, kybernetická bezpečnost: zdokumentované metriky výkonu, Test odolnosti proti distribučním posunům a protichůdným vstupům.
Sankce AI Act: Nejsou teoretické
Sankce za nedodržení zákona o AI jsou přísné: až 30 milionů EUR nebo 6 % z toho ročního celosvětového obratu za porušení týkající se systémů s nepřijatelným rizikem; až 20 milionů EUR nebo 4 % pro jiné závazky; až 10 milionů EUR nebo 2 % za nesprávné informace úřadům. První kritický termín pro vysoce rizikové systémy a 2. srpna 2026: méně než 6 měsíců od data tohoto článku. Pokud váš model působí v úvěrové, náborové nebo kritické infrastruktuře, plán shody musí začít dnes.
Modelové karty: Standardizovaná dokumentace modelu
Le modelová karta, představený společností Google v roce 2019 a nyní přijatý jako průmyslový standard, jsou strukturované dokumenty, které popisují model ML: účel, výkon pro různé podskupiny demografické údaje, známá omezení, zamýšlené a doporučené použití. Zákon o AI je implicitně vyžaduje v sekce "technická dokumentace". Vygenerujte je ručně a náchylné k chybám: automatizuje se následující kód vytváření z metadat školení a výsledků hodnocení.
# 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/")Spravedlnost a detekce zkreslení s Fairlearn
Spravedlnost v ML není jediný koncept: matematicky existuje několik definic vzájemně nekompatibilní. Zákon o AI nepředepisuje, které metriky se mají používat, ale vyžaduje tyto systémy ve vysokém riziku jsou hodnoceny a dokumentovány ve srovnání s chráněnými skupinami. Dvě nejvíce metriky společné a doplňkové jsou Demografická parita (demografická rovnost) e aVyrovnané kurzy (rovnost šancí podmíněna skutečnými štítky).
La Demografická parita vyžaduje, aby pravděpodobnost získání pozitivního výsledku je stejný pro všechny skupiny: P(Y_pred=1 | A=0) = P(Y_pred=1 | A=1). A nejintuitivnější metrika ale může skrýt rozdíly ve skutečném výkonu, pokud mají skupiny různé sazby štítků. THE'Vyrovnané kurzy a přísnější: vyžaduje, aby byly TPR i FPR stejné napříč skupinami, což zajišťuje, že model dělá chyby se stejnou frekvencí bez ohledu na příslušnost k citlivé skupině.
# 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.")Demografická parita vs. Vyrovnané kurzy: Kterou použít?
- Demografická parita: ideální, když se neočekává žádný rozdíl v distribucích mezi skupinami (např. schválení úvěru: nesmí být žádný rozdíl v sazbách mezi muži a ženami bez ohledu na skutečnou solventnost). Může penalizovat celkovou přesnost.
- Vyrovnané kurzy: ideální, když se skutečné štítky mohou mezi skupinami lišit (např. lékařské vyšetření: různé míry onemocnění podle věku). Zaručuje, že falešně pozitivní a falešné zápory jsou distribuovány rovnoměrně.
- Varování (teorém nemožnosti): demografickou paritu a vyrovnání kurzů nikoli mohou být uspokojeny současně, pokud nejsou základní sazby totožné mezi skupinami. Vyberte metriku vhodnou pro váš kontext a zdokumentujte ji na kartě modelu.
Vysvětlitelnost pomocí SHAP: Místní a globální transparentnost
Vysvětlitelnost není jen regulačním požadavkem: je to praktický požadavek na modely ladění, získat důvěru zainteresovaných stran a odhalit latentní předsudky. SHAP (Shapley Additive exPlanations) se stal de facto standardem pro vysvětlitelnost v ML, protože nabízí vlastnosti solidní matematika: konzistence, místní přesnost a podpora pro stromové modely (s variantou TreeSHAP, O(TLD^2) místo O(TL2^M) obecné verze).
SHAP vypočítává mezní příspěvek každého prvku k predikci na základě teorie her kooperativní (Shapleyho hodnoty). Výsledkem je hodnota SHAP pro každý prvek každého vzorku: kladné hodnoty zvyšují předpověď, záporné hodnoty ji snižují. Součet hodnot SHAP všech vlastností a rovna rozdílu mezi modelovou predikcí a průměrnou predikcí (základní hodnota).
# 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}")Audit Trail: Neměnné protokoly pro shodu
Zákon o AI vyžaduje, aby vysoce rizikové systémy automaticky zaznamenávaly relevantní události během jejich provozu s protokoly dostatečně podrobnými, aby to úřady mohly ověřit shodu. Tyto protokoly musí být neměnné, s časovým razítkem a dohledatelné. Robustní systém auditních záznamů zahrnuje tři úrovně: Protokol predikcí jednotlivé protokoly, protokol přechodu fáze modelu a protokol událostí řízení.
# 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 resultsProtokol auditu: Uchovávání a ochrana
Zákon o AI nestanoví přesnou dobu uchovávání protokolů z vysoce rizikových systémů, ale pokyny naznačují, že musí být zachovány minimálně po celou dobu životnosti systému, obvykle 3-10 let. Ukládejte protokoly v neměnném úložišti (např. S3 s Object Lock, GCS s Retention Policy nebo databáze pouze pro připojení). Nikdy nepoužívejte databázi standardní relační jako jediné úložiště pro audit trail: řádky lze mazat popř upraveno. Hašovaná struktura SHA-256 ve výše uvedeném kódu umožňuje detekci manipulace a posteriori, ale nebrání fyzickému smazání souborů.
Modelujte správu registrů pomocí MLflow
Il Registr modelu MLflow a ústřední složkou vládnutí: centralizovat všechny modely ve výrobě s jejich životním cyklem, od experimentálních verzí až po modely ve výrobě, s kompletní historií přechodů, komentářů a značek řízení. S MLflow 3.0, zaveden v roce 2025, registr nativně podporuje modelové aliasy (produkce, inscenace, šampion, challenger) k nahrazení zastaralých fází a integruje se s Databricks' Unity Catalog for řízení napříč pracovními prostory.
# 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 reportZodpovědný rámec AI: Kontrolní seznam pro projekty ML
Operační rámec odpovědné umělé inteligence nemusí být teoretickým dokumentem: musí to být a konkrétní kontrolní seznam integrovaný do procesu vývoje. Následující rámec je založen na principech zákona o AI, pokyny NIST AI RMF (Risk Management Framework) a osvědčené postupy společnosti nejpokročilejších týmů ML. A navrženy tak, aby byly spustitelné pomocí nástrojů s otevřeným zdrojovým kódem za nulové náklady.
# 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
}Nejlepší postupy a anti-vzorce správy věcí veřejných
Poté, co si prohlédnete technické nástroje, je nezbytné pochopit, jak integrovat správu věcí veřejných každodenní život týmu ML bez vytváření neudržitelné režie. Musí existovat efektivní správa plně automatizované a integrované do potrubí CI/CD, nikoli ruční proces přidaný dodatečně.
ML Governance Best Practices
- Governance-as-Code: integruje všechny kontroly (karta modelu, poctivost, vysvětlitelnost) v potrubí CI/CD jako automatické kroky. Pokud kontrola spravedlivosti selže, k nasazení nedojde. Nulové ruční rozhodování pro technické brány, lidský dohled pouze pro organizační brány.
- Centralizace v registru modelů: Registr modelu MLflow je jediným zdrojem pravdy. Každá verze musí mít standardizované značky: risk_category, schválené_by, fairness_checked. Zakázat nasazení z jakékoli verze bez požadovaných značek.
- Dokumentujte rozhodnutí, nejen výsledky: také přihlásit, kdy se rozhodnete přijmout zbytkovou zaujatost s odůvodněním (např. „DP rozdíl 0,12 přijat protože překračuje práh pouze pro podskupinu X s n=45, není statisticky významný“).
- Šampion/vyzyvatel vždy: nikdy neodstraňujte předchozí verzi šablonu z registru, když propagujete novou. Nechte si to jako rollback vyzyvatele okamžitě v případě férovosti nebo problémů s výkonem ve výrobě.
- Pravidelná kontrola posunu spravedlnosti: zaujatost modelu se může změnit v průběhu času, pokud se změní rozložení vstupů. Každý měsíc opakujte analýzu spravedlnosti na produkčních datech, nejen při počátečním nasazení.
- Human-in-the-loop pro vysoké riziko: pro systémy AI Act High-Risk implementujte mechanismus označování, který posílá rozhodnutí s vysokým rizikem (např. pravděpodobnost mezi 0,45 a 0,55) lidskému operátorovi k ruční kontrole před jejich sdělením koncovému uživateli.
Anti-vzory, kterým je třeba se vyhnout
- Kontrola poctivosti pouze během tréninku: Spravedlnost ve výrobě se v čase mění. Model, který je při školení spravedlivý, se může stát nespravedlivým, pokud se výrobní data změní (datový drift). Průběžně sledujte férovost pomocí nástrojů Evidently nebo vlastních nástrojů.
- Statický model karty: karta modelu napsaná při prvním nasazení a již zastaralá při druhá aktualizace. Automatizujte generování s každým novým školením: to nemůže být a Dokument aplikace Word aktualizován ručně.
- SHAP na celý tréninkový set: TreeSHAP na milionech vzorků datových sad může trvat hodiny. Vždy použijte reprezentativní vzorek (500-2000 vzorků). globální pohledy. Pro místní vysvětlení ve výrobě počítejte SHAP pouze pro předpovědi, které vyžadují vysvětlení (Vysoce rizikové nebo blízko rozhodovací prahové hodnoty).
- Protokol auditu v transakční databázi: standardní SQL databáze umožňuje UPDATE a VYMAZAT. Pro protokoly auditu AI Act nepoužívejte upravitelné tabulky SQL: používejte soubory pouze pro připojení na neměnném úložišti s integritou hash.
- Schválení prostřednictvím e-mailu/chatu: schválení správy musí být sledována v systému (značky MLflow, lístky Jira atd.), nikoli ve vláknech Slack nebo e-mailech. V auditu regulační, "Marco napsal na Teams, že je to v pořádku" není dostatečná dokumentace.
Rozpočet pro malé a střední podniky: ML Governance s méně než 5 000 EUR/rok
ML governance nevyžaduje podnikové rozpočty. Celý rámec popsaný v tomto článku a 100% open-source:
- MLflow: zdarma, samostatně hostované na jediném serveru (2 vCPU, 4 GB RAM stačí pro malé týmy). Cena: ~15-20 EUR/měsíc v cloudu.
- Fairlearn + SHAP + scikit-learn: Open source knihovny Python s nulovými náklady udělování licencí.
- Protokoly auditu: Soubor JSONL na objektovém úložišti kompatibilním s S3 (vlastně hostované MinIO nebo cloud). 100 GB protokolů: přibližně 2-3 EUR/měsíc.
- Prometheus + Grafana pro sledování metrik spravedlivosti ve výrobě: zdarma, instalovatelné na K3s nebo Docker Compose.
Celkem: méně než 300 EUR/rok pro komplexní a vyhovující systém řízení podle požadavků zákona AI, s automatickými modelovými kartami, kontrolou poctivosti, vysvětlitelností SHAP a neměnná auditní stopa.
Závěry
ML governance v roce 2026 již není volitelná pro ty, kteří vyvíjejí systémy umělé inteligence v EU: Zákon o AI stanovil konkrétní požadavky s přísnými sankcemi. Ale když se podíváme za rámec souladu, rámec dobré správy přináší konkrétní výhody: robustnější modely díky spravedlnosti analýza, rychlejší ladění díky SHAP (o 31 % rychlejší podle průmyslových dat), větší důvěra zúčastněných stran a bezpečný návrat v případě problémů.
Přístup popsaný v tomto článku, zcela open-source a automatizovatelný v CI/CD, umožňuje týmům jakékoli velikosti implementovat správu ML bez neudržitelné režie. Klíč a automatizace: kódem generované modelové karty, kontroly poctivosti jako brány CI, audity stezka jako postranní vozík, SHAP počítá při každém nasazení. Vládnutí se stává součástí inženýrský proces, nikoli dokument, který je třeba vyplnit před auditem.
V dalším a posledním článku série, Případová studie Předpověď odchodu ve výrobě (ID 315), integrujeme všechny komponenty série: experimentální sledování MLflow, DVC pro verzování, Obsluha FastAPI, nasazení Kubernetes, monitorování pomocí Prometheus a rámec pro správu popsané v tomto článku v jediném fungujícím end-to-end potrubí.
Související články v této sérii MLOps
- MLOps: Od experimentu k produkci - Základy a kompletní životní cyklus
- ML kanál s CI/CD: GitHub Actions + Docker - Integrovat brány vládnutí do CI
- Sledování experimentů s MLflow - Modelový registr jako základ pro správu a řízení
- Detekce driftu modelu a automatické přeškolení - Fairness drift a monitorování
- Škálování ML na Kubernetes - Nasadit rámec řízení na K8
- A/B testování ML modelů - Řízení šampionů/vyzývatelů
Cross-Series
- Pokročilá řada hlubokého učení - Řízení pro hluboké učení a modely LLM
- Data & AI Business Series - AI governance v obchodním kontextu