ML 거버넌스: 프로덕션의 규정 준수, 감사 추적 및 책임 있는 AI
이탈 예측 모델은 92% 정확도에 도달하고 FastAPI 제공은 안정적입니다. Kubernetes 클러스터는 자동으로 확장됩니다. 그런 다음 법무 부서에서 "우리의 것"이라는 이메일이 도착합니다. 신용 모델은 AI법 EU의 고위험 범주에 속합니다. 그렇지 않다는 것을 증명해야 합니다. 성별이나 나이로 차별을 합니다. 지난 6개월 동안의 결정 로그는 어디에 있나요? 승인한 사람 프로덕션 배포?"
이러한 질문에 대한 즉각적인 답변이 없으면 ML 프로젝트가 위험에 처한 것입니다. 그만큼 인공 지능에 관한 EU 규정 (AI법) 일부 시행 2025년 8월에 고위험 시스템의 최종 기한이 2026년 8월 2일로 설정되었습니다. ML 거버넌스는 선택적 모범 사례에서 최대 6%의 벌금이 부과되는 법적 의무로 전환되었습니다. 연간 글로벌 매출액. MLOps 시장, 2026년 43억 8천만 달러 규모로 CAGR 39.8% 성장 또한 기업은 규정을 준수하기 위해 거버넌스 인프라를 갖추어야 하기 때문입니다.
이 백서에서는 포괄적인 오픈 소스 ML 거버넌스 프레임워크를 구축합니다. AI 법의 요구 사항, 모델 카드 자동 생성, MLflow를 통한 감사 추적, Fairlearn을 통한 공정성 감지, SHAP를 통한 설명 가능성까지. 모두 작동하는 Python 코드 포함 제한된 예산으로도 즉시 적용 가능합니다.
무엇을 배울 것인가
- 고위험 ML 시스템에 대한 EU AI Act 실무 요구 사항(마감일: 2026년 8월)
- 표준화된 문서로 자동 모델 카드 생성
- MLflow 및 구조화된 로그를 사용하여 포괄적인 감사 추적 구현
- Fairlearn(인구통계학적 동등성, 균등 확률)으로 편견을 측정하고 완화합니다.
- SHAP를 사용한 설명 가능성: 전역 기능 중요성 및 로컬 설명
- 모델 거버넌스 레지스트리: 추적된 승인을 통한 단계 전환
- AI 모델을 분류하기 위한 위험 평가 프레임워크
- 예산이 연간 5,000유로 미만인 중소기업을 위한 책임 있는 AI 체크리스트
EU AI법: ML 모델의 변경 사항
AI법은 AI 시스템을 4가지 위험 수준으로 분류하고 의무도 강화한다. 이전 거버넌스 프레임워크를 구현하려면 모델이 어디에 적합한지 이해해야 합니다. 분류. 분류는 사용된 기술에 의존하지 않습니다(XGBoost는 본질적으로 변압기보다 다소 위험하지만)의도된 용도 및 적용 도메인.
AI법의 4가지 위험 범주
허용할 수 없는 위험(금지): 사회적 점수 시스템, 잠재의식 조작,
취약점 악용. 배포가 불가능합니다.
고위험(엄격한 요구 사항): 학점, 채용, 교육 시스템,
중요 인프라, 의료 기기, 마이그레이션. 규정 준수 기한: 2026년 8월 2일.
제한된 위험(필수 투명성): 챗봇, 딥페이크, 추천 시스템.
AI와 상호작용하는 사용자에게 이를 알릴 의무.
최소 위험(자발적): 스팸 필터, AI 게임, 표준 권장 사항.
특별한 규제 의무는 없습니다.
시스템용 위험이 높은, AI법은 여러 기술적 의무를 부과하며 ML팀이 MLOps 파이프라인에서 구현해야 하는 조직적 단계입니다. 주요 요구사항은, 2026년 8월부터 시행되는 우려사항:
- 위험 관리 시스템: i를 식별, 분석 및 완화하기 위한 지속적인 프로세스 전체 모델 수명주기에 걸쳐 위험이 발생합니다.
- 데이터 거버넌스: 교육, 검증 및 테스트 데이터 세트를 문서화해야 합니다. 대표적이고 편견이 없으며 명시된 목적에 적합합니다.
- 기술 문서: 입증하기에 충분한 기술 문서 감독 당국의 준수. 아키텍처, 교육 절차, 측정항목 포함 성능 및 알려진 제한 사항.
- 자동 기록 보관: 시스템은 이벤트를 자동으로 기록해야 합니다. 감사를 위해 변경할 수 없는 로그를 사용하여 작업 중에 관련됩니다.
- 투명성 및 사용자 정보: 사용자는 자신이 누구인지 알아야 합니다. AI 시스템과 상호작용하고 그 기능과 한계에 대해 이해할 수 있는 정보를 받습니다.
- 인간의 감독: 인간의 감독, 개입을 가능하게 하는 기술적 메커니즘 그리고 자동화된 결정을 무시합니다.
- 정확성, 견고성, 사이버 보안: 문서화된 성능 지표, 분포 변화 및 적대적 입력에 대한 견고성 테스트.
AI법 제재: 이론적인 것이 아니다
AI법 위반에 대한 처벌은 심각합니다. 3천만 유로 또는 6% 연간 글로벌 매출액 허용할 수 없는 위험이 있는 시스템과 관련된 위반 ~까지 2천만 유로 또는 4% 기타 의무의 경우 ~까지 1천만 유로 또는 2% 당국에 잘못된 정보를 제공합니다. 고위험 시스템에 대한 첫 번째 중요한 기한과 2026년 8월 2일: 이 글이 작성된 날로부터 6개월 이내입니다. 귀하의 모델인 경우 신용, 고용 또는 중요 인프라에서 운영되는 경우 규정 준수 계획은 오늘부터 시작해야 합니다.
모델 카드: 표준화된 모델 문서
Le 모델 카드, 2019년에 Google에서 도입하여 현재 업계 표준으로 채택되었습니다. ML 모델(다양한 하위 그룹의 목적, 성능)을 설명하는 구조화된 문서입니다. 인구통계, 알려진 제한 사항, 의도 및 권장 사용. AI법은 암묵적으로 이를 요구합니다. "기술 문서" 섹션. 수동으로 생성하고 오류가 발생하기 쉽습니다. 다음 코드는 자동화합니다. 훈련 메타데이터 및 평가 결과로부터 생성.
# 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/")Fairlearn을 통한 공정성 및 편향 탐지
ML의 공정성은 단일 개념이 아닙니다. 수학적으로 여러 정의가 있습니다. 서로 호환되지 않습니다. AI법은 사용할 측정항목을 규정하지 않지만 해당 시스템을 요구합니다. 위험이 높은 그룹을 보호 그룹과 비교하여 평가하고 문서화합니다. 가장 많은 두 가지 지표 공통성과 보완성은 인구통계학적 동등성 (인구학적 평등) e 는균등 확률 (진정한 라벨을 조건으로 하는 확률의 동일성).
La 인구통계학적 동등성 긍정적인 결과를 얻을 확률이 필요합니다. 모든 그룹에 대해 동일합니다: P(Y_pred=1 | A=0) = P(Y_pred=1 | A=1). 그리고 가장 직관적인 측정항목 하지만 그룹의 라벨 비율이 서로 다른 경우 실제 실적의 차이를 숨길 수 있습니다. 그만큼'균등 확률 더욱 엄격함: TPR과 FPR이 모두 필요합니다. 그룹 전체에서 동일하므로 모델이 동일한 빈도로 오류를 발생시킵니다. 민감한 그룹에 속해 있는지 여부와 관계없이.
# 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.")인구통계학적 패리티와 균등 확률: 어느 것을 사용해야 할까요?
- 인구학적 동등성: 분포에 차이가 없을 것으로 예상되는 경우 이상적 그룹 간(예: 신용 승인: 남성과 여성 간 비율에 차이가 없어야 함) 실제 지불능력과 관계없이). 전반적인 정확성에 불이익을 줄 수 있습니다.
- 균등 확률: 실제 라벨이 그룹 간에 다를 수 있는 경우에 이상적입니다(예: 의료 검진: 연령에 따라 질병 발생률이 다름). 거짓 긍정과 거짓을 보장합니다. 네거티브는 균등하게 분배됩니다.
- 경고(불가능 정리): 인구통계학적 동등성과 균등 확률은 아님 기본 요율이 동일하지 않는 한 동시에 충족 가능 그룹 사이. 상황에 맞는 측정항목을 선택하고 이를 모델 카드에 기록하세요.
SHAP를 통한 설명 가능성: 로컬 및 글로벌 투명성
설명가능성은 단순한 규제 요구사항이 아닙니다. 모델 디버깅을 위한 실질적인 요구사항입니다. 이해관계자의 신뢰를 얻고 잠재적인 편견을 탐지합니다. SHAP (S하플리 첨가제 exPlanations)는 속성을 제공하므로 ML 설명 가능성에 대한 사실상의 표준이 되었습니다. 견고한 수학: 일관성, 로컬 정확도 및 트리 기반 모델 지원(변형 포함) TreeSHAP, 일반 버전의 O(TL2^M) 대신 O(TLD^2).
SHAP는 게임 이론을 기반으로 예측에 대한 각 기능의 한계 기여도를 계산합니다. 협동조합(Shapley 가치). 결과는 각 샘플의 각 기능에 대한 SHAP 값입니다. 양수 값은 예측을 증가시키고, 음수 값은 예측을 감소시킵니다. SHAP 값의 합 모든 특성의 모델 예측과 평균 예측의 차이와 같습니다. (기본값).
# 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}")감사 추적: 규정 준수를 위한 변경 불가능한 로그
AI법은 고위험 시스템이 관련 이벤트를 자동으로 기록하도록 요구합니다. 당국이 다음을 수행할 수 있을 만큼 상세한 로그를 보관합니다. 준수 여부를 확인합니다. 이러한 로그는 다음과 같아야 합니다. 변경할 수 없고 타임스탬프가 지정되어 있으며 추적 가능. 강력한 감사 추적 시스템에는 세 가지 수준이 포함됩니다. 예측 로그 개별 로그, 모델 단계 전환 로그 및 거버넌스 이벤트 로그.
# 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 results감사 로그: 보존 및 보호
AI법에는 고위험 시스템 로그에 대한 정확한 보관 기간이 명시되어 있지 않지만, 지침은 이를 유지해야 함을 나타냅니다. 적어도 전체 작동 수명 동안 시스템의, 일반적으로 3~10년입니다. 변경 불가능한 저장소(예: S3 객체 잠금, 보존 정책이 포함된 GCS 또는 추가 전용 데이터베이스). Never use a database 감사 추적을 위한 단일 저장소로 사용되는 표준 관계형: 행을 삭제하거나 수정되었습니다. 위 코드의 SHA-256 해시 구조는 변조 감지를 허용합니다. 사후적으로 가능하지만 파일의 물리적 삭제를 방지하지는 않습니다.
MLflow를 사용한 모델 레지스트리 거버넌스
Il MLflow 모델 레지스트리 거버넌스의 핵심 구성 요소: 중앙 집중화 실험 버전부터 모델까지 수명 주기에 따라 생산 중인 모든 모델 프로덕션에서는 전환, 댓글 및 거버넌스 태그의 전체 기록이 포함됩니다. MLflow 3.0을 사용하면, 2025년에 도입된 레지스트리는 기본적으로 모델 별칭(프로덕션, 스테이징, 챔피언, 도전자)는 더 이상 사용되지 않는 단계를 대체하고 Databricks의 Unity 카탈로그와 통합하여 작업공간 간 거버넌스.
# 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 report책임 있는 AI 프레임워크: ML 프로젝트용 체크리스트
책임 있는 AI 운영 프레임워크는 이론적인 문서일 필요는 없습니다. 구체적인 체크리스트가 개발 프로세스에 통합되었습니다. 다음 프레임워크는 원칙을 기반으로 합니다. AI법, NIST AI RMF(위험 관리 프레임워크) 지침 및 모범 사례 가장 발전된 ML 팀으로 구성된 회사입니다. 오픈 소스 도구로 실행 가능하도록 설계되었습니다. 비용이 전혀 들지 않습니다.
# 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
}ML 거버넌스 모범 사례 및 안티패턴
After seeing the technical tools, it is essential to understand how to integrate governance into 지속 불가능한 오버헤드를 생성하지 않고 ML 팀의 일상 생활을 수행합니다. 효과적인 거버넌스는 다음과 같습니다. 이후에 수동 프로세스가 추가되지 않고 완전히 자동화되어 CI/CD 파이프라인에 통합됩니다.
ML 거버넌스 모범 사례
- 코드형 거버넌스: 모든 검사 통합(모델 카드, 공정성, 설명 가능성) CI/CD 파이프라인에서 자동 단계로 공정성 검사에 실패하면 배포가 발생하지 않습니다. 기술 게이트에 대한 수동 결정이 없으며 조직 게이트에 대해서만 사람이 감독합니다.
- 모델 레지스트리에 중앙 집중화: MLflow Model Registry는 단일 정보 소스입니다. 각 버전에는 표준화된 태그(risk_category, Approved_by, fairness_checked)가 있어야 합니다. 필수 태그가 없는 모든 버전의 배포를 금지합니다.
- 결과뿐만 아니라 의사결정도 문서화하세요. 다음과 같은 경우에도 로그를 남깁니다. 정당화(예: "DP diff 0.12 허용됨)와 함께 잔여 편향을 허용하기로 결정했습니다. n=45인 부분군 X에 대해서만 임계값을 초과하고 통계적으로 유의하지 않기 때문입니다.")
- 챔피언/도전자는 항상: 이전 버전을 제거하지 마십시오. 새 템플릿을 승격할 때 레지스트리에서 템플릿을 가져옵니다. 롤백 도전자로 유지 생산 공정성이나 성능 문제가 있는 경우 즉시 조치합니다.
- 공정성 드리프트에 대한 정기적인 검토: 모델의 편향은 변경될 수 있습니다. 시간이 지남에 따라 입력의 분포가 변하는 경우. 매월 공정성 분석을 다시 실행하세요. 초기 배포뿐만 아니라 프로덕션 데이터에 대해서도 마찬가지입니다.
- 고위험군에 대한 인간 개입: AI법 고위험 시스템 구현 위험도가 높은 결정을 보내는 플래그 지정 메커니즘(예: 0.45에서 0.45 사이의 확률) 0.55) 최종 사용자에게 전달하기 전에 수동 검토를 위해 운영자에게 전달합니다.
피해야 할 안티패턴
- 훈련 중에만 공정성 검사: 생산의 공정성은 시간이 지나면서 변합니다. 학습 시 공정한 모델은 프로덕션 데이터가 변경되면(데이터 드리프트) 불공평해질 수 있습니다. Evidently 또는 맞춤형 도구를 사용하여 지속적으로 공정성을 모니터링하세요.
- 정적 모델 카드: 첫 번째 배포 시에 작성된 모델 카드는 이미 폐기되었습니다. 두 번째 업데이트. 각각의 새로운 훈련으로 생성을 자동화하세요. Word 문서가 수동으로 업데이트되었습니다.
- 전체 훈련 세트에 대한 SHAP: 백만 개의 샘플 데이터 세트에 대한 TreeSHAP 몇 시간이 걸릴 수 있습니다. 항상 대표 샘플(500-2000개 샘플)을 사용하십시오. 글로벌 견해. 프로덕션의 로컬 설명을 위해 SHAP만 계산합니다. 설명이 필요한 예측(고위험 또는 결정 임계값에 근접)
- 트랜잭션 데이터베이스의 감사 로그: 표준 SQL 데이터베이스는 UPDATE를 허용합니다. 그리고 삭제하세요. AI Act 감사 로그의 경우 편집 가능한 SQL 테이블을 사용하지 말고 추가 전용 파일을 사용하세요. 무결성 해시가 있는 불변 저장소에 있습니다.
- 이메일/채팅을 통한 승인: 거버넌스 승인을 추적해야 합니다. Slack 스레드나 이메일이 아닌 시스템(MLflow 태그, Jira 티켓 등)에 있습니다. 감사에서 규정에 따라 "Marco가 Teams에 괜찮다고 썼다"는 문서는 충분하지 않습니다.
중소기업을 위한 예산: 연간 5,000 EUR 미만의 ML 거버넌스
ML 거버넌스에는 기업 예산이 필요하지 않습니다. 이 문서에 설명된 전체 프레임워크 100% 오픈 소스:
- ML흐름: 무료, 단일 서버에서 자체 호스팅(vCPU 2개, 4GB RAM이면 충분) 소규모 팀의 경우). 비용: 클라우드 기준 ~15-20 EUR/월.
- Fairlearn + SHAP + scikit-learn: 오픈 소스, 무료 Python 라이브러리 라이센스.
- 감사 로그: S3 호환 객체 스토리지의 JSONL 파일(자체 호스팅 MinIO 또는 클라우드). 100GB 로그: 약 2-3 EUR/월.
- 프로메테우스 + 그라파나 프로덕션에서 공정성 측정항목을 모니터링하려면 다음을 수행하세요. 무료이며 K3s 또는 Docker Compose에 설치할 수 있습니다.
총액: 연간 300유로 미만 포괄적이고 규정을 준수하는 거버넌스 시스템을 위해 자동 모델 카드, 공정성 검사, SHAP 설명 가능성 및 AI법 요구 사항을 충족합니다. 변경할 수 없는 감사 추적.
결론
2026년의 ML 거버넌스는 인공지능 시스템을 개발하는 사람들에게 더 이상 선택 사항이 아닙니다. EU: AI법은 엄격한 제재와 함께 구체적인 요구 사항을 설정했습니다. 하지만 규정 준수를 넘어, 좋은 거버넌스 프레임워크는 구체적인 이점을 제공합니다. 공정성 덕분에 더욱 강력한 모델이 됩니다. 분석, SHAP 덕분에 더 빠른 디버깅(업계 데이터에 따르면 31% 더 빠름), 문제 발생 시 이해관계자의 신뢰와 안전한 롤백.
이 문서에 설명된 접근 방식은 완전히 오픈 소스이며 CI/CD에서 자동화 가능합니다. 모든 규모의 팀이 지속 불가능한 오버헤드 없이 ML 거버넌스를 구현할 수 있도록 지원합니다. 핵심 및 자동화: 코드 생성 모델 카드, CI 게이트로서의 공정성 검사, 감사 트레일을 서비스 사이드카로 사용하고 SHAP는 각 배포에서 계산됩니다. 거버넌스는 다음의 일부가 됩니다. 감사 전에 작성해야 하는 문서가 아닌 엔지니어링 프로세스입니다.
시리즈의 다음 기사이자 마지막 기사에서는 사례 연구 생산 중 이탈 예측 (ID 315), MLflow 실험 추적, 버전 관리용 DVC, 시리즈의 모든 구성 요소를 통합합니다. FastAPI 제공, Kubernetes 배포, Prometheus 및 거버넌스 프레임워크를 사용한 모니터링 이 문서에서는 작동하는 단일 엔드투엔드 파이프라인에서 설명합니다.
이 MLOps 시리즈의 관련 기사
- MLOps: 실험에서 프로덕션까지 - 기본 및 전체 수명주기
- CI/CD가 포함된 ML 파이프라인: GitHub Actions + Docker - 거버넌스 게이트를 CI에 통합
- MLflow를 사용한 실험 추적 - 거버넌스의 기반이 되는 모델 레지스트리
- 모델 드리프트 감지 및 자동 재훈련 - 공정성 드리프트 및 모니터링
- Kubernetes에서 ML 확장 - K8s에 거버넌스 프레임워크 배포
- ML 모델의 A/B 테스트 - 챔피언/챌린저 거버넌스
시리즈 간
- 고급 딥러닝 시리즈 - 딥 러닝 및 LLM 모델 거버넌스
- 데이터 및 AI 비즈니스 시리즈 - 비즈니스 맥락에서의 AI 거버넌스