Data Pipeline

Step-by-Step Workflow

Phase 1 — Data Ingestion & Profiling

Objective: Load data and create a comprehensive quality profile.

1. Load data (auto-detect format):

   import pandas as pd
   
   # Auto-detect format
   if filepath.endswith('.csv'):
       df = pd.read_csv(filepath, encoding='utf-8-sig')
   elif filepath.endswith('.json'):
       df = pd.read_json(filepath)
   elif filepath.endswith('.jsonl'):
       df = pd.read_json(filepath, lines=True)
   elif filepath.endswith('.parquet'):
       df = pd.read_parquet(filepath)
   

2. Generate data profile:

   profile = {
       "total_rows": len(df),
       "total_columns": len(df.columns),
       "column_types": df.dtypes.to_dict(),
       "missing_values": df.isnull().sum().to_dict(),
       "missing_percentage": (df.isnull().sum() / len(df) * 100).to_dict(),
       "duplicates": df.duplicated().sum(),
       "memory_usage_mb": df.memory_usage(deep=True).sum() / 1e6,
   }
   

3. Generate data_quality_report.md with all findings.

Phase 2 — Data Cleaning & Transformation

Objective: Clean data based on modality.

NLP Data Cleaning:

import re
import unicodedata

def clean_text(text):
    """Standard NLP text cleaning pipeline."""
    if pd.isna(text):
        return ""
    text = unicodedata.normalize("NFKD", str(text))
    text = re.sub(r'<[^>]+>', '', text)          # Remove HTML
    text = re.sub(r'http\S+|www.\S+', '', text)  # Remove URLs
    text = re.sub(r'\s+', ' ', text).strip()      # Normalize whitespace
    return text

Tabular Data Cleaning:

def clean_tabular(df):
    """Standard tabular data cleaning."""
    # Remove constant columns
    constant_cols = [c for c in df.columns if df[c].nunique() <= 1]
    df = df.drop(columns=constant_cols)
    
    # Handle missing values
    numeric_cols = df.select_dtypes(include='number').columns
    categorical_cols = df.select_dtypes(include='object').columns
    
    df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())
    df[categorical_cols] = df[categorical_cols].fillna(df[categorical_cols].mode().iloc[0])
    
    # Remove duplicates
    df = df.drop_duplicates()
    
    return df

Decision Table — Missing Value Strategy:

Output: Cleaned dataset + preprocessing_pipeline.py

Phase 3 — Data Validation & Quality Checks

20-Item Quality Checklist:

Phase 4 — Model Evaluation & Metrics

Objective: Evaluate an ML/LLM model with structured metrics.

Metrics by Task Type:

Evaluation Harness Template:

class ModelEvaluator:
    def __init__(self, model, tokenizer, metrics=None):
        self.model = model
        self.tokenizer = tokenizer
        self.metrics = metrics or ["accuracy", "f1"]
    
    def evaluate(self, test_dataset):
        predictions = []
        references = []
        
        for sample in test_dataset:
            pred = self.model.predict(sample["input"])
            predictions.append(pred)
            references.append(sample["label"])
        
        results = {}
        for metric_name in self.metrics:
            metric = evaluate.load(metric_name)
            results[metric_name] = metric.compute(
                predictions=predictions,
                references=references
            )
        
        return results
    
    def generate_report(self, results, output_path):
        # Generate markdown report with charts
        ...

Output: eval_report.md with metrics, charts, analysis

Phase 5 — Performance & Load Testing

Objective: Test AI API performance under load.

Locust Load Test Template:

from locust import HttpUser, task, between

class AIAPIUser(HttpUser):
    wait_time = between(0.5, 2)
    
    @task
    def predict(self):
        self.client.post("/predict", json={
            "text": "Sample input for prediction"
        })
    
    @task
    def health(self):
        self.client.get("/health")

Metrics to collect:

• Latency: p50, p95, p99
• Throughput: requests/second
• Error rate: % of 4xx/5xx responses
• Response time distribution

Run: locust -f locustfile.py --host http://localhost:8000 --headless -u 100 -r 10 -t 60s

Output: performance_report.md + locustfile.py

Phase 6 — Report Generation

Compile all findings into structured reports:

• data_quality_report.md — from Phase 1-3
• eval_report.md — from Phase 4
• performance_report.md — from Phase 5

All reports include: findings, charts, recommendations, next steps.

Anti-Patterns — DO NOT

Skill Coordination

• Use ai-solution-dev when the data workflow is part of a broader AI product delivery effort.
• Use finetuning when evaluation results indicate model adaptation is required.
• Use ai-integration when the validated model must be exposed through serving endpoints.
• Use ai-reliability when performance testing needs to become long-term monitoring or SLO work.
• Use documentation to capture data reports, evaluation reports, and test evidence.