Version: 0.9.1
This guide walks you through using the Fairness Pipeline Development Toolkit across the complete model development cycle—from initial data exploration to production monitoring. Whether you’re building a new model or improving an existing one, this toolkit provides the tools and workflows to ensure fairness at every stage.
The Fairness Pipeline Development Toolkit is a unified, statistically-rigorous framework for detecting, mitigating, training, and validating fairness in ML workflows. It provides:
# Create virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Install core toolkit
pip install -e .[adapters]
# Optional: Install training and monitoring extras
pip install -e .[training,monitoring]
For reproducible environments:
pip install pip-tools
pip-compile --extra training --extra monitoring --extra adapters \
--output-file=requirements.txt requirements-dev.in
pip install -r requirements.txt
Note: PyTorch wheels depend on platform/accelerator support. Follow the commands from pytorch.org/get-started before enabling the
trainingextra.
fairpipe version
You should see the version number (e.g., 0.9.1).
The toolkit supports fairness work at every stage of the ML lifecycle:
┌─────────────────────────────────────────────────────────────┐
│ MODEL DEVELOPMENT CYCLE │
└─────────────────────────────────────────────────────────────┘
Phase 1: Data Preparation & Exploration
↓
Phase 2: Baseline Fairness Measurement
↓
Phase 3: Bias Detection & Mitigation
↓
Phase 4: Fairness-Aware Model Training
↓
Phase 5: Validation & Testing
↓
Phase 6: Deployment Preparation
↓
Phase 7: Production Monitoring
Each phase has specific tools and workflows. You can:
import pandas as pd
import numpy as np
# Load your dataset
df = pd.read_csv("your_data.csv")
# Inspect structure
print(f"Dataset shape: {df.shape}")
print(f"Columns: {list(df.columns)}")
print(df.head())
print(df.describe())
Sensitive attributes are protected characteristics that should not lead to unfair treatment. Common examples:
# Example: Identify sensitive attributes
sensitive_attributes = ["gender", "race"] # Adjust to your dataset
# Check distributions
for attr in sensitive_attributes:
print(f"\n{attr} distribution:")
print(df[attr].value_counts(normalize=True))
# Check for missing values
print("Missing values:")
print(df.isnull().sum())
# Check for sufficient group sizes
for attr in sensitive_attributes:
print(f"\n{attr} group sizes:")
print(df[attr].value_counts())
Minimum Group Size: Ensure each group has at least 30 samples for statistical reliability. Smaller groups may need special handling.
# Identify target column
target_column = "y" # or "label", "target", etc.
# Check target distribution
print(f"Target distribution:")
print(df[target_column].value_counts(normalize=True))
# Check target by sensitive attribute (initial bias check)
for attr in sensitive_attributes:
print(f"\nTarget distribution by {attr}:")
print(pd.crosstab(df[attr], df[target_column], normalize='index'))
At this stage, you should have:
Next: Move to Phase 2 to measure baseline fairness.
The fastest way to measure baseline fairness:
fairpipe validate \
--csv your_data.csv \
--y-true y_true \
--y-pred y_pred \
--sensitive gender \
--with-ci \
--ci-level 0.95 \
--with-effects \
--out baseline_report.md
For more control and integration:
from fairpipe.metrics import FairnessAnalyzer
# Initialize analyzer
analyzer = FairnessAnalyzer(backend="native") # or "fairlearn", "aequitas"
# Load data
df = pd.read_csv("your_data.csv")
# Measure baseline fairness
# Note: For baseline, you may not have predictions yet
# You can measure data-level fairness or use a simple baseline model
# If you have predictions:
baseline_results = analyzer.analyze(
y_true=df["y_true"],
y_pred=df["y_pred"], # or baseline model predictions
sensitive_features=df[["gender", "race"]],
min_group_size=30
)
# View results
print("Baseline Fairness Metrics:")
for metric_name, result in baseline_results.items():
print(f"\n{metric_name}:")
print(f" Value: {result.value:.4f}")
if result.ci:
print(f" 95% CI: [{result.ci[0]:.4f}, {result.ci[1]:.4f}]")
if result.effect_size:
print(f" Effect Size: {result.effect_size:.4f}")
max(selection_rate) - min(selection_rate)The toolkit provides confidence intervals and effect sizes:
# With confidence intervals
results = analyzer.analyze(
y_true=df["y_true"],
y_pred=df["y_pred"],
sensitive_features=df[["gender"]],
with_ci=True,
ci_level=0.95,
with_effects=True
)
# Results include:
# - Bootstrap confidence intervals
# - Effect sizes (risk ratio, Cohen's d)
# - Sample sizes per group
Analyze fairness across combinations of sensitive attributes:
# Analyze intersectional groups
results = analyzer.analyze(
y_true=df["y_true"],
y_pred=df["y_pred"],
sensitive_features=df[["gender", "race"]],
intersections=[["gender", "race"]], # Analyze gender×race combinations
min_group_size=30 # Minimum samples per intersectional group
)
At this stage, you should have:
Next: Move to Phase 3 to detect and mitigate bias.
Create a YAML configuration file (pipeline.config.yml):
# Required: Sensitive attributes
sensitive: ["gender", "race"]
# Optional: Population benchmarks for representation checks
benchmarks:
gender:
M: 0.5
F: 0.5
race:
A: 0.3
B: 0.3
C: 0.4
# Statistical test parameters
alpha: 0.05
proxy_threshold: 0.30 # Correlation threshold for proxy detection
# Pipeline: Bias mitigation transformers (applied in order)
pipeline:
- name: reweigh
transformer: "InstanceReweighting"
params: {}
- name: repair
transformer: "DisparateImpactRemover"
params:
features: ["score", "age", "education"]
sensitive: "gender"
repair_level: 0.8
- name: drop_proxies
transformer: "ProxyDropper"
params:
threshold: 0.30
sensitive: ["gender", "race"]
features: List of feature columns to repairsensitive: Sensitive attribute to userepair_level: Strength of repair (0.0 to 1.0)strategy (e.g., “none”, “demographic_parity”)threshold: Correlation threshold (default: 0.30)sensitive: List of sensitive attributesThe pipeline automatically detects bias before mitigation:
fairpipe pipeline \
--config pipeline.config.yml \
--csv your_data.csv \
--out-csv artifacts/transformed_data.csv \
--detector-json artifacts/detectors.json \
--report-md artifacts/pipeline_report.md
This will:
import json
# Load detection results
with open("artifacts/detectors.json", "r") as f:
detectors = json.load(f)
print("Bias Detection Results:")
for detector_name, results in detectors.items():
print(f"\n{detector_name}:")
print(f" Issues Found: {results.get('issues_found', 0)}")
if results.get('issues'):
for issue in results['issues']:
print(f" - {issue}")
# Load original and transformed data
original = pd.read_csv("your_data.csv")
transformed = pd.read_csv("artifacts/transformed_data.csv")
# Compare distributions
print("Original data - Gender distribution:")
print(original["gender"].value_counts(normalize=True))
print("\nTransformed data - Gender distribution:")
print(transformed["gender"].value_counts(normalize=True))
# Compare feature statistics
print("\nFeature statistics comparison:")
print(original[["score", "age"]].describe())
print(transformed[["score", "age"]].describe())
At this stage, you should have:
Next: Move to Phase 4 to train a fairness-aware model.
The toolkit supports three training approaches:
Best for: scikit-learn models, constraint-based fairness
from fairpipe.training import ReductionsWrapper
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
# Load transformed data
df = pd.read_csv("artifacts/transformed_data.csv")
# Prepare features and target
X = df.drop(columns=["y", "gender", "race"])
y = df["y"]
A = df[["gender"]] # Sensitive attributes
# Split data
X_train, X_test, y_train, y_test, A_train, A_test = train_test_split(
X, y, A, test_size=0.2, random_state=42
)
# Create base estimator
base_estimator = LogisticRegression()
# Wrap with fairness constraints
fair_model = ReductionsWrapper(
estimator=base_estimator,
constraint="demographic_parity", # or "equalized_odds"
eps=0.01 # Constraint tolerance
)
# Train
fair_model.fit(X_train, y_train, sensitive_features=A_train)
# Predict
y_pred = fair_model.predict(X_test)
y_proba = fair_model.predict_proba(X_test)[:, 1]
Best for: Deep learning models, exploring fairness-accuracy trade-offs
from fairpipe.training.torch_.losses import FairnessRegularizerLoss
import torch
import torch.nn as nn
# Define model
class SimpleNN(nn.Module):
def __init__(self, input_dim):
super().__init__()
self.fc1 = nn.Linear(input_dim, 64)
self.fc2 = nn.Linear(64, 32)
self.fc3 = nn.Linear(32, 1)
self.sigmoid = nn.Sigmoid()
def forward(self, x):
x = torch.relu(self.fc1(x))
x = torch.relu(self.fc2(x))
return self.sigmoid(self.fc3(x))
# Prepare data
X_tensor = torch.FloatTensor(X_train.values)
y_tensor = torch.FloatTensor(y_train.values).unsqueeze(1)
A_tensor = torch.FloatTensor(A_train.values)
# Initialize model
model = SimpleNN(X_train.shape[1])
# Define loss with fairness regularizer
criterion = FairnessRegularizerLoss(
base_loss=nn.BCELoss(),
eta=0.5, # Fairness regularization strength
sensitive_features=A_tensor
)
# Training loop
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(50):
optimizer.zero_grad()
outputs = model(X_tensor)
loss = criterion(outputs, y_tensor, sensitive_features=A_tensor)
loss.backward()
optimizer.step()
Best for: Strict fairness constraints, dual optimization
from fairpipe.training import LagrangianFairnessTrainer
# Prepare data
train_data = {
"X": X_train.values,
"y": y_train.values,
"sensitive": A_train["gender"].values
}
# Initialize trainer
trainer = LagrangianFairnessTrainer(
model=model,
fairness="demographic_parity", # or "equal_opportunity"
dp_tol=0.02, # Demographic parity tolerance
model_lr=0.001,
lambda_lr=0.01,
epochs=50,
batch_size=128,
device="cpu" # or "cuda"
)
# Train
history = trainer.train(train_data)
# Access trained model
trained_model = trainer.model
Use the Pareto Frontier visualization to explore trade-offs:
fairpipe train-regularized \
--csv artifacts/transformed_data.csv \
--etas "0.0,0.2,0.5,1.0,2.0" \
--epochs 50 \
--lr 1e-3 \
--out-json artifacts/pareto_points.json \
--out-png artifacts/pareto.png
This generates a visualization showing the trade-off between accuracy and fairness for different regularization strengths.
Calibrate probabilities to ensure fairness across groups:
fairpipe calibrate \
--csv scores.csv \
--method platt \
--min-samples 20 \
--out-csv artifacts/calibrated_scores.csv
Or in Python:
from fairpipe.training.postproc import GroupFairnessCalibrator
calibrator = GroupFairnessCalibrator(
method="platt", # or "isotonic"
min_samples=20
)
calibrator.fit(y_true=y_train, y_score=y_proba_train, sensitive_features=A_train)
y_calibrated = calibrator.transform(y_score=y_proba_test, sensitive_features=A_test)
At this stage, you should have:
Next: Move to Phase 5 to validate the model.
The integrated workflow automatically validates:
fairpipe run-pipeline \
--config config.yml \
--csv your_data.csv \
--output-dir artifacts/ \
--mlflow-experiment fairness_workflow \
--min-group-size 30 \
--train-size 0.8
This runs:
from fairpipe.metrics import FairnessAnalyzer
# Measure final model fairness
analyzer = FairnessAnalyzer(backend="native")
final_results = analyzer.analyze(
y_true=y_test,
y_pred=y_pred,
sensitive_features=A_test,
min_group_size=30,
with_ci=True,
with_effects=True
)
# Compare to baseline
print("Baseline vs Final Metrics:")
baseline_dp = baseline_results["demographic_parity_difference"].value
final_dp = final_results["demographic_parity_difference"].value
print(f"Demographic Parity Difference:")
print(f" Baseline: {baseline_dp:.4f}")
print(f" Final: {final_dp:.4f}")
print(f" Improvement: {baseline_dp - final_dp:.4f}")
# Check threshold
threshold = 0.05
if abs(final_dp) <= threshold:
print(f"✅ Validation PASSED: {final_dp:.4f} <= {threshold}")
else:
print(f"❌ Validation FAILED: {final_dp:.4f} > {threshold}")
Use the pytest plugin for automated testing:
# test_fairness.py
import pytest
from fairpipe.integration.pytest_plugin import assert_fairness
def test_model_fairness():
"""Test that model meets fairness threshold."""
assert_fairness(
y_true=y_test,
y_pred=y_pred,
sensitive_features=A_test,
metric="demographic_parity_difference",
threshold=0.05,
min_group_size=30
)
Run tests:
pytest test_fairness.py -v
Generate a comprehensive validation report:
from fairpipe.integration.reporting import generate_validation_report
report = generate_validation_report(
baseline_metrics=baseline_results,
final_metrics=final_results,
validation_result=validation_result,
output_path="artifacts/validation_report.md"
)
Log results to MLflow for tracking:
from fairpipe.integration.mlflow_logger import log_workflow_results
import mlflow
mlflow.set_experiment("fairness_workflow")
with mlflow.start_run():
log_workflow_results(
baseline_metrics=baseline_results,
final_metrics=final_results,
validation_result=validation_result,
model=model,
config_path="config.yml"
)
At this stage, you should have:
Next: Move to Phase 6 to prepare for deployment.
import joblib
import json
from pathlib import Path
# Create artifacts directory
artifacts_dir = Path("artifacts/deployment")
artifacts_dir.mkdir(parents=True, exist_ok=True)
# Save model
joblib.dump(model, artifacts_dir / "model.pkl")
# Save configuration
with open(artifacts_dir / "config.json", "w") as f:
json.dump({
"sensitive_attributes": ["gender", "race"],
"fairness_metric": "demographic_parity_difference",
"validation_threshold": 0.05,
"min_group_size": 30,
"model_type": "reductions",
"constraint": "demographic_parity"
}, f, indent=2)
# Save feature names
with open(artifacts_dir / "feature_names.json", "w") as f:
json.dump(list(X_train.columns), f, indent=2)
# Save validation results
with open(artifacts_dir / "validation_results.json", "w") as f:
json.dump({
"baseline_metrics": {k: v.value for k, v in baseline_results.items()},
"final_metrics": {k: v.value for k, v in final_results.items()},
"validation_passed": validation_result.passed
}, f, indent=2)
# Model Deployment Documentation
## Model Information
- **Model Type**: ReductionsWrapper (Fairlearn)
- **Base Estimator**: LogisticRegression
- **Training Date**: 2024-01-15
- **Version**: 1.0.0
## Fairness Characteristics
- **Sensitive Attributes**: gender, race
- **Fairness Metric**: demographic_parity_difference
- **Validation Threshold**: 0.05
- **Final Metric Value**: 0.032
- **Validation Status**: ✅ PASSED
## Model Performance
- **Accuracy**: 0.85
- **AUC-ROC**: 0.91
- **Baseline Fairness**: 0.15 (demographic_parity_difference)
- **Final Fairness**: 0.032 (demographic_parity_difference)
- **Improvement**: 0.118
## Monitoring Requirements
- Track fairness metrics in production
- Alert if demographic_parity_difference > 0.05
- Minimum group size: 30 samples per group
Create monitoring configuration for production:
# monitoring_config.py
from fairpipe.monitoring import (
TrackerConfig,
DriftConfig,
MonitoringSettings
)
monitoring_config = MonitoringSettings(
tracker=TrackerConfig(
window_size=10_000, # Sliding window size
min_group_size=30, # Minimum samples per group
metrics=["demographic_parity_difference", "equalized_odds_difference"]
),
drift=DriftConfig(
alpha=0.05, # Significance level for drift detection
threshold=0.05, # Alert threshold for fairness metric
use_wavelets=False # Enable for multi-scale analysis
),
alerts={
"demographic_parity_difference": {
"threshold": 0.05,
"severity": "high"
}
}
)
# Save configuration
import json
with open("artifacts/deployment/monitoring_config.json", "w") as f:
json.dump(monitoring_config.dict(), f, indent=2)
# Deployment Checklist
## Pre-Deployment
- [ ] Model validation passed
- [ ] Fairness thresholds met
- [ ] Model artifacts saved
- [ ] Configuration documented
- [ ] Monitoring setup configured
- [ ] Stakeholder approval obtained
## Deployment
- [ ] Model deployed to staging
- [ ] Monitoring active
- [ ] Alerts configured
- [ ] Dashboard accessible
## Post-Deployment
- [ ] Baseline metrics established
- [ ] Monitoring verified
- [ ] Team trained on monitoring dashboard
- [ ] Runbook created
At this stage, you should have:
Next: Move to Phase 7 for production monitoring.
from fairpipe.monitoring import (
RealTimeFairnessTracker,
ColumnMap,
TrackerConfig
)
# Initialize tracker
tracker = RealTimeFairnessTracker(
config=TrackerConfig(
window_size=10_000, # Process last 10k samples
min_group_size=30, # Minimum samples per group
),
artifacts_dir="artifacts/monitoring"
)
# Define column mapping
cmap = ColumnMap(
y_pred="predictions",
y_true="labels", # Optional, if available
protected=["gender", "race"],
intersections=[["gender", "race"]] # Optional intersectional analysis
)
# In your production inference pipeline
def process_batch(predictions_df):
"""
Process a batch of predictions and update fairness tracking.
predictions_df should have columns:
- predictions: model predictions
- labels: true labels (if available)
- gender, race: sensitive attributes
"""
# Update tracker
tracker.process_batch(predictions_df, cmap)
# Get current metrics
current_metrics = tracker.get_current_metrics()
return current_metrics
from fairpipe.monitoring import (
FairnessDriftAndAlertEngine,
DriftConfig
)
# Initialize drift engine
drift_engine = FairnessDriftAndAlertEngine(
config=DriftConfig(
alpha=0.05,
threshold=0.05,
use_wavelets=False
)
)
# Analyze for drift (call periodically, e.g., daily)
alerts = drift_engine.analyze(tracker.metrics_ts)
# Check for alerts
if alerts:
print("⚠️ Fairness Alerts Detected:")
for alert in alerts:
print(f" - {alert['metric']}: {alert['value']:.4f} "
f"(threshold: {alert['threshold']:.4f})")
print(f" Severity: {alert['severity']}")
print(f" Message: {alert['message']}")
from fairpipe.monitoring import FairnessReportingDashboard
# Initialize dashboard
dashboard = FairnessReportingDashboard(
tracker=tracker,
artifacts_dir="artifacts/monitoring"
)
# Generate report
dashboard.generate_report(
output_path="artifacts/monitoring/report.md",
include_plots=True
)
Use the provided Streamlit or Dash apps:
Streamlit:
streamlit run apps/monitoring_streamlit_app.py
Dash:
python apps/monitoring_dash_app.py
Compare fairness between model versions:
from fairpipe.monitoring import FairnessABTestAnalyzer
# Compare two model versions
ab_analyzer = FairnessABTestAnalyzer()
results = ab_analyzer.compare(
metrics_a=tracker_a.metrics_ts,
metrics_b=tracker_b.metrics_ts,
metric="demographic_parity_difference"
)
print(f"A/B Test Results:")
print(f" Model A: {results['mean_a']:.4f}")
print(f" Model B: {results['mean_b']:.4f}")
print(f" Difference: {results['difference']:.4f}")
print(f" P-value: {results['p_value']:.4f}")
print(f" Significant: {results['significant']}")
At this stage, you should have:
For users who want to run the complete workflow in one command:
fairpipe run-pipeline \
--config config.yml \
--csv your_data.csv \
--output-dir artifacts/ \
--mlflow-experiment fairness_workflow
Create config.yml:
# Required: Sensitive attributes
sensitive: ["gender", "race"]
# Pipeline: Bias mitigation
pipeline:
- name: reweigh
transformer: "InstanceReweighting"
- name: repair
transformer: "DisparateImpactRemover"
params:
features: ["score", "age"]
sensitive: "gender"
repair_level: 0.8
# Training: Fairness-aware model
training:
method: "reductions"
target_column: "y"
params:
constraint: "demographic_parity"
eps: 0.01
# Validation: Threshold check
fairness_metric: "demographic_parity_difference"
validation_threshold: 0.05
StandardScaler (transform only), trains a fairness-constrained model, and predicts on the test setvalidation_thresholdexecute_workflow() supports runtime parameters that are not in YAML or fairpipe run-pipeline:
from fairpipe.integration import execute_workflow
from fairpipe.pipeline import load_config
from fairpipe import load_data
cfg = load_config("config.yml")
df = load_data("your_data.csv")
result = execute_workflow(
cfg,
df,
output_dir="artifacts/",
class_weight="balanced", # default; helps imbalanced hiring / fraud labels
decision_threshold=0.7, # optional selective classifier cutoff
random_state=42,
)
class_weight: Passed to baseline and reductions-path LogisticRegression (default "balanced").decision_threshold: If set, both steps threshold predict_proba at that value; if None, uses predict() (0.5).See demo.ipynb for a complete example.
Problem: Groups have fewer than 30 samples.
Solutions:
min_group_size parameter (with caution)Problem: Model doesn’t meet fairness threshold.
Solutions:
eps for reductions)Problem: Many features are correlated with sensitive attributes.
Solutions:
ProxyDropper transformerProblem: Fairness constraints reduced model performance.
Solutions:
demo.ipynb for complete examplestests/ for usage patternsREADME.md for module-specific documentationCHANGELOG.md for recent updatesImplement custom fairness metrics:
from fairpipe.metrics.base import BaseMetric
class CustomFairnessMetric(BaseMetric):
def compute(self, y_true, y_pred, sensitive_features):
# Your custom metric logic
return result
Create custom bias mitigation transformers:
from sklearn.base import BaseEstimator, TransformerMixin
class CustomTransformer(BaseEstimator, TransformerMixin):
def fit(self, X, y=None, sensitive_features=None):
# Fit logic
return self
def transform(self, X):
# Transform logic
return X_transformed
Combine multiple fairness objectives:
# Use Lagrangian trainer with multiple constraints
trainer = LagrangianFairnessTrainer(
model=model,
fairness="demographic_parity", # Primary constraint
# Can extend to support multiple constraints
...
)
Advanced MLflow logging:
import mlflow
from fairpipe.integration.mlflow_logger import log_workflow_results
with mlflow.start_run():
# Log custom parameters
mlflow.log_params({
"fairness_threshold": 0.05,
"training_method": "reductions",
"constraint": "demographic_parity"
})
# Log workflow results
log_workflow_results(
baseline_metrics=baseline_results,
final_metrics=final_results,
validation_result=validation_result,
model=model,
config_path="config.yml"
)
This guide has walked you through using the Fairness Pipeline Development Toolkit across the complete model development cycle. The toolkit provides:
fairpipe run-pipeline for a quick end-to-end exampledemo.ipynb for the full pipeline (baseline → transform+train → validate)Version: 0.9.1
Last Updated: 2026-05-22 (integrated workflow: class_weight, decision_threshold, StandardScaler)
For questions, issues, or contributions, please see CONTRIBUTING.md.