Zarządzanie danymi i jakość danych dla niezawodnej sztucznej inteligencji
Il 72% projektów sztucznej inteligencji przedsiębiorstw kończy się niepowodzeniem, zanim trafią do produkcji. Nie z braku wyrafinowanych algorytmów, nie z powodu nieodpowiednich architektur, nie z powodu braku talentów w zespole. Zawodzą za dane o niewystarczającej jakości, za brak zarządzania, za rurociągi produkujące po cichu błędne wyniki, których nikt nie zauważy.
W kontekście, w którymUstawa o sztucznej inteligencji UE po raz pierwszy nakłada wymagania prawne dotyczące jakości danych szkoleniowych dla systemów wysokiego ryzyka i w które inwestują włoskie MŚP coraz częściej w projektach opartych na danych dzięki Przejście PNRR 5.0, zbuduj taki solidne podstawy zarządzania danymi i jakości danych nie są już wyborem: są warunkiem wstępnym konkuruj i przestrzegaj zasad.
W tym artykule omówiono struktury operacyjne, narzędzia open source i implementacje praktyki, które pozwalają budować niezawodne systemy AI: od zdefiniowania wymiarów jakości danych, po realizację automatycznych testów z Wielkie oczekiwania e db, aż do zarządzania linią danych OtwórzMetadane i zgodność z ustawą AI Act.
Czego dowiesz się w tym artykule
- 6 wymiarów DAMA jakości danych i jak je mierzyć w sposób zautomatyzowany
- Wdrożenie kontroli jakości danych za pomocą Great Hopements i dbt-oczekiwania
- Katalog danych i pochodzenie danych z OpenMetadata i Apache Atlas
- Ramy zarządzania danymi: role, procesy i struktura organizacyjna
- AI Act Wymogi UE (art. 10) dotyczące danych szkoleniowych: co zrobić przed sierpniem 2026 r
- Obserwowalność danych za pomocą Soda Core i Monte Carlo dla rurociągów produkcyjnych
- Wykrywanie błędów w zbiorach danych ML i strategie łagodzenia
- Studium przypadku: ramy zarządzania danymi dla włoskiego MŚP produkcyjnego
Seria hurtowni danych, sztucznej inteligencji i transformacji cyfrowej
| # | Przedmiot | Centrum |
|---|---|---|
| 1 | Ewolucja hurtowni danych | Od SQL Server do Data Lakehouse |
| 2 | Siatka danych i zdecentralizowana architektura | Własność domeny danych |
| 3 | ETL kontra nowoczesny ELT | dbt, Airbyte i Fivetran |
| 4 | Orkiestracja rurociągów | Przepływ powietrza, Dagster i Prefekt |
| 5 | Sztuczna inteligencja w produkcji | Konserwacja predykcyjna i cyfrowy bliźniak |
| 6 | AI w finansach | Wykrywanie oszustw i scoring kredytowy |
| 7 | Sztuczna inteligencja w handlu detalicznym | Prognozowanie popytu i rekomendacje |
| 8 | AI w służbie zdrowia | Diagnostyka i odkrywanie leków |
| 9 | AI w logistyce | Optymalizacja tras i automatyzacja magazynu |
| 10 | LLM w biznesie | RAG Enterprise i poręcze |
| 11 | Przedsiębiorstwo baz danych wektorowych | pgvector, Pinecone i Weaviate |
| 12 | MLOps dla biznesu | Modele AI w produkcji za pomocą MLflow |
| 13 | Jesteś tutaj - Zarządzanie danymi | Jakość danych dla niezawodnej sztucznej inteligencji |
| 14 | Plan działania oparty na danych dla MŚP | Jak przyjąć AI i DWH |
Problem jakości danych w erze AI
„Śmieci na wejściu, śmieci na zewnątrz” to zasada znana inżynierom danych od dziesięcioleci, ale w przypadku generatywnej sztucznej inteligencji i modele uczenia maszynowego w produkcji, stały się konsekwencje złej jakości danych wykładniczo poważniejsze. Model wykrywania oszustw wyszkolony na niezrównoważonych danych generuje podróbki pozytywy, które blokują legalne transakcje. System rekomendacyjny przeszkolony na danych za pomocą uprzedzenia demograficzne wzmacniają dyskryminację. Model prognozowania popytu, na którym się uczono niespójne dane generują nieprawidłowe zamówienia, które mają bezpośredni wpływ na przepływy pieniężne.
Koszt danych niskiej jakości – dane za 2025 r
| Wskaźnik | Wartość | Źródło |
|---|---|---|
| Średni koszt na firmę (dane niskiej jakości) | 12,9 miliona dolarów rocznie | Gartnera 2024 |
| Projekty AI, które kończą się niepowodzeniem ze względu na jakość danych | 72% | McKinsey’a 2025 |
| Czas analityka danych na czyszczenie danych | 45-60% | Wiele ankiet |
| Włoskie MŚP posiadające formalny program jakości danych | 18% | Obserwatorium AI Polimi 2025 |
| Zredukuj błędy uczenia maszynowego dzięki zarządzaniu danymi | do 35% | Instytut IBM 2025 |
Kluczowym rozróżnieniem, którego brakuje wielu organizacjom, jest różnica pomiędzy jakość danych do celów analitycznych e jakość danych dla sztucznej inteligencji. W raportowaniu tradycyjne, anomalne dane dają błędny numer na dashboardzie – ktoś to zauważa, poprawia go, idziemy dalej. W uczeniu maszynowym nieprawidłowe dane w zestawie szkoleniowym powodują zanieczyszczenie cichy model, który następnie przez wiele miesięcy wcześniej zachowywał się niewłaściwie w produkcji że problem został zidentyfikowany. Opóźnienie między problemem a jego manifestacją, np nieskończenie wyżej.
6 wymiarów DAMA jakości danych
Ramy DAMA-DMBOK (Data Management Body of Knowledge) definiuje 6 wymiarów podstaw jakości danych, które w 2025 r. pozostaną standardowym punktem odniesienia dla każdego program jakości danych przedsiębiorstwa:
Wymiary DAMA dla AI - Extended Framework
| Rozmiar | Definicja | Wskaźnik specyficzny dla AI | Próg krytyczny |
|---|---|---|---|
| Dokładność | Dane prawidłowo odzwierciedlają rzeczywistość | % poprawnych etykiet w zestawie treningowym | >99% dla klasyfikacji krytycznej |
| Kompletność | Wszystkie niezbędne dane są obecne | % wartości innych niż null dla funkcji krytycznych | >95% dla funkcji wejściowych |
| Konsystencja | Dane są spójne w różnych systemach | % zgodnych zapisów pomiędzy źródłami | > 98% dla funkcji współdzielonych |
| Aktualność | Dane są aktualizowane i dostępne w razie potrzeby | Średnie opóźnienie danych produkcyjnych w porównaniu ze szkoleniem | <24h dla modeli czasu rzeczywistego |
| Ważność | Dane przestrzegają zdefiniowanych formatów i ograniczeń | % naruszeń wzorców i zakresów | <0,1% naruszeń |
| Wyjątkowość | Żadnych niezamierzonych duplikatów | % zduplikowanych rekordów w zestawie treningowym | <0,5% duplikatów |
Praktyczne wdrożenie: jakość danych z dużymi oczekiwaniami
Wielkie oczekiwania i najpopularniejszy framework Pythona typu open source do testowania jakość danych. Podejście jest podobne do testów jednostkowych kodu: nazywane są „oczekiwaniami” (oczekiwania) dotyczące danych, są automatycznie wykonywane w potoku i generowana jest dokumentacja automatyczne. Integracja z Airflow, Prefect i dbt sprawia, że jest to naturalny składnik dowolny nowoczesny stos danych.
# data_quality_pipeline.py
# Framework completo di data quality con Great Expectations
import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest
import pandas as pd
import numpy as np
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class DataQualityFramework:
"""
Framework di data quality per pipeline ML enterprise.
Implementa le 6 dimensioni DAMA con Great Expectations.
"""
def __init__(self, datasource_name: str = "ml_training_data"):
self.context = gx.get_context()
self.datasource_name = datasource_name
self.validation_results = {}
def build_expectation_suite_for_ml(
self,
suite_name: str,
target_column: str,
feature_columns: list[str]
):
"""
Crea una suite di expectations per dataset ML.
Copre tutte le 6 dimensioni DAMA.
"""
suite = self.context.add_expectation_suite(
expectation_suite_name=suite_name
)
# === DIMENSIONE: COMPLETENESS ===
# Feature critiche devono avere meno del 5% di valori nulli
for col in feature_columns:
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_not_be_null",
kwargs={
"column": col,
"mostly": 0.95 # 95% non-null
}
)
)
# Target column deve essere 100% non-null
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_not_be_null",
kwargs={"column": target_column}
)
)
# === DIMENSIONE: VALIDITY ===
# Valori numerici in range atteso (es. eta 0-120)
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_be_between",
kwargs={
"column": "customer_age",
"min_value": 18,
"max_value": 120,
"mostly": 0.999
}
)
)
# === DIMENSIONE: UNIQUENESS ===
# Chiavi primarie devono essere uniche
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_be_unique",
kwargs={"column": "transaction_id"}
)
)
# === DIMENSIONE: CONSISTENCY ===
# Il target deve avere solo valori attesi (no label noise)
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_be_in_set",
kwargs={
"column": target_column,
"value_set": [0, 1], # Classificazione binaria
}
)
)
# === DIMENSIONE: ACCURACY - Distribuzione statistica ===
# La distribuzione non deve essere degenere
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_proportion_of_unique_values_to_be_between",
kwargs={
"column": "customer_id",
"min_value": 0.3, # Almeno 30% utenti unici
"max_value": 1.0
}
)
)
# Bilanciamento classi per training set (max 10:1 ratio)
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_most_common_value_to_be_in_set",
kwargs={
"column": target_column,
"value_set": [0, 1],
"ties_okay": True
}
)
)
# === DIMENSIONE: TIMELINESS ===
# Dati non più vecchi di 90 giorni
suite.add_expectation(
gx.core.ExpectationConfiguration(
expectation_type="expect_column_values_to_be_between",
kwargs={
"column": "event_timestamp",
"min_value": "2024-11-01", # Rolling window
"max_value": datetime.now().strftime("%Y-%m-%d"),
"parse_strings_as_datetimes": True,
"mostly": 0.99
}
)
)
self.context.save_expectation_suite(suite)
logger.info(f"Suite {suite_name} creata con {len(suite.expectations)} expectations")
return suite
def validate_dataset(
self,
df: pd.DataFrame,
suite_name: str,
run_name: str = None
) -> dict:
"""
Valida un DataFrame contro la suite definita.
Restituisce risultati strutturati per monitoring.
"""
run_name = run_name or f"run_{datetime.now().isoformat()}"
batch_request = RuntimeBatchRequest(
datasource_name=self.datasource_name,
data_connector_name="runtime_connector",
data_asset_name="training_data",
runtime_parameters={"batch_data": df},
batch_identifiers={"run_id": run_name}
)
checkpoint = self.context.add_or_update_checkpoint(
name="ml_data_checkpoint",
validations=[
{
"batch_request": batch_request,
"expectation_suite_name": suite_name
}
]
)
results = checkpoint.run(run_name=run_name)
# Estrai metriche aggregate
validation_result = results.list_validation_results()[0]
stats = validation_result.statistics
quality_report = {
"run_name": run_name,
"timestamp": datetime.now().isoformat(),
"success": results.success,
"evaluated_expectations": stats["evaluated_expectations"],
"successful_expectations": stats["successful_expectations"],
"unsuccessful_expectations": stats["unsuccessful_expectations"],
"success_rate": stats["success_percent"] / 100,
"failed_checks": [
{
"expectation": r.expectation_config.expectation_type,
"column": r.expectation_config.kwargs.get("column"),
"details": r.result
}
for r in validation_result.results
if not r.success
]
}
self.validation_results[run_name] = quality_report
return quality_report
# Utilizzo nella pipeline
def run_data_quality_checks(raw_data_path: str):
framework = DataQualityFramework()
# Definisci le expectations per il dataset
framework.build_expectation_suite_for_ml(
suite_name="ml_training_v1",
target_column="churn_label",
feature_columns=["customer_age", "tenure_months", "monthly_charges",
"total_charges", "contract_type", "payment_method"]
)
# Carica e valida i dati
df = pd.read_parquet(raw_data_path)
report = framework.validate_dataset(df, "ml_training_v1")
if not report["success"]:
failed = report["unsuccessful_expectations"]
logger.error(f"Data quality FALLITA: {failed} check falliti su {report['evaluated_expectations']}")
for check in report["failed_checks"]:
logger.error(f" - {check['expectation']} su colonna '{check['column']}'")
raise DataQualityError(f"Dataset non conforme agli standard di qualità")
logger.info(f"Data quality OK: {report['success_rate']:.1%} success rate")
return reportJakość danych w dbt: testowanie deklaratywne w procesie transformacji
Dla zespołów używających db jako warstwa transformacji, pakiet oczekiwania-dbt zapewnia te same możliwości co Great Hopeations bezpośrednio w modelach dbt, definiując testy w YAML blisko kodu SQL. To Podejście „jakość jako kod” gwarantuje, że każda transformacja jest automatycznie sprawdzana.
# models/schema.yml
# Test di data quality con dbt-expectations
version: 2
models:
- name: ml_features_customer
description: "Feature store per modello churn prediction"
columns:
- name: customer_id
description: "Identificativo univoco cliente"
tests:
- unique
- not_null
- dbt_expectations.expect_column_values_to_be_of_type:
column_type: VARCHAR
- name: customer_age
description: "Eta del cliente in anni"
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_between:
min_value: 18
max_value: 120
mostly: 0.999
- name: monthly_charges
description: "Addebiti mensili in EUR"
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_between:
min_value: 0
max_value: 10000
- dbt_expectations.expect_column_mean_to_be_between:
min_value: 50
max_value: 200
- name: churn_label
description: "Label target: 1=churned, 0=retained"
tests:
- not_null
- accepted_values:
values: [0, 1]
# Verifica che il dataset non sia troppo sbilanciato
- dbt_expectations.expect_column_proportion_of_unique_values_to_be_between:
min_value: 0.01 # Almeno 1% di churn (altrimenti dataset anomalo)
max_value: 0.5 # Max 50% di churn (dataset equilibrato)
- name: event_date
description: "Data dell'evento"
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_between:
min_value: "2023-01-01"
max_value: "{{ var('max_date', modules.datetime.date.today().isoformat()) }}"
parse_strings_as_datetimes: true
- name: ml_features_customer
tests:
# Test a livello di tabella: volumetria attesa
- dbt_expectations.expect_table_row_count_to_be_between:
min_value: 10000
max_value: 10000000
# Nessuna riga duplicata su chiave composta
- dbt_expectations.expect_compound_columns_to_be_unique:
column_list: ["customer_id", "snapshot_date"]Katalog danych i pochodzenie danych z OpenMetadata
La dany rodowód - możliwość prześledzenia ścieżki danych od źródła aż do ostatecznego modelu sztucznej inteligencji – stało się wymogiem zasadniczym z dwóch zbiegających się powodów: zgodność z AI Act EU (która wymaga udokumentowania pochodzenia danych treningowych) oraz praktyczna potrzeba debugowania, gdy model daje nieoczekiwane wyniki.
OtwórzMetadane oraz najbardziej zaawansowana platforma katalogów danych i linii rodowodowych o otwartym kodzie źródłowym w 2025 r., zbudowany przez byłych inżynierów Ubera i założycieli Apache Hadoop. Obsługuje rodowód na poziomie kolumn natywna integracja z dbt, Airflow, Spark i głównymi hurtowniami danych.
# openmetadata_lineage.py
# Registrazione automatica della data lineage per pipeline ML
from metadata.ingestion.api.workflow import Workflow
from metadata.generated.schema.entity.data.pipeline import Pipeline
from metadata.generated.schema.entity.data.table import Table
from metadata.generated.schema.type.entityLineage import (
ColumnLineage,
EntitiesEdge,
LineageDetails
)
from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
OpenMetadataJWTClientConfig
)
import json
from typing import Optional
class MLPipelineLineageTracker:
"""
Tracker automatico di data lineage per pipeline ML.
Registra ogni trasformazione in OpenMetadata.
"""
def __init__(self, server_url: str, jwt_token: str):
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
OpenMetadataConnection
)
server_config = OpenMetadataConnection(
hostPort=server_url,
authProvider="openmetadata",
securityConfig=OpenMetadataJWTClientConfig(jwtToken=jwt_token)
)
self.metadata = OpenMetadata(server_config)
def register_training_data_lineage(
self,
source_tables: list[str],
feature_store_table: str,
ml_model_name: str,
transformation_description: str
):
"""
Registra il percorso: raw_data -> feature_store -> ml_model
Essenziale per compliance AI Act (Art. 10 - Data Governance)
"""
# 1. Recupera entità sorgente
source_entities = []
for table_fqn in source_tables:
table = self.metadata.get_by_name(
entity=Table,
fqn=table_fqn
)
if table:
source_entities.append(table)
# 2. Recupera/crea feature store table
feature_table = self.metadata.get_by_name(
entity=Table,
fqn=feature_store_table
)
# 3. Costruisci lineage con dettagli di trasformazione
for source in source_entities:
lineage_request = AddLineageRequest(
edge=EntitiesEdge(
fromEntity={
"id": str(source.id.__root__),
"type": "table"
},
toEntity={
"id": str(feature_table.id.__root__),
"type": "table"
},
lineageDetails=LineageDetails(
description=transformation_description,
columnsLineage=[
ColumnLineage(
fromColumns=[f"{source.fullyQualifiedName.__root__}.customer_id"],
toColumn=f"{feature_store_table}.customer_id"
),
ColumnLineage(
fromColumns=[f"{source.fullyQualifiedName.__root__}.event_date"],
toColumn=f"{feature_store_table}.snapshot_date"
)
]
)
)
)
self.metadata.add_lineage(lineage_request)
# 4. Aggiungi tag di compliance (AI Act)
self._tag_for_ai_act_compliance(feature_store_table, ml_model_name)
def _tag_for_ai_act_compliance(
self,
table_fqn: str,
model_name: str
):
"""
Aggiunge tag AI Act per tracciabilita compliance.
Essenziale per sistemi high-risk (AI Act Art. 10).
"""
table = self.metadata.get_by_name(entity=Table, fqn=table_fqn)
# Tag strutturati per AI Act
compliance_tags = {
"ai_act_article_10": "training_data",
"ai_system": model_name,
"data_governance_reviewed": "true",
"bias_assessment_date": "2025-01-15",
"data_quality_score": "98.5"
}
# Aggiorna custom properties
self.metadata.patch_entity(
entity=Table,
source=table,
destination=table.copy(
update={"customMetrics": compliance_tags}
)
)
print(f"Compliance tags aggiunti per {table_fqn} -> modello {model_name}")
# Utilizzo pratico
def setup_ml_lineage():
tracker = MLPipelineLineageTracker(
server_url="http://openmetadata.internal:8585/api",
jwt_token="your-jwt-token"
)
tracker.register_training_data_lineage(
source_tables=[
"default.raw_crm.customers",
"default.raw_billing.transactions",
"default.raw_support.tickets"
],
feature_store_table="default.feature_store.ml_churn_features_v3",
ml_model_name="churn_prediction_xgboost_v2",
transformation_description="Feature engineering per churn prediction: "
"aggregazioni mensili, calcolo tenure, "
"encoding categoriche. Approvato data steward 2025-01-15."
)Ramy zarządzania danymi: struktura organizacyjna i procesy
Zarządzanie danymi to nie narzędzie programowe: to system ludzie, procesy i technologie co zapewnia, że dane są zarządzane jako aktywa strategiczne. Dla MŚP Włoskie firmy, które zbliżają się do AI, budują tę strukturę w sposób pragmatyczny i merytoryczny.
Kluczowe role zarządzania danymi
Struktura zarządzania dla MŚP (50–500 pracowników)
| Rola | Odpowiedzialność | Szacowany pełny etat | Profil |
|---|---|---|---|
| Dyrektor ds. danych (CDO) | Strategia dotycząca danych, sponsor wykonawczy, budżet, zgodność z ustawą o sztucznej inteligencji | 0,25 FTE (częściowe) | Menedżer na poziomie C lub wyższego szczebla |
| Stewardzy danych | Własność domeny danych, definicja standardu, akceptacja zmian | 1 na domenę danych | Biznes + techniczny |
| Inżynier danych | Rurociąg, jakość techniczna, narzędzia monitorowania | 1-2 etaty | Profil techniczny |
| Analityk ds. jakości danych | Definicja KPI jakości, audyty, raportowanie | 0,5 etatu | Analityczny/hybrydowy |
| IOD (Inspektor Ochrony Danych) | RODO, ustawa o sztucznej inteligencji, bezpieczeństwo danych, prywatność już w fazie projektowania | 0,25-0,5 etatu | Prawne/techniczne |
Polityka zarządzania danymi: 5 podstawowych procesów
An operational data governance program is based on 5 recurring processes that must be udokumentowane, przypisane i monitorowane:
- Klasyfikacja danych: Każdy zbiór danych sklasyfikowany według wrażliwości (publiczny, wewnętrzne, poufne, tajne) i według rodzaju (PII, dane finansowe, dane operacyjne, dane szkoleniowe ML). Ta klasyfikacja określa kontrolę dostępu, zasady przechowywania i wymagania o odpowiednich właściwościach.
- Umowa SLA dotycząca jakości danych: Dla każdego krytycznego zestawu danych formalna definicja SLA jakość: kompletność > 98%, opóźnienie < 2h, dokładność sprawdzana kwartalnie. Umowy SLA muszą być mierzalne i automatycznie monitorowane.
- Zarządzanie zmianami: Formalny proces zmian schematu danych krytycy. Żadnych zmian bez oceny wpływu, testów regresyjnych i zatwierdzenia zarządcy danych domeny.
- Zarządzanie dostępem do danych: Zasada najmniejszych uprawnień stosowana do danych. Dostęp do danych szkoleniowych ML ograniczony do zespołu ML z pełnym dziennikiem audytu. Recenzja kwartalnie dostępów.
- Reakcja na incydent: Procedura eskalacji w przypadku incydentów związanych z jakością danych: automatyczne wykrywanie (alarmowanie), segregacja (<4h), remediacja, sekcja zwłok aktualizacja automatycznych kontroli.
Ustawa AI UE i wymagania dotyczące danych szkoleniowych
L'Ustawa o sztucznej inteligencji UE po raz pierwszy w historii wprowadza wymogi prawne ograniczenia dotyczące jakości danych szkoleniowych na potrzeby systemów sztucznej inteligencji wysokiego ryzyka. THE'Artykuł 10 rozporządzenia i w szczególności poświęcony „Danym i Danym”. Zarządzanie” i od tego czasu zaczął działać w modelach GPAI 2 sierpnia 2025 r. W przypadku systemów wysokiego ryzyka wymagana jest pełna zgodność 2 sierpnia 2026 r.
Harmonogram UE dotyczący ustawy o sztucznej inteligencji – co należy zrobić przed sierpniem 2026 r
| Data | Kamienie milowe | Wymagane działanie |
|---|---|---|
| luty 2025 | Operacyjne zabronione praktyki AI | Audytuj systemy AI pod kątem zabronionych praktyk |
| sierpień 2025 | Modele operacyjne i zarządzanie GPAI | Zarządzanie modelami LLM i podstawowymi |
| sierpień 2026 | Pełna zgodność systemów AI wysokiego ryzyka | Zarządzanie danymi 10 w pełni wdrożonych |
| sierpień 2027 | Zgodność ze starszymi systemami | Odpowiednie istniejące systemy sztucznej inteligencji |
Artykuł 10 Ustawy o AI: Wymagania techniczne dotyczące danych
Art. 10 ustawy o AI nakłada szczególne wymagania na zbiory danych wykorzystywane w szkoleniach, walidacja i testowanie systemów sztucznej inteligencji wysokiego ryzyka. Oto, co należy udokumentować i zaimplementowane konkretnie:
# ai_act_compliance_checker.py
# Verifica compliance AI Act Art. 10 per dataset di training
from dataclasses import dataclass, field
from typing import Optional
import pandas as pd
import numpy as np
from scipy import stats
import json
from datetime import datetime
@dataclass
class DatasetComplianceReport:
"""Report di compliance AI Act Articolo 10."""
dataset_name: str
assessment_date: str
assessor: str
compliant: bool = False
checks: dict = field(default_factory=dict)
recommendations: list = field(default_factory=list)
class AIActArticle10Checker:
"""
Verifica conformità AI Act Art. 10 per dataset di training high-risk.
Art. 10 richiede che i dataset siano:
1. Rilevanti e sufficientemente rappresentativi
2. Liberi da errori per quanto possibile
3. Completi rispetto allo scopo
4. Con proprietà statistiche appropriate
5. Privi di bias che potrebbero discriminare gruppi protetti
"""
def __init__(self, dataset_name: str, assessor: str):
self.report = DatasetComplianceReport(
dataset_name=dataset_name,
assessment_date=datetime.now().isoformat(),
assessor=assessor
)
def check_representativeness(
self,
df: pd.DataFrame,
demographic_columns: list[str],
reference_distributions: dict
) -> bool:
"""
Art. 10(3): Verifica rappresentativita demografica.
Confronta distribuzione del dataset con popolazione di riferimento.
"""
is_representative = True
self.report.checks["representativeness"] = {}
for col in demographic_columns:
if col not in df.columns:
continue
# Distribuzione nel dataset
observed = df[col].value_counts(normalize=True).to_dict()
# Distribuzione attesa (es. ISTAT per demografia italiana)
expected = reference_distributions.get(col, {})
if not expected:
continue
# Test chi-quadro per significativita statistica
categories = list(set(observed.keys()) | set(expected.keys()))
obs_counts = [df[col].value_counts().get(c, 0) for c in categories]
exp_props = [expected.get(c, 0.001) for c in categories]
# Normalizza expected
total_exp = sum(exp_props)
exp_normalized = [p / total_exp * len(df) for p in exp_props]
chi2, p_value = stats.chisquare(obs_counts, exp_normalized)
deviation = {
"chi2_statistic": chi2,
"p_value": p_value,
"is_representative": p_value > 0.05, # p > 0.05 = no significant deviation
"observed_distribution": observed,
"expected_distribution": expected
}
self.report.checks["representativeness"][col] = deviation
if p_value <= 0.05:
is_representative = False
self.report.recommendations.append(
f"CRITICO: Colonna '{col}' non e rappresentativa della popolazione target "
f"(chi2={chi2:.2f}, p={p_value:.4f}). "
f"Richiesta oversampling o raccolta dati aggiuntiva."
)
return is_representative
def check_bias_protected_attributes(
self,
df: pd.DataFrame,
target_column: str,
protected_attributes: list[str]
) -> bool:
"""
Art. 10(5): Verifica assenza bias su attributi protetti.
Usa disparate impact ratio (soglia 0.8 = 80% rule).
"""
is_unbiased = True
self.report.checks["bias_assessment"] = {}
for attr in protected_attributes:
if attr not in df.columns:
continue
groups = df[attr].unique()
positive_rates = {}
for group in groups:
mask = df[attr] == group
group_df = df[mask]
if len(group_df) == 0:
continue
positive_rate = (group_df[target_column] == 1).mean()
positive_rates[str(group)] = positive_rate
if len(positive_rates) < 2:
continue
max_rate = max(positive_rates.values())
min_rate = min(positive_rates.values())
# Disparate Impact = min_rate / max_rate
# AI Act richiede > 0.8 (80% rule)
disparate_impact = min_rate / max_rate if max_rate > 0 else 1.0
bias_result = {
"disparate_impact": disparate_impact,
"compliant": disparate_impact >= 0.8,
"group_rates": positive_rates,
"max_group": max(positive_rates, key=positive_rates.get),
"min_group": min(positive_rates, key=positive_rates.get)
}
self.report.checks["bias_assessment"][attr] = bias_result
if disparate_impact < 0.8:
is_unbiased = False
self.report.recommendations.append(
f"CRITICO: Bias rilevato su attributo protetto '{attr}'. "
f"Disparate impact = {disparate_impact:.3f} (soglia AI Act: 0.80). "
f"Applicare re-weighting, resampling o fairness constraints."
)
return is_unbiased
def check_completeness_and_errors(
self,
df: pd.DataFrame,
critical_features: list[str]
) -> bool:
"""
Art. 10(2): Verifica completezza e assenza di errori.
"""
is_complete = True
self.report.checks["completeness"] = {}
for col in critical_features:
null_rate = df[col].isna().mean()
is_col_complete = null_rate < 0.05 # Soglia 5%
self.report.checks["completeness"][col] = {
"null_rate": null_rate,
"compliant": is_col_complete
}
if not is_col_complete:
is_complete = False
self.report.recommendations.append(
f"AVVISO: Feature '{col}' ha {null_rate:.1%} valori mancanti. "
f"Applicare imputation strategy o raccogliere dati aggiuntivi."
)
return is_complete
def generate_compliance_report(self) -> str:
"""Genera report JSON per audit AI Act."""
self.report.compliant = all([
check.get("is_representative", True)
for check in self.report.checks.get("representativeness", {}).values()
] + [
check.get("compliant", True)
for check in self.report.checks.get("bias_assessment", {}).values()
] + [
check.get("compliant", True)
for check in self.report.checks.get("completeness", {}).values()
])
return json.dumps({
"dataset": self.report.dataset_name,
"assessment_date": self.report.assessment_date,
"assessor": self.report.assessor,
"ai_act_article_10_compliant": self.report.compliant,
"checks": self.report.checks,
"recommendations": self.report.recommendations,
"next_review_date": "2025-04-15"
}, indent=2, default=str)
# Esempio di utilizzo per un sistema HR high-risk
def assess_hr_model_dataset(df: pd.DataFrame):
checker = AIActArticle10Checker(
dataset_name="hr_promotion_training_v3",
assessor="Maria Rossi, Data Steward HR"
)
# Distribuzione di riferimento: composizione forza lavoro italiana
reference_distributions = {
"gender": {"M": 0.54, "F": 0.46},
"age_group": {"18-30": 0.20, "31-45": 0.40, "46-60": 0.35, "60+": 0.05}
}
# Esegui tutti i check
checker.check_representativeness(df, ["gender", "age_group"], reference_distributions)
checker.check_bias_protected_attributes(df, "promoted", ["gender", "age_group", "nationality"])
checker.check_completeness_and_errors(df, ["performance_score", "tenure", "education_level"])
report = checker.generate_compliance_report()
print(report)
return reportObserwowalność danych: ciągłe monitorowanie rurociągów produkcyjnych
Jakość danych nie jest gwarantowana tylko w momencie ich przetworzenia: dane z czasem ulegają degradacji czas. Zjawisko dryf danych - gdzie dystrybucja danych w produkcja stopniowo odbiega od zbioru treningowego – i jest to jedna z głównych przyczyn cichej degradacji modeli sztucznej inteligencji. Obserwowalność danych rozwiązuje ten problem z ciągłym monitorowaniem i proaktywnym ostrzeganiem.
Porównanie narzędzi obserwacji danych 2025
| Instrument | Typ | Mocne strony | Idealny przypadek użycia |
|---|---|---|---|
| Rdzeń sodowy | Otwarte źródło | Oparty na YAML, CLI, osadzany w CI/CD | MŚP z ograniczonym budżetem, rurociąg dbt |
| Monte Carlo | SaaS (korporacja) | Wykrywanie anomalii przy zerowej konfiguracji oparte na technologii ML | Przedsiębiorstwo, duży wolumen, mały zespół |
| Wielkie oczekiwania | Otwarte źródło | Natywna dla języka Python, elastyczna, automatyczna dokumentacja | Zespół inżynierii danych, potok Pythona |
| testy dbt | Otwarte źródło | Jakość jako kod, zintegrowana z przepływem pracy dbt | Zespoły używające dbt jako warstwy głównej |
| Metaplan | SaaS (średni rynek) | Pochodzenie + obserwowalność, intuicyjny interfejs użytkownika | Skalowanie z wieloma źródłami danych |
Wykrywanie dryfu danych za pomocą rdzenia sodowego
Rdzeń sodowy pozwala zdefiniować kontrole jakości w YAML i je wykonać zgodnie z harmonogramem lub na żądanie. W przypadku monitorowania dryfu danych możliwe jest porównanie automatyczne wykrywanie rozkładów cech krytycznych pomiędzy różnymi okresami znaczące zmiany, które mogą pogorszyć modele sztucznej inteligencji.
# soda_checks_ml_pipeline.yml
# Check Soda Core per data observability pipeline ML
checks for ml_features_customer:
# === FRESHNESS: timeliness dei dati ===
- freshness(event_date) < 24h:
name: "Dati non più vecchi di 24 ore"
fail:
when freshness(event_date) > 24h
warn:
when freshness(event_date) > 12h
# === VOLUME: anomalie di volumetria ===
- row_count > 50000:
name: "Minimo 50K record per inference affidabile"
- row_count between 50000 and 5000000:
name: "Volume nel range atteso (no spike anomali)"
# === COMPLETENESS ===
- missing_count(customer_age) = 0:
name: "Customer age: zero null tollerati"
- missing_percent(monthly_charges) < 2%:
name: "Monthly charges: max 2% null"
# === VALIDITY ===
- invalid_percent(customer_age) < 0.1%:
name: "Customer age nel range 18-120"
valid min: 18
valid max: 120
- duplicate_count(customer_id) = 0:
name: "Nessun customer_id duplicato"
# === DISTRIBUTION DRIFT ===
# Confronto con baseline storica (riferimento: settimana precedente)
- distribution_difference_index(monthly_charges) < 0.1:
name: "Monthly charges: drift < 10% vs baseline"
method: psi # Population Stability Index
baseline: soda://ml_features_customer/monthly_charges_baseline
- distribution_difference_index(customer_age) < 0.1:
name: "Customer age: drift < 10% vs baseline"
method: ks # Kolmogorov-Smirnov test
# === BUSINESS RULES ===
- failed_rows(monthly_charges_anomaly):
name: "Nessun addebito negativo"
fail condition: monthly_charges < 0
- failed_rows(contract_consistency):
name: "Contratti annuali: addebito >= 100 EUR"
fail condition: contract_type = 'annual' AND monthly_charges < 100
# Configurazione alerting
alert config:
slack:
webhook: "https://hooks.slack.com/services/..."
channel: "#data-quality-alerts"
email:
to:
- "data-team@azienda.it"
- "ml-team@azienda.it"
severity:
fail: immediate
warn: daily_digestWykrywanie błędów w zbiorach danych ML: podejście systematyczne
Il stronniczość w danych i jeden z najtrudniejszych problemów uczenia maszynowego w przedsiębiorstwie: ciche, trudne do wykrycia bez odpowiednich metod i potencjalnie dyskryminujące wobec grup chronionych. Ustawa o sztucznej inteligencji klasyfikuje systemy sztucznej inteligencji stosowane jako „wysokiego ryzyka”. decyzje dotyczące dostępu do pracy, kredytów, usług publicznych i edukacji, wykrywanie uprzedzeń i stało się wymogiem zgodności i etyki zawodowej.
Błędy w danych szkoleniowych wynikają z trzech głównych źródeł: historyczne nastawienie (dane odzwierciedlające dyskryminację w przeszłości, np. historycznie niezrównoważone stawki awansów), stronniczość reprezentacji (grupy niedostatecznie reprezentowane w zbiorze uczącym), tj błąd pomiaru (funkcje, które służą jako proxy dla chronionych atrybutów, np. kod pocztowy jako wskaźnik dochodu i pochodzenia etnicznego).
Metryki uczciwości – standardy IBM AIF360 i AI Act
| Metryczny | Formuła | Próg ustawy AI | Typowe zastosowanie |
|---|---|---|---|
| Zróżnicowany wpływ | P(Y=1|A=0) / P(Y=1|A=1) | >= 0,80 | Zatrudnianie, kredyt |
| Równe szanse | TPR(A=0) - TPR(A=1) | |różnica| < 0,05 | Opieka zdrowotna, sprawiedliwość |
| Parytet demograficzny | P(Y=1|A=0) - P(Y=1|A=1) | |różnica| < 0,05 | Zalecenia |
| Kalibrowanie | P(Y=1|wynik=s, A=a) jednolite | RMSE < 0,02 | Punktacja ryzyka |
Studium przypadku: Zarządzanie danymi dla włoskiego MŚP produkcyjnego
MŚP produkcyjne z Lombardii zatrudniające 180 pracowników rozpoczęło program daty w 2024 r zarządzanie w celu wspierania trzech inicjatyw związanych ze sztuczną inteligencją: model konserwacji predykcyjnej maszyn CNC, wizualny system kontroli jakości oparty na wizji komputerowej i prognozowaniu popytu do planowania produkcji.
Wyzwanie: Dane były wszędzie i nikt nie był za nie odpowiedzialny
Początkowa sytuacja była typowa dla włoskich MŚP: czujniki IoT, do których zapisywały dane autorskie formaty, dane produkcyjne w SAP Business One ERP, dane jakościowe w Arkusze Excel udostępniane mailem, dane sprzedażowe w niezintegrowanym CRM. Zespół IT miał techniczną wizję infrastruktury, ale bez formalnego zarządzania.
Wyniki programu zarządzania danymi (12 miesięcy)
| Wskaźnik | Zanim | Po 12 miesiącach | Uderzenie |
|---|---|---|---|
| Średni wynik jakości danych | 61% | 94% | +33 punkty |
| Czas przygotowania danych ML | 3-4 tygodnie | 2-3 dni | -90% |
| Zbiór danych z udokumentowanym pochodzeniem | 0% | 87% | +87 punktów |
| Modele AI w produkcji | 0 | 3 | +3 systemy |
| Zgodność z ustawą o AI art. 10 | Nie ocenione | Uległy | Gotowy do audytów |
Plan wdrożenia w 4 fazach
Program składał się z 4 kwartalnych etapów, a łączna wartość inwestycji wyniosła ok 85 000 EUR (częściowo objęte PNRR Transition 5.0):
- Faza 1 – Inwentaryzacja i ocena (I kwartał 2024 r.): Mapowanie wszystkich systemów danych, klasyfikacja zbiorów danych według krytyczności AI, wstępna ocena jakości. Narzędzia: arkusz inwentarzowy + kwestionariusz samooceny dla menadżerów domen. Wynik: kompletna mapa krytycznych danych z analizą luk.
- Faza 2 - Fundacja (2 kwartał 2024): Wdrożenie OpenMetadata lokalnie, konfiguracja pierwszych katalogów danych, określenie stewardów danych dla każdej domeny (produkcja, jakość, sprzedaż). Implementation of the first automatic checks with dbt tests.
- Faza 3 – Automatyzacja Jakości (3 kwartał 2024): Wdrażaj wielkie oczekiwania do automatycznej walidacji zbiorów danych ML, integracja z Airflow w celu alertowania, zdefiniowanie jakościowych umów SLA dla 3 projektów AI. Pierwsza ocena Ustawa o AI art. 10.
- Faza 4 – Optymalizacja i dojrzałe zarządzanie (IV kwartał 2024 r.): Rada Danych aktywne co miesiąc, wskaźniki jakości KPI na pulpicie nawigacyjnym Metabase, proces zarządzania zmianami formalne, szkolenie personelu. 3 modele AI zostały wdrożone w produkcji z pełnym zarządzaniem danymi.
Wyciągnięte wnioski: 5 błędów, których należy unikać
- Zaczynając od narzędzi zamiast procesów: Technologia przychodzi później ludzie i procesy. Wyznacz stewardów danych przed instalacją jakiegokolwiek oprogramowania.
- Chcąc zrobić wszystko od razu: Ewolucyjny program zarządzania i nie tylko trwałe niż wielki wybuch. Zacznij od krytycznych zbiorów danych dla AI i stopniowo je rozwijaj.
- Niedocenianie treningu: Zainwestowano 40% budżetu szkolenia i zarządzanie zmianą. Bez tego narzędzia są ignorowane.
- Nie angażuj firmy: Stewardzy danych muszą być postaciami hybrydowymi biznesowo-techniczne, nie tylko IT. Biznes wie, które dane są najważniejsze, IT wie jak zmierzyć je.
- Ignorowanie zgodności od początku: Wymagania ustawy AI nie są ograniczenie: stanowią one strukturę uzasadniającą w oczach inwestycję w zarządzanie zarządzania.
Najlepsze praktyki i anty-wzorce
Podstawowe najlepsze praktyki
- Jakość jako kod: Zdefiniuj wymagania jakościowe w kodzie (YAML, Python) wersjonowane w Git, a nie w dokumentach Worda. Dzięki temu kontrole są automatyczne i powtarzalne i część procesu CI/CD.
- Porażka szybka, porażka głośna: Kontrole jakości danych muszą blokować potok (a nie tylko generować ostrzeżenie), gdy dane są poza zakresem. Modelka trenowana słabe dane powodują większe szkody niż zatrzymany rurociąg.
- Oddziel walidację od transformacji: Sprawdź poprawność danych przed ich przekształceniem (walidacja schematu i zakresu przy przyjęciu), w trakcie transformacji (test dbt w modelach), i przed użyciem w ML (Wielkie oczekiwania w sklepie z funkcjami). Jakość pojawia się w wielu momentach, a nie tylko na końcu.
- Monitoruj dryf, a nie tylko jakość statyczną: Dane zmieniają się w czasie. Testy PSI (wskaźnik stabilności populacji) i KS (Kołmogorow-Smirnow) to standardowe narzędzia w celu wykrycia dryfów dystrybucji, które po cichu degradują modele.
- Decyzje dotyczące zarządzania dokumentami: Each choice (quality threshold, strategia atrybucji, wyłączenie zbioru danych) muszą być udokumentowane datą, autorski i racjonalny. Dokumentacja ta jest wymagana przez ustawę AI dla systemów wysokiego ryzyka.
Anty-wzorce, których należy unikać
Krytyczne antywzorce w zarządzaniu danymi dla sztucznej inteligencji
- Jakość danych „najlepiej”: Zdefiniuj niejasne umowy SLA („jak najwięcej danych ukończone”) zamiast precyzyjnych i mierzalnych progów liczbowych. Bez wskaźników nie ma żadnego prawdziwe rządzenie.
- Wyciszanie alertów: Skonfiguruj kontrole jakości, a następnie wycisz alerts because they "disturb". Każdy zignorowany alert jest problemem przyszłego modelu w produkcji.
- Zarządzanie wyłącznie nowymi projektami: Starsze zestawy danych używane do przekwalifikowania modeli mają takie same wymagania jakościowe. Często jest ich najwięcej problematyczne.
- Jednorazowa kontrola odchylenia: Sprawdź nastawienie tylko przed treningiem początkowy. Z biegiem czasu w przypadku nowych danych produkcyjnych mogą pojawić się błędy systematyczne (dryf koncepcji).
- Myląca jakość danych dla analityki i uczenia maszynowego: Dopuszczalne progi dla pulpitu nawigacyjnego (5% wartości null w kolumnie) może mieć katastrofalne skutki dla danej funkcji wejść ML. Te dwa konteksty wymagają różnych standardów.
Wnioski i dalsze kroki
Zarządzanie danymi i jakość danych w przypadku sztucznej inteligencji to nie biurokracja: to infrastruktura invisible, który określa, czy Twoje modele działają w produkcji, czy po cichu zawodzą. Dzięki unijnej ustawie o sztucznej inteligencji wprowadzającej prawnie wiążące wymogi dotyczące danych szkoleniowych, Inwestowanie w zarządzanie to nie tylko dobra praktyka – to warunek wstępny funkcjonowania na rynku europejskim z systemami AI wysokiego ryzyka.
Il praktyczny punkt wyjścia dla dowolnej organizacji - MŚP lub przedsiębiorstwa - i to samo: zidentyfikuj 3-5 najbardziej krytycznych zbiorów danych dla Twoich inicjatyw AI, nazwij a administrator danych dla każdego z nich, wdrażaj automatyczne kontrole za pomocą narzędzi open source opisane w tym artykule (Wielkie oczekiwania, dbt-oczekiwania, Soda Core) oraz stopniowo buduj strukturę zarządzania wokół tych zbiorów danych.
Perfekcja nie jest wymogiem początkowym: 94% wyniku jakości danych osiągnięty przez PMI dla przemysłu w naszym studium przypadku po 12 miesiącach zaczynał od 61%. Podróż Liczy się tak samo jak miejsce docelowe i każdy punkt procentowy poprawy jakości danych przekłada się bezpośrednio na bardziej niezawodne modele sztucznej inteligencji i mniej wypadków produkcji i trafniejsze decyzje biznesowe.
Zarządzanie danymi na potrzeby listy kontrolnej uruchamiania programu AI
- Zakończono inwentaryzację krytycznych zbiorów danych dla sztucznej inteligencji
- Wyznaczeni stewardowie danych dla każdej kluczowej domeny danych
- Zdefiniowana i zatwierdzona umowa SLA dotycząca jakości (kompletność, świeżość, ważność)
- Automatyczne kontrole wdrożone w rurociągu (GE, dbt-oczekiwania lub Soda)
- Katalog danych ze skonfigurowanym pochodzeniem (OpenMetadata lub Apache Atlas)
- Ocena błędu systematycznego przeprowadzona dla wszystkich zbiorów danych wykorzystywanych w systemach wysokiego ryzyka
- Art. ustawy o AI. Rozpoczęto 10 dokumentacji dla systemów wysokiego ryzyka
- Skonfigurowane alerty i zdefiniowane kanały eskalacji
- Planowana pierwsza miesięczna Rada Danych
- Szkolenie drużyny zakończone
Powiązane spostrzeżenia
- MLOps dla biznesu: Jak monitorować dryf modeli w produkcji z MLflow - Artykuł nr 12 z tej serii
- LLM w firmie: Zarządzanie danymi dla przedsiębiorstwa i bezpieczeństwa RAG danych w modelach wielkojęzykowych - Artykuł nr 10 z tej serii
- Inżynieria sztucznej inteligencji: Magazyny funkcji i osadzanie zarządzania systemami Przedsiębiorstwo RAG - Seria inżynierii AI
- Sztuczna inteligencja PostgreSQL: pgvector i jakość danych dla wektorowych baz danych - Seria PostgreSQL AI