안정적인 AI를 위한 데이터 거버넌스 및 데이터 품질
Il 엔터프라이즈 AI 프로젝트의 72%가 프로덕션에 도달하기 전에 실패합니다.. 부족해서가 아니다 부적절한 아키텍처나 팀에서 누락된 인재 때문이 아니라 정교한 알고리즘을 사용하는 것입니다. 그들은 실패한다 품질이 부족한 데이터, 거버넌스 부족, 자동으로 생산되는 파이프라인 아무도 눈치채지 못하는 잘못된 결과.
다음과 같은 맥락에서AI법 EU 최초로 품질에 대한 법적 요구사항을 부과합니다. 이탈리아 중소기업이 투자하고 있는 고위험 시스템에 대한 교육 데이터 덕분에 데이터 기반 프로젝트가 점점 더 늘어나고 있습니다. PNRR 전환 5.0, 하나 만들어 보세요 데이터 거버넌스 및 데이터 품질의 견고한 기반은 더 이상 선택이 아닙니다. 경쟁하고 규정을 준수하세요.
이 문서에서는 운영 프레임워크, 오픈 소스 도구 및 구현을 안내합니다. 신뢰할 수 있는 AI 시스템을 구축할 수 있는 관행: 품질 차원의 정의부터 데이터의 자동 테스트 구현에 이르기까지 큰 기대 e DBT, 데이터 계보 관리까지 오픈메타데이터 그리고 AI법을 준수합니다.
이 기사에서 배울 내용
- 데이터 품질의 6가지 DAMA 차원 및 이를 자동화된 방식으로 측정하는 방법
- Great Expectations 및 dbt-expectations를 통한 데이터 품질 검사 구현
- OpenMetadata 및 Apache Atlas를 사용한 데이터 카탈로그 및 데이터 계보
- 데이터 거버넌스 프레임워크: 역할, 프로세스 및 조직 구조
- 학습 데이터에 대한 AI Act EU 요구 사항(10조): 2026년 8월 이전에 수행해야 할 작업
- 생산 파이프라인을 위한 Soda Core 및 Monte Carlo를 사용한 데이터 관측성
- ML 데이터세트의 편향 감지 및 완화 전략
- 사례 연구: 이탈리아 제조 SME를 위한 데이터 거버넌스 프레임워크
데이터 웨어하우스, AI 및 디지털 혁신 시리즈
| # | Articolo | 집중하다 |
|---|---|---|
| 1 | 데이터 웨어하우스의 진화 | SQL Server에서 데이터 레이크하우스로 |
| 2 | 데이터 메시 및 분산형 아키텍처 | 데이터의 도메인 소유권 |
| 3 | ETL과 최신 ELT | dbt, 에어바이트, Fivetran |
| 4 | 파이프라인 오케스트레이션 | Airflow, Dagster 및 Prefect |
| 5 | 제조 분야의 AI | 예측 유지 관리 및 디지털 트윈 |
| 6 | 금융 속의 AI | 사기 탐지 및 신용 점수 |
| 7 | 소매업의 AI | 수요예측 및 추천 |
| 8 | 헬스케어 분야의 AI | 진단 및 약물 발견 |
| 9 | 물류 분야의 AI | 경로 최적화 및 창고 자동화 |
| 10 | 비즈니스 LLM | RAG Enterprise 및 가드레일 |
| 11 | 벡터 데이터베이스 엔터프라이즈 | pgVector, Pinecone 및 Weaviate |
| 12 | 비즈니스용 MLOps | MLflow를 사용하여 프로덕션 중인 AI 모델 |
| 13 | 현재 위치 - 데이터 거버넌스 | 안정적인 AI를 위한 데이터 품질 |
| 14 | 중소기업을 위한 데이터 기반 로드맵 | AI와 DWH를 도입하는 방법 |
AI 시대의 데이터 품질 문제
"가비지 인, 쓰레기 아웃"은 데이터 엔지니어가 수십 년 동안 알고 있던 원칙이지만 생성 AI에서는 그리고 프로덕션의 기계 학습 모델이 부족한 데이터로 인해 발생하는 결과는 다음과 같습니다. 기하급수적으로 더 심각해졌습니다. 불균형 데이터에 대해 훈련된 사기 탐지 모델은 가짜를 생성합니다. 합법적인 거래를 차단하는 긍정적인 요소입니다. 데이터에 대해 훈련된 추천 시스템 인구통계학적 편견은 차별을 증폭시킵니다. 훈련된 수요 예측 모델 일관되지 않은 데이터는 현금 흐름에 직접적인 영향을 미치는 잘못된 주문을 생성합니다.
품질이 낮은 데이터의 비용 - 2025년 데이터
| 지시자 | Valore | 원천 |
|---|---|---|
| 회사당 평균 비용(낮은 품질의 데이터) | 1,290만 달러/년 | 가트너 2024 |
| 데이터 품질 때문에 실패하는 AI 프로젝트 | 72% | 맥킨지 2025 |
| 데이터 과학자가 데이터 정리에 소요한 시간 | 45-60% | 여러 설문조사 |
| 공식 데이터 품질 프로그램을 갖춘 이탈리아 SME | 18% | AI 폴리미 2025 전망대 |
| 데이터 거버넌스를 통해 ML 오류 감소 | 최대 35% | IBM 연구소 2025 |
많은 조직이 간과하는 주요 차이점은 분석을 위한 데이터 품질 e AI를 위한 데이터 품질. 보고 중 전통적으로 비정상적인 데이터는 대시보드에 잘못된 숫자를 생성합니다. 누군가가 이를 알아차리면 그가 그를 바로잡으면 우리는 계속 진행합니다. 기계 학습에서는 훈련 세트의 변칙 데이터가 오염됩니다. 자동으로 모델을 생성한 후 몇 달 동안 프로덕션에서 오작동함 문제가 확인되었다는 것입니다. 문제와 그 발현 사이의 대기 시간 e 한없이 높아요.
데이터 품질의 6가지 DAMA 차원
프레임워크 DAMA-DMBOK (Data Management Body of Knowledge)는 6가지 차원을 정의합니다. 2025년에도 모든 분야의 표준 참조로 남아 있는 데이터 품질 기본 사항 기업 데이터 품질 프로그램:
AI용 DAMA 차원 - 확장 프레임워크
| 크기 | 정의 | AI 관련 지표 | 중요 임계값 |
|---|---|---|---|
| 정확성 | 데이터는 현실을 정확하게 나타냅니다. | 훈련 세트의 올바른 라벨 % | 중요 분류의 경우 >99% |
| 완전성 | 필요한 모든 데이터가 존재합니다 | 중요한 기능에 대한 null이 아닌 값 % | 입력 기능의 경우 >95% |
| 일관성 | 데이터는 다양한 시스템에서 일관됩니다. | 소스 간 일치 레코드 % | 공유 기능의 경우 >98% |
| 적시 | 데이터는 업데이트되어 필요할 때 사용할 수 있습니다. | 평균 생산 데이터 지연과 교육 비교 | 실시간 모델의 경우 24시간 미만 |
| 효력 | 데이터는 정의된 형식과 제약 조건을 따릅니다. | % 패턴 및 범위 위반 | <0.1% 위반 |
| 독창성 | 의도하지 않은 중복 없음 | 훈련 세트의 중복 레코드 % | <0.5% 중복 |
실제 구현: 기대가 큰 데이터 품질
큰 기대 테스트를 위한 가장 인기 있는 오픈 소스 Python 프레임워크 데이터 품질. 이 접근 방식은 코드에 대한 단위 테스트와 유사합니다. 이를 "기대"라고 합니다. (기대값)이 파이프라인에서 자동으로 실행되고 문서가 생성됩니다. 자동. Airflow, Prefect 및 dbt와의 통합으로 다음의 자연스러운 구성 요소가 됩니다. 모든 최신 데이터 스택.
# 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 reportdbt를 사용한 데이터 품질: 변환 파이프라인의 선언적 테스트
사용하는 팀의 경우 DBT 변환 레이어로 패키지 DBT 기대 Great Expectations와 동일한 기능을 제공합니다. dbt 모델에서 직접 SQL 코드에 가까운 YAML로 테스트를 정의합니다. 이 "코드로서의 품질" 접근 방식을 통해 각 변환이 자동으로 검증됩니다.
# 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"]OpenMetadata를 사용한 데이터 카탈로그 및 데이터 계보
La 주어진 혈통 - 소스에서 데이터의 경로를 추적하는 기능 최종 AI 모델까지 - 다음 두 가지 이유로 인해 필수 요구 사항이 되었습니다. AI Act EU 준수(훈련 데이터의 출처에 대한 문서화 필요) 모델이 예상치 못한 결과를 생성할 때 디버깅이 실제로 필요합니다.
오픈메타데이터 가장 발전된 오픈 소스 데이터 카탈로그 및 계보 플랫폼 2025년에는 전 Uber 엔지니어와 Apache Hadoop 창립자가 구축했습니다. 계보를 지원합니다 열 수준에서는 dbt, Airflow, Spark 및 기본 데이터 웨어하우스와의 기본 통합이 가능합니다.
# 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."
)데이터 거버넌스 프레임워크: 조직 구조 및 프로세스
데이터 거버넌스는 소프트웨어 도구가 아니라 시스템입니다. 사람, 프로세스 그리고 기술 이는 데이터가 전략적 자산으로 관리되도록 보장합니다. 중소기업용 AI에 접근하는 이탈리아 기업은 실용적이고 필수적인 방식으로 이러한 구조를 구축합니다.
데이터 거버넌스의 주요 역할
중소기업을 위한 지배구조(직원 50~500명)
| 역할 | 책임 | 예상 FTE | 윤곽 |
|---|---|---|---|
| 최고 데이터 책임자(CDO) | 데이터 전략, 임원 후원, 예산, AI법 준수 | 0.25 FTE(일부) | C 레벨 또는 고위 관리자 |
| 데이터 관리자 | 데이터 도메인 소유권, 표준 정의, 변경 승인 | 데이터 도메인당 1개 | 비즈니스 + 기술 |
| 데이터 엔지니어 | 파이프라인, 기술 품질, 모니터링 도구 | FTE 1~2명 | 기술 프로필 |
| 데이터 품질 분석가 | 품질 KPI, 감사, 보고 정의 | 0.5 FTE | 분석/하이브리드 |
| DPO(데이터 보호 책임자) | GDPR, AI법, 데이터 보안, 개인정보 보호 설계 | 0.25-0.5 FTE | 법률/기술 |
데이터 거버넌스 정책: 5가지 기본 프로세스
운영 데이터 거버넌스 프로그램은 반드시 수행해야 하는 5가지 반복 프로세스를 기반으로 합니다. 문서화, 할당 및 모니터링:
- 데이터 분류: 민감도별로 분류된 각 데이터 세트(공개, 내부, 기밀, 비밀) 및 유형별(PII, 재무 데이터, 운영 데이터, ML 교육 데이터). 이 분류에 따라 액세스 제어, 보존 정책 및 요구 사항이 결정됩니다. 적용 가능한 품질의.
- 데이터 품질 SLA: 각 중요 데이터 세트에 대해 공식적인 SLA 정의는 다음과 같습니다. 품질: 완전성 > 98%, 지연 시간 < 2시간, 정확도는 분기별로 검증됩니다. SLA는 측정 가능하고 자동으로 모니터링되어야 합니다.
- 변경 관리: 데이터 스키마 변경을 위한 공식 프로세스 비평가. 영향 평가, 회귀 테스트 및 승인 없이는 변경 사항이 없습니다. 도메인 데이터 관리자의.
- 데이터 액세스 관리: 데이터에 적용되는 최소 권한 원칙. 전체 감사 로그가 포함된 ML 팀만 ML 교육 데이터에 액세스할 수 있습니다. 검토 분기별 액세스.
- 사고 대응: 데이터 품질 사고에 대한 에스컬레이션 절차: 자동 감지(경고), 분류(<4시간), 교정, 사후 분석 자동 점검 업데이트.
AI Act EU 및 교육 데이터 요구 사항
L'AI법 EU 역사상 처음으로 법적 요구 사항을 도입합니다. 고위험 AI 시스템의 훈련 데이터 품질에 대한 제약. 그만큼'제10조 규정에 따라 특별히 "데이터 및 데이터"를 전담합니다. 거버넌스"이며 이후 GPAI 모델에 대해 작동하게 되었습니다. 2025년 8월 2일. 고위험 시스템의 경우 다음 사항을 완전히 준수해야 합니다. 2026년 8월 2일.
AI Act EU 타임라인 - 2026년 8월 이전에 해야 할 일
| 날짜 | 마일스톤 | 필요한 조치 |
|---|---|---|
| 2025년 2월 | 운영상 금지된 AI 관행 | 금지된 관행에 대한 AI 시스템 감사 |
| 2025년 8월 | GPAI 운영 모델 및 거버넌스 | LLM 및 기본 모델에 대한 거버넌스 |
| 2026년 8월 | 고위험 AI 시스템 전체 규정 준수 | 데이터 거버넌스 예술. 10개 완전 구현 |
| 2027년 8월 | 레거시 시스템 규정 준수 | 적절한 기존 AI 시스템 |
제10조 AI법: 데이터에 대한 기술적 요구사항
AI법 10조는 훈련에 사용되는 데이터 세트에 대한 구체적인 요구 사항을 규정하고 있습니다. 고위험 AI 시스템의 검증 및 테스트. 문서화해야 할 사항은 다음과 같습니다. 구체적으로 구현:
# 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 report데이터 관찰 가능성: 생산 파이프라인의 지속적인 모니터링
데이터 품질은 수집 시점에만 보장되지 않습니다. 시간이 지남에 따라 데이터 품질이 저하됩니다. 시간. 현상 데이터 드리프트 - 데이터 배포 위치 생산이 훈련 세트에서 점차 벗어나는 주요 원인 중 하나 AI 모델의 조용한 저하. 데이터 관측 가능성이 이 문제를 해결합니다. 지속적인 모니터링과 사전 경고를 통해
2025년 데이터 관찰 도구 비교
| 기구 | 유형 | 강점 | 이상적인 사용 사례 |
|---|---|---|---|
| 소다 코어 | 오픈 소스 | YAML 기반, CLI, CI/CD에 내장 가능 | 예산이 제한된 중소기업, DBT 파이프라인 |
| 몬테카를로 | SaaS(기업) | ML 기반의 제로 구성 이상 탐지 | 기업, 대규모, 소규모 팀 |
| 큰 기대 | 오픈 소스 | Python 기반의 유연한 자동 문서화 | 데이터 엔지니어링 팀, Python 파이프라인 |
| DBT 테스트 | 오픈 소스 | dbt 워크플로우에 통합된 코드로서의 품질 | dbt를 메인 레이어로 사용하는 팀 |
| 메타플레인 | SaaS(중간 규모 시장) | 계보 + 가시성, 직관적인 UI | 여러 데이터 소스로 확장 |
Soda Core를 사용한 데이터 드리프트 감지
소다 코어 YAML에서 품질 검사를 정의하고 실행할 수 있습니다. 예약된 방식 또는 주문형 방식으로. 데이터 드리프트 모니터링의 경우 비교가 가능합니다. 서로 다른 기간 사이의 중요한 특징 분포를 자동으로 감지 AI 모델을 저하시킬 수 있는 중대한 변화.
# 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_digestML 데이터 세트의 편향 감지: 체계적인 접근 방식
Il 데이터의 편향 엔터프라이즈 머신러닝에서 가장 까다로운 문제 중 하나는 다음과 같습니다. 조용하고 적절한 방법론 없이는 탐지하기 어렵고 잠재적으로 차별적임 보호받는 집단을 향해. AI법에서는 사용되는 AI 시스템을 "고위험"으로 분류합니다. 직업, 신용, 공공 서비스 및 교육에 대한 접근 결정, 편견 탐지 이는 직업 윤리뿐만 아니라 규정 준수의 요구 사항이 되었습니다.
훈련 데이터의 편향은 세 가지 주요 소스에서 발생합니다. 역사적 편견 (과거 차별을 반영하는 데이터, 예: 역사적으로 불균형한 승진률) 표현편향 (훈련 세트에서 과소 대표되는 그룹) 전자 측정 편향 (보호된 속성에 대한 프록시 역할을 하는 기능, 예를 들어 우편번호는 소득과 민족을 대신합니다.
공정성 지표 - IBM AIF360 및 AI Act 표준
| 미터법 | 공식 | AI법 기준 | 일반적인 사용 |
|---|---|---|---|
| 이질적인 영향 | P(Y=1|A=0) / P(Y=1|A=1) | >= 0.80 | 채용, 신용 |
| 기회균등 | TPR(A=0) - TPR(A=1) | |차이| < 0.05 | 의료, 정의 |
| 인구통계학적 동등성 | P(Y=1|A=0) - P(Y=1|A=1) | |차이| < 0.05 | 권장 사항 |
| 구경 측정 | P(Y=1|점수=s, A=a) 균일 | RMSE < 0.02 | 위험 점수 |
사례 연구: 이탈리아 제조 중소기업을 위한 데이터 거버넌스
직원이 180명인 롬바르디아의 한 제조 중소기업은 2024년에 데이트 프로그램을 시작했습니다. 세 가지 AI 이니셔티브를 지원하는 거버넌스: 기계에 대한 예측 유지 관리 모델 컴퓨터 비전 및 수요 예측 기반의 시각적 품질 관리 시스템인 CNC 생산 계획을 위해.
과제: 데이터는 어디에나 있었고 이에 대해 책임을 지는 사람은 없었습니다.
초기 상황은 이탈리아 SME의 전형적인 상황이었습니다. 독점 형식, SAP Business One ERP의 생산 데이터, 품질 데이터 이메일을 통해 공유되는 Excel 시트, 통합되지 않은 CRM의 판매 데이터. IT팀은 인프라에 대한 기술적 비전은 있지만 공식적인 거버넌스는 없습니다.
데이터 거버넌스 프로그램 결과(12개월)
| 지시자 | 전에 | 12개월 후 | 영향 |
|---|---|---|---|
| 평균 데이터 품질 점수 | 61% | 94% | +33포인트 |
| ML 데이터 준비 시간 | 3~4주 | 2~3일 | -90% |
| 문서화된 계보가 있는 데이터세트 | 0% | 87% | +87점 |
| 생산 중인 AI 모델 | 0 | 3 | +3 시스템 |
| AI법 준수 조항. 10 | 평가되지 않음 | 준수 | 감사 준비 |
4단계 구현 로드맵
이 프로그램은 4개 분기별 단계로 구성되었으며 총 투자액은 대략 85,000 EUR(PNRR 전환 5.0에서 일부 보장):
- 1단계 - 인벤토리 및 평가(2024년 1분기): 모든 시스템의 매핑 데이터, AI 중요도에 따른 데이터세트 분류, 초기 품질 평가. 도구: 도메인 관리자를 위한 인벤토리 시트 + 자체 평가 설문지. 결과: 격차 분석을 통해 중요한 데이터의 전체 지도.
- 2단계 - 기초(2024년 2분기): OpenMetadata 온프레미스 배포, 첫 번째 데이터 카탈로그 구성, 각 도메인에 대한 데이터 관리자 정의 (생산, 품질, 판매). DBT 테스트를 통한 최초의 자동 검사 구현.
- 3단계 - 품질 자동화(2024년 3분기): 큰 기대를 품으세요 ML 데이터 세트의 자동 검증, 알림을 위한 Airflow와의 통합, 3가지 AI 프로젝트에 대한 품질 SLA 정의. 첫 평가 AI Act Art. 10.
- 4단계 - 최적화 및 성숙한 거버넌스(2024년 4분기): 데이터 위원회 활성 월별, 메타베이스 대시보드의 품질 KPI, 변경 관리 프로세스 공식 직원 교육. 3가지 AI 모델이 프로덕션에 배포되었습니다. 완벽한 데이터 거버넌스를 통해
교훈: 피해야 할 5가지 실수
- 프로세스 대신 도구에서 시작: 기술은 나중에 나온다 사람과 프로세스. 소프트웨어를 설치하기 전에 데이터 관리자를 지정하십시오.
- 모든 일을 즉시 하고 싶다: 진화적인 거버넌스 프로그램 등 빅뱅보다 지속 가능하다. AI를 위한 중요한 데이터 세트로 시작하여 점차 확장하세요.
- 훈련을 과소평가: 예산의 40%가 투자됐다. 교육 및 변화 관리. 이것이 없으면 도구는 무시됩니다.
- 비즈니스에 관여하지 마세요. 데이터 관리자는 하이브리드 인물이어야 합니다. IT뿐만 아니라 비즈니스 기술도 마찬가지입니다. 기업은 어떤 데이터가 중요한지 알고 있고, IT는 어떻게 중요한지 알고 있습니다. 그들을 측정하십시오.
- 처음부터 규정 준수를 무시합니다. AI법 요구사항은 다음과 같습니다. 제약: 거버넌스에 대한 투자를 정당화하는 구조입니다. 관리의.
모범 사례 및 안티 패턴
기본 모범 사례
- 코드로서의 품질: 코드에서 품질 요구 사항 정의(YAML, Python) Word 문서가 아닌 Git 버전으로 관리됩니다. 이를 통해 검사를 자동으로 재현 가능하게 만듭니다. CI/CD 프로세스의 일부입니다.
- 빠른 실패, 큰 실패: 데이터 품질 검사는 파이프라인을 차단해야 합니다. (단지 경고를 생성하는 것이 아님) 데이터가 범위를 벗어날 때. 훈련된 모델 잘못된 데이터는 파이프라인이 중단된 것보다 더 큰 피해를 입힙니다.
- 변환과 별도의 검증: 데이터를 변환하기 전에 유효성을 검사하세요. (수집 시 구성표 및 범위 검증), 변환 중(모델의 DBT 테스트), ML에서 사용하기 전(피처 스토어의 Great Expectations) 품질이 발생합니다 끝뿐만 아니라 여러 지점에서.
- 정적 품질뿐만 아니라 드리프트도 모니터링합니다. 시간이 지남에 따라 데이터가 변경됩니다. PSI(인구 안정성 지수) 및 KS(Kolmogorov-Smirnov) 테스트는 표준 도구입니다. 모델을 조용히 저하시키는 분포 드리프트를 감지합니다.
- 문서 거버넌스 결정: 각 선택(품질 임계값, 기여 전략, 데이터 세트 제외)는 날짜와 함께 문서화되어야 합니다. 저자이자 합리적이다. 이 문서는 고위험 시스템에 대한 AI법에 의해 요구됩니다.
피해야 할 안티패턴
AI 데이터 거버넌스의 중요한 안티 패턴
- "최선의 노력" 데이터 품질: 모호한 SLA를 정의합니다("데이터를 최대한 많이 완전") 대신 정확하고 측정 가능한 수치 임계값을 사용합니다. 측정항목이 없으면 아무것도 없습니다. 진정한 거버넌스.
- 알림 음소거: 품질 검사를 구성한 후 침묵 "방해"하기 때문에 경고합니다. 무시된 모든 경고는 미래 모델 문제입니다. 생산 중.
- 새 프로젝트에만 적용되는 거버넌스: 다음 용도로 사용되는 레거시 데이터 세트 모델 재훈련에는 동일한 품질 요구 사항이 있습니다. 그들은 종종 가장 문제가 있다.
- 원샷 바이어스 검사: 훈련 전에만 편향을 확인하세요. 초기. 시간이 지남에 따라 새로운 생산 데이터로 인해 편견이 나타날 수 있습니다(개념 드리프트).
- 분석 및 ML의 혼란스러운 데이터 품질: 허용되는 임계값 대시보드의 경우(열의 5% null) 기능에 치명적일 수 있습니다. ML 입력의 수입니다. 두 가지 상황에는 서로 다른 표준이 필요합니다.
결론 및 다음 단계
AI를 위한 데이터 거버넌스와 데이터 품질은 관료주의가 아니라 인프라입니다. 모델이 프로덕션에서 작동하는지 또는 자동으로 실패하는지 여부를 결정하는 보이지 않습니다. AI Act EU가 훈련 데이터에 대한 법적 구속력 있는 요구 사항을 가져오면서, 거버넌스에 대한 투자는 단순한 모범 사례가 아니라 운영을 위한 전제 조건입니다. 고위험 AI 시스템을 갖춘 유럽 시장에서.
Il 실용적인 출발점 모든 조직 - 중소기업 또는 기업 - 그리고 동일합니다. AI 이니셔티브에 가장 중요한 데이터세트 3~5개를 식별하고 이름을 지정하세요. 각각에 대한 데이터 관리자, 오픈 소스 도구를 사용하여 자동 검사 구현 이 기사(Great Expectations, dbt-expectations, Soda Core)에 설명되어 있습니다. 점차적으로 이러한 데이터 세트를 중심으로 거버넌스 구조를 구축합니다.
완벽함은 초기 요구사항이 아닙니다. 94%의 데이터 품질 점수를 달성했습니다. 우리 사례 연구에서 12개월 후 제조업 PMI는 61%로 시작되었습니다. 여행 목적지만큼 중요하며 품질 향상의 모든 백분율 포인트 데이터가 더욱 신뢰할 수 있는 AI 모델로 직접 변환되고 사고가 줄어듭니다. 생산 및 보다 강력한 비즈니스 결정을 내릴 수 있습니다.
AI 프로그램 출시를 위한 데이터 거버넌스 체크리스트
- AI 핵심 데이터세트 인벤토리 완료
- 각 주요 데이터 도메인에 대해 지정된 데이터 관리자
- 품질 SLA 정의 및 승인(완전성, 신선도, 타당성)
- 파이프라인(GE, dbt-expectations 또는 Soda)에 자동 검사 구현
- 계보가 구성된 데이터 카탈로그(OpenMetadata 또는 Apache Atlas)
- 고위험 시스템에 사용되는 모든 데이터 세트에 대해 편향 평가 수행
- AI 행위 예술. 고위험 시스템에 대한 10개 문서화 시작
- 구성된 경고 및 정의된 에스컬레이션 채널
- 첫 번째 월간 데이터 협의회 예정
- 팀 훈련 완료
관련 통찰력
- 비즈니스용 MLOps: 생산 중인 모델의 드리프트를 모니터링하는 방법 MLflow를 사용하여 - 이 시리즈의 기사 #12
- 회사 내 LLM: RAG 엔터프라이즈 및 보안을 위한 데이터 거버넌스 대규모 언어 모델의 데이터 - 이 시리즈의 기사 #10
- AI 엔지니어링: 시스템에 대한 특성 저장소 및 내장형 거버넌스 RAG 기업 - AI 엔지니어링 시리즈
- 포스트그레SQL AI: 벡터 데이터베이스의 pgVector 및 데이터 품질 - PostgreSQL AI 시리즈