Capstone 3: AI Data Quality & Anomaly Detector

+---------------------------------------------------------------------------------+
|                           PROJECT 3 ARCHITECTURE                                |
+---------------------------------------------------------------------------------+
|  Data Source (Pandas/Snowflake) -> Profile Stats (Min, Max, Nulls, Distinct)    |
|                                             |                                   |
|                                             v                                   |
|  LLM Quality Inspector          -> Detects semantic drift, formatting rot, & PII|
|                                             |                                   |
|                                             v                                   |
|  Automated RCA & Alert          -> Generates Slack markdown report & dbt tests  |
+---------------------------------------------------------------------------------+

Objective

Detect silent data failures that pass traditional schema and null checks, such as corrupted currency symbols, mock test strings, or prompt injection attempts in incoming data lakes.

Complete Implementation

import pandas as pd
from pydantic import BaseModel, Field
from openai import OpenAI
import instructor

client = instructor.from_openai(OpenAI())

class ColumnAnomaly(BaseModel):
    column_name: str
    anomaly_detected: bool
    risk_level: str = Field(description="LOW, MEDIUM, HIGH, CRITICAL")
    reason: str
    sample_bad_records: list[str]
    suggested_dbt_test: str

class DataQualityAudit(BaseModel):
    dataset_name: str
    passed_audit: bool
    findings: list[ColumnAnomaly]

def audit_dataset_semantics(df: pd.DataFrame, dataset_name: str) -> DataQualityAudit:
    column_summaries = {}
    for col in df.columns:
        column_summaries[col] = {
            "dtype": str(df[col].dtype),
            "sample_distinct": df[col].dropna().astype(str).unique()[:10].tolist()
        }

    prompt = f"""
You are a Principal Data Quality Engineer.
Analyze the following sample column profiles for semantic drift, injection attacks, mock test data (e.g. 'asdf', 'test@test.com'), or formatting inconsistencies.

DATASET: {dataset_name}
PROFILES:
{column_summaries}
"""
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=DataQualityAudit,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0
    )