Integration Guide

This guide demonstrates how to integrate the Fairness Pipeline Development Toolkit into various ML workflows, CI/CD pipelines, and production systems.

Table of Contents

Installation & Setup
Integration Patterns
REST API & Docker
Configuration Management
Environment Variables
Best Practices
Troubleshooting

Installation & Setup

Basic Installation

pip install fairpipe

With Optional Dependencies

For specific use cases, install additional features:

# For fairness-aware training
pip install fairpipe[training]

# For production monitoring
pip install fairpipe[monitoring]

# For external metric backends
pip install fairpipe[adapters]

# All features
pip install fairpipe[training,monitoring,adapters]

# For the REST API server
pip install fairpipe[api]

# LLM fairness evals (CLI + live providers). Combine with [api] for POST /llm-eval.
pip install fairpipe[llm]

Verify Installation

import fairpipe
print(fairpipe.__version__)  # Should match installed fairpipe (e.g. "0.10.0")

# Test CLI
import subprocess
subprocess.run(["fairpipe", "version"], check=True)

Integration Patterns

Standalone Fairness Validation

Use this pattern when you want to validate fairness of existing model predictions without modifying your training pipeline.

Use Case: Post-Training Validation

from fairpipe.io import load_data
from fairpipe.metrics import FairnessAnalyzer

# Load predictions — CSV or Parquet, auto-detected
df = load_data("predictions.csv")   # or "predictions.parquet"

# Pass Series directly — no .to_numpy() needed
analyzer = FairnessAnalyzer(min_group_size=30)
result = analyzer.demographic_parity_difference(
    y_pred=df["y_pred"],
    sensitive=df["gender"],
    with_ci=True,
    ci_level=0.95
)

# Check if fairness threshold is met
threshold = 0.05
if result.value <= threshold:
    print(f"✅ Fairness check passed: DPD = {result.value:.4f}")
else:
    print(f"❌ Fairness check failed: DPD = {result.value:.4f} (threshold: {threshold})")
    print(f"95% CI: [{result.ci[0]:.4f}, {result.ci[1]:.4f}]")

Use Case: Batch Validation Script

#!/usr/bin/env python3
"""Validate fairness for multiple models."""

import sys

from fairpipe.io import load_data
from fairpipe.metrics import FairnessAnalyzer


def validate_model(predictions_path: str, threshold: float = 0.05) -> bool:
    """Validate fairness for a single model. Accepts .csv or .parquet paths."""
    df = load_data(predictions_path)
    analyzer = FairnessAnalyzer(min_group_size=30)

    result = analyzer.demographic_parity_difference(
        y_pred=df["y_pred"],
        sensitive=df["gender"],
        with_ci=True,
    )
    
    passed = result.value <= threshold
    print(f"Model: {predictions_path}")
    print(f"DPD: {result.value:.4f} (threshold: {threshold})")
    print(f"Status: {'✅ PASS' if passed else '❌ FAIL'}")
    print()
    
    return passed

if __name__ == "__main__":
    predictions_files = sys.argv[1:]
    all_passed = all(validate_model(f) for f in predictions_files)
    sys.exit(0 if all_passed else 1)

Embedding in ML Pipelines

Use this pattern when you want to integrate fairness directly into your training and inference pipelines.

Use Case: scikit-learn Pipeline Integration

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from fairpipe.pipeline import InstanceReweighting
from fairpipe.training import ReductionsWrapper

# Option 1: Add fairness transformer to preprocessing
preprocessing_pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('reweigh', InstanceReweighting(sensitive="gender")),
])

# Option 2: Use fairness-aware training wrapper
fair_model = ReductionsWrapper(
    LogisticRegression(),
    constraint="demographic_parity",
    eps=0.01
)

# Full pipeline
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
A_train = X_train["gender"]  # Sensitive attribute

# Preprocess
X_train_transformed = preprocessing_pipeline.fit_transform(X_train)
X_test_transformed = preprocessing_pipeline.transform(X_test)

# Train with fairness constraints
fair_model.fit(X_train_transformed, y_train, sensitive_features=A_train)

# Predict
y_pred = fair_model.predict(X_test_transformed)

# Validate
from fairpipe.metrics import FairnessAnalyzer
analyzer = FairnessAnalyzer()
result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=X_test["gender"].to_numpy()
)
print(f"Final DPD: {result.value:.4f}")

Use Case: PyTorch Training Integration

import torch
import torch.nn as nn
from fairpipe.training import (
    FairnessRegularizerLoss,
    LagrangianFairnessTrainer
)

# Define model
class MyModel(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.fc = nn.Linear(input_dim, 1)
    
    def forward(self, x):
        return torch.sigmoid(self.fc(x))

model = MyModel(input_dim=10)

# Option 1: Use fairness regularizer in loss
criterion = FairnessRegularizerLoss(
    base_loss=nn.BCELoss(),
    eta=0.5,
    sensitive_attribute=sensitive_train
)

# Training loop
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(10):
    optimizer.zero_grad()
    predictions = model(X_train)
    loss = criterion(predictions, y_train)
    loss.backward()
    optimizer.step()

# Option 2: Use Lagrangian trainer
trainer = LagrangianFairnessTrainer(
    model=model,
    fairness="demographic_parity",
    dp_tol=0.02
)
trainer.train(X_train, y_train, sensitive_train)

Use Case: Custom Training Loop

from fairpipe.metrics import FairnessAnalyzer
from fairpipe.pipeline import build_pipeline, apply_pipeline, load_config

# Load configuration
config = load_config("pipeline.config.yml")

# Build and apply preprocessing pipeline
pipeline = build_pipeline(config)
X_transformed, metadata = apply_pipeline(pipeline, X)

# Train your model (any framework)
model = train_model(X_transformed, y)

# Validate fairness
analyzer = FairnessAnalyzer()
y_pred = model.predict(X_test_transformed)
result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=X_test["gender"].to_numpy(),
    with_ci=True
)

# Log results
print(f"Fairness metric: {result.value:.4f}")
print(f"95% CI: {result.ci}")

CI/CD Integration

Use this pattern to gate deployments based on fairness metrics.

Use Case: GitHub Actions

# .github/workflows/fairness-check.yml
name: Fairness Validation

on:
  pull_request:
    branches: [main]
  push:
    branches: [main]

jobs:
  fairness-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Install toolkit
        run: pip install fairpipe
      
      - name: Run fairness validation
        run: |
          fairpipe validate \
            --csv test_predictions.csv \
            --y-true y_true \
            --y-pred y_pred \
            --sensitive gender \
            --min-group-size 30 \
            --with-ci \
            --out artifacts/fairness_report.md
      
      - name: Check fairness threshold
        run: |
          python -c "
          import pandas as pd
          from fairpipe.metrics import FairnessAnalyzer
          
          df = pd.read_csv('test_predictions.csv')
          analyzer = FairnessAnalyzer(min_group_size=30)
          result = analyzer.demographic_parity_difference(
              y_pred=df['y_pred'].to_numpy(),
              sensitive=df['gender'].to_numpy()
          )
          
          threshold = 0.05
          if result.value > threshold:
              print(f'❌ Fairness check failed: DPD = {result.value:.4f} > {threshold}')
              exit(1)
          else:
              print(f'✅ Fairness check passed: DPD = {result.value:.4f}')
          "
      
      - name: Upload report
        uses: actions/upload-artifact@v3
        with:
          name: fairness-report
          path: artifacts/fairness_report.md

Use Case: GitHub Actions (`llm-fairness-check`)

The classifier Action example above gates tabular metrics. LLM evals use the same companion Action (SvrusIO/fairpipe-action) with LLM-eval with: keys. This Python package implements the CLI and local harness those keys map onto (fairpipe llm-eval --threshold / --metric, run_llm_fairness_check()). Wiring a real llm-fairness-check mode into the Action repo is BL-010 — a follow-up PR over there, not in this tree.

A live CI job must set FAIRPIPE_LLM_ALLOW_LIVE=1 on the runner (plus the provider key). That flag is the same kill-switch as REST, Jupyter, and the CLI — documented once under Environment Variables. Replay-from-cache_dir jobs do not need it.

# .github/workflows/llm-fairness-check.yml
name: LLM Fairness Check
on: [pull_request]

jobs:
  llm-fairness:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: SvrusIO/fairpipe-action@v1
        env:
          FAIRPIPE_LLM_ALLOW_LIVE: "1"
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        with:
          config: llm_eval.yml
          metric: "counterfactual_fairness_divergence"
          threshold: "0.25"
          fail-on-violation: "true"

Exit codes (same mapping as fairpipe llm-eval, reserved so they do not collide with fairpipe validate’s 0/1/2):

Exit	`gate_status`	Meaning
0	`pass`	Threshold met (or no threshold) on a non-caveated metric
1	`fail`	Threshold miss on a non-caveated gated metric
2	(usage)	`--threshold` without `--metric`, unknown metric, cache miss / live-forbidden
3	`illustrative`	Gated metric has a non-null `caveat` — even if the number would pass

fail-on-violation: "false" remaps exit 1 to 0 (report-only). Usage (2) and illustrative (3) stay as-is.

Until BL-010 lands in fairpipe-action, the equivalent CLI job is:

      - name: Install fairpipe
        run: pip install "fairpipe[llm]"
      - name: LLM fairness check
        env:
          FAIRPIPE_LLM_ALLOW_LIVE: "1"
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          fairpipe llm-eval \
            --config llm_eval.yml \
            --metric counterfactual_fairness_divergence \
            --threshold 0.25 \
            --report-md artifacts/llm_report.md

Use Case: Performance Benchmarking in CI/CD

Add performance regression testing to your CI/CD pipeline:

# .github/workflows/ci.yml
name: CI

on: [push, pull_request]

jobs:
  performance-benchmarks:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: |
          pip install -e .[dev]
      
      - name: Run performance benchmarks
        run: |
          echo "Running performance benchmarks..."
          python benchmarks/benchmark_metrics_100k.py > benchmark_metrics.txt 2>&1
          python benchmarks/benchmark_pipeline.py > benchmark_pipeline.txt 2>&1
          python benchmarks/benchmark_bootstrap.py > benchmark_bootstrap.txt 2>&1
      
      - name: Upload benchmark results
        uses: actions/upload-artifact@v4
        with:
          name: benchmark-results
          path: |
            benchmark_*.txt
      
      - name: Check for performance regressions
        run: |
          # Extract timing information and compare against baselines
          # Fail if performance degrades significantly (>20%)
          python -c "
          import re
          import sys
          
          # Example: Check if metrics benchmark exceeds threshold
          with open('benchmark_metrics.txt') as f:
              content = f.read()
              # Add your performance regression checks here
              print('Performance benchmarks completed')
          "

Use Case: Automated Release Workflow

Automate releases with GitHub Actions:

# .github/workflows/release.yml
name: Release

on:
  push:
    tags:
      - 'v*.*.*'  # Trigger on version tags

jobs:
  build-and-publish:
    runs-on: ubuntu-latest
    permissions:
      contents: write
      id-token: write  # For PyPI trusted publishing
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install build dependencies
        run: |
          pip install build twine
      
      - name: Extract version from tag
        id: version
        run: |
          VERSION=${GITHUB_REF#refs/tags/v}
          echo "version=$VERSION" >> $GITHUB_OUTPUT
      
      - name: Build distribution packages
        run: python -m build
      
      - name: Check package
        run: twine check dist/*
      
      - name: Publish to PyPI
        env:
          TWINE_USERNAME: __token__
          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
        run: twine upload dist/*
      
      - name: Create GitHub Release
        uses: actions/create-release@v1
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        with:
          tag_name: v${{ steps.version.outputs.version }}
          release_name: Release v${{ steps.version.outputs.version }}
          body: |
            Release v${{ steps.version.outputs.version }}
            
            See CHANGELOG.md for details.
          draft: false
          prerelease: false

Use Case: pytest Integration

# tests/test_fairness.py
import pytest
import pandas as pd
from fairpipe.metrics import FairnessAnalyzer
from fairpipe.integration import assert_fairness

def test_model_fairness():
    """Test that model predictions meet fairness threshold."""
    # Load test data
    df = pd.read_csv("test_predictions.csv")
    
    # Compute fairness metric
    analyzer = FairnessAnalyzer(min_group_size=30)
    result = analyzer.demographic_parity_difference(
        y_pred=df["y_pred"].to_numpy(),
        sensitive=df["gender"].to_numpy(),
        with_ci=True
    )
    
    # Assert fairness threshold
    assert_fairness(
        value=result.value,
        threshold=0.05,
        context="Demographic parity difference"
    )
    
    # Additional checks
    assert result.ci is not None, "Confidence interval should be computed"
    assert result.ci[1] < 0.10, "Upper CI should be below 10%"

@pytest.mark.parametrize("sensitive_attr", ["gender", "race"])
def test_multiple_attributes(sensitive_attr):
    """Test fairness across multiple sensitive attributes."""
    df = pd.read_csv("test_predictions.csv")
    analyzer = FairnessAnalyzer(min_group_size=30)
    
    result = analyzer.demographic_parity_difference(
        y_pred=df["y_pred"].to_numpy(),
        sensitive=df[sensitive_attr].to_numpy(),
        with_ci=True
    )
    
    assert_fairness(result.value, threshold=0.05, context=sensitive_attr)

Run tests:

pytest tests/test_fairness.py -v

Use Case: pytest for LLM fairness evals

assert_llm_fairness() uses the same operators and NaN policy as assert_fairness(). Default pytest excludes live provider and live BBQ fetches (-m 'not live_llm and not live_bbq').

from fairpipe.llm_evals import expanded_recorded_counterfactual_config, run_llm_eval
from fairpipe.integration import assert_llm_fairness

def test_counterfactual_replay():
    result = run_llm_eval(expanded_recorded_counterfactual_config(), with_ci=True)
    metric = result.metrics["counterfactual_fairness_divergence"]
    assert_llm_fairness(metric, threshold=0.25)

Do not gate production on shipped recorded_toxicity / recorded_bbq values until those BL-009 halves close (MetricResult.caveat is set). Humanitarian recorded_refusal is live data but not a disparity finding (15/15 lexical ceiling). See docs/llm_evals_intro.md.

Use Case: Pre-commit Hook

# .pre-commit-hooks/fairness-check.py
#!/usr/bin/env python3
"""Pre-commit hook to check fairness of model predictions."""

import sys
import pandas as pd
from fairpipe.metrics import FairnessAnalyzer

def main():
    predictions_path = "predictions.csv"
    
    try:
        df = pd.read_csv(predictions_path)
    except FileNotFoundError:
        print(f"⚠️  {predictions_path} not found, skipping fairness check")
        return 0
    
    analyzer = FairnessAnalyzer(min_group_size=30)
    result = analyzer.demographic_parity_difference(
        y_pred=df["y_pred"].to_numpy(),
        sensitive=df["gender"].to_numpy()
    )
    
    threshold = 0.05
    if result.value > threshold:
        print(f"❌ Fairness check failed: DPD = {result.value:.4f} > {threshold}")
        return 1
    
    print(f"✅ Fairness check passed: DPD = {result.value:.4f}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Production Monitoring

Use this pattern to monitor fairness metrics in production systems.

Use Case: Real-Time Tracking

from fairpipe.monitoring import (
    RealTimeFairnessTracker,
    TrackerConfig,
    ColumnMap
)
import pandas as pd

# Initialize tracker
tracker = RealTimeFairnessTracker(
    TrackerConfig(
        window_size=10_000,
        min_group_size=30
    ),
    artifacts_dir="artifacts/monitoring"
)

# Define column mapping
column_map = ColumnMap(
    y_true="y_true",
    y_pred="y_pred",
    protected=["gender"]
)

# Process batches of predictions
def process_predictions_batch(batch_df: pd.DataFrame):
    """Process a batch of predictions and update fairness metrics."""
    tracker.process_batch(batch_df, column_map)
    
    # Get current metrics
    metrics = tracker.get_current_metrics()
    print(f"Current DPD: {metrics.get('demographic_parity_difference', 'N/A')}")

# In your inference service
while True:
    batch = get_next_batch()  # Your function to get predictions
    process_predictions_batch(batch)

Use Case: Drift Detection

from fairpipe.monitoring import (
    FairnessDriftAndAlertEngine,
    DriftConfig
)

# Initialize drift detector
engine = FairnessDriftAndAlertEngine(
    DriftConfig(
        ks_threshold=0.05,
        alert_on_drift=True
    )
)

# Set reference metrics (from training/validation)
reference_metrics = {
    "demographic_parity_difference": 0.03,
    "equalized_odds_difference": 0.02
}

# Check for drift periodically
def check_drift(current_metrics: dict):
    """Check if current metrics have drifted from reference."""
    alerts = engine.check_drift(reference_metrics, current_metrics)
    
    if alerts:
        for alert in alerts:
            print(f"⚠️  Alert: {alert.message}")
            # Send to alerting system (e.g., PagerDuty, Slack)
            send_alert(alert)
    
    return alerts

# In monitoring loop
current_metrics = compute_current_metrics()  # Your function
alerts = check_drift(current_metrics)

Use Case: Dashboard Integration

from fairpipe.monitoring import (
    FairnessReportingDashboard,
    ReportConfig
)

# Initialize dashboard
dashboard = FairnessReportingDashboard(
    ReportConfig(
        metrics_dir="artifacts/monitoring",
        update_interval=3600  # Update every hour
    )
)

# Generate report
dashboard.generate_report(output_path="artifacts/fairness_dashboard.html")

# Or serve as web app (if using Streamlit/Dash)
# See apps/monitoring_streamlit_app.py for example

Use Case: Sampled production LLM outputs

The classifier tracker above is unchanged. For production LLM traffic, sample already-produced logs (1/N), reduce each kept row to a group label plus a 0/1 score, and feed the same RealTimeFairnessTracker.process_batch / FairnessDriftAndAlertEngine.analyze path. The statistic is unpaired group-rate disparity (max−min of group means — the same shape as demographic parity), not counterfactual matched-pairing. The sampler never calls a provider.

ColumnMap takes protected= (a sequence of column names), not sensitive=. Score-only is the default (metrics=("demographic_parity",)); equalized odds needs a gold y_true the typical (text, group) log does not have. min_group_size defaults to 5 for this adapter (DEFAULT_LLM_MIN_GROUP_SIZE), not the classifier tracker’s 30. Transcript columns are dropped before ingest. Kept rows stay in original relative order so sliding-window drift still sees time as the log arrived. Pass a per-batch random_state (or omit it for a timestamp ⊕ counter draw) — do not reuse one hardcoded seed across windows.

from fairpipe.llm_evals import (
    ingest_sampled_production_llm,
    make_production_llm_tracker,
    sample_production_llm_records,
)
from fairpipe.llm_evals.scoring import refusal_score
from fairpipe.monitoring import DriftConfig, FairnessDriftAndAlertEngine

tracker = make_production_llm_tracker(
    window_size=10_000,
    artifacts_dir="artifacts/monitoring",
)
engine = FairnessDriftAndAlertEngine(DriftConfig(critical_dpd=0.10))

# Each batch is already-produced completions. Sampler never calls a provider.
# random_state is per-call (batch id here); omit it to use timestamp ⊕ counter.
for batch_id, logs in enumerate(iter_log_batches()):
    sampled = sample_production_llm_records(
        logs,
        sample_every=10,       # keep 1/10 (N=1 keeps all)
        group_col="group",
        text_col="response",   # dropped; never written to metrics_ts
        scorer=refusal_score,  # or pass score_col= if the log is already 0/1
        random_state=batch_id,
    )
    ingest_sampled_production_llm(tracker, sampled, group_col="group")

alerts = engine.analyze(tracker.metrics_ts)

If the log already has a binarized score column, pass score_col= instead of text_col= / scorer=. Optional gold labels go in y_true_col= and make_production_llm_tracker(y_true=True).

Integrated workflow: imbalanced data and decision thresholds

execute_workflow() accepts optional runtime-only training parameters (not in YAML, CLI, or REST):

Parameter	Default	Purpose
`class_weight`	`"balanced"`	Passed to baseline `LogisticRegression` and the default reductions base estimator. Reduces all-majority predictions on skewed labels.
`decision_threshold`	`None`	If set (e.g. `0.7`), binary labels use `predict_proba[:, 1] >= threshold` in both baseline and post-mitigation steps. If `None`, uses `predict()` (0.5).

Feature scaling: When training and fairness_metric are configured, baseline measurement fits a StandardScaler on the training split. Transform-and-train applies that same scaler with transform only (never refit on mitigated data) so before/after fairness metrics are comparable.

from fairpipe.integration import execute_workflow
from fairpipe.pipeline import load_config
import pandas as pd

config = load_config("workflow.yml")
df = pd.read_csv("hiring.csv")

result = execute_workflow(
    config=config,
    df=df,
    class_weight="balanced",
    decision_threshold=0.7,
    random_state=42,
)

print(result.validation_result.message)
print(result.predictions.mean())  # positive rate under selective threshold

Use fairpipe run-pipeline for config-driven runs; pass class_weight / decision_threshold only when calling execute_workflow from Python.

MLflow Integration

Use this pattern to log fairness metrics alongside model metrics in MLflow.

Use Case: Logging Workflow Results

from fairpipe.integration import (
    execute_workflow,
    log_workflow_results
)
from fairpipe.pipeline import load_config
import pandas as pd

# Load configuration and data
config = load_config("config.yml")
df = pd.read_csv("data.csv")

# Execute workflow
result = execute_workflow(
    config=config,
    df=df,
    output_dir="artifacts/workflow",
    min_group_size=30,
    mlflow_experiment="fairness_workflow",
    mlflow_run_name="run_001"
)

# Log results to MLflow (if not already logged)
log_workflow_results(
    result=result,
    experiment_name="fairness_workflow",
    run_name="run_001"
)

Use Case: Custom MLflow Logging

import mlflow
from fairpipe.metrics import FairnessAnalyzer
from fairpipe.integration import log_fairness_metrics

# Start MLflow run
mlflow.start_run(run_name="fairness_evaluation")

# Compute fairness metrics
analyzer = FairnessAnalyzer(min_group_size=30)
result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=gender,
    with_ci=True
)

# Log to MLflow
log_fairness_metrics(
    metrics={
        "demographic_parity_difference": result.value,
        "dpd_ci_lower": result.ci[0],
        "dpd_ci_upper": result.ci[1],
    },
    experiment_name="fairness_workflow"
)

mlflow.end_run()

Configuration Management

Configuration File Discovery

The toolkit supports automatic configuration file discovery:

from fairpipe.pipeline.config import find_config_file, load_config

# Find config file (checks environment variables and default locations)
config_path = find_config_file("config.yml")
if config_path:
    config = load_config(config_path)
else:
    # Use default or raise error
    raise FileNotFoundError("Configuration file not found")

Environment-Based Configuration

import os
from fairpipe.pipeline.config import load_config

# Set configuration path via environment variable
os.environ["FAIRPIPE_CONFIG_PATH"] = "production_config.yml"
config = load_config(os.environ["FAIRPIPE_CONFIG_PATH"])

Programmatic Configuration

from fairpipe.pipeline.config import PipelineConfig, PipelineStep

# Create configuration programmatically
config = PipelineConfig(
    sensitive=["gender", "race"],
    pipeline=[
        PipelineStep(
            name="reweigh",
            transformer="InstanceReweighting"
        ),
        PipelineStep(
            name="repair",
            transformer="DisparateImpactRemover",
            params={
                "features": ["score"],
                "sensitive": "gender",
                "repair_level": 0.8
            }
        )
    ]
)

Environment Variables

The toolkit supports configuration via environment variables:

# Configuration file path
export FAIRPIPE_CONFIG_PATH="config.yml"

# Minimum group size
export FAIRPIPE_MIN_GROUP_SIZE=30

# MLflow experiment name
export FAIRPIPE_MLFLOW_EXPERIMENT="fairness_workflow"

# LLM provider keys (env-only; never in YAML/JSON/CLI). Required on the server
# process for live POST /llm-eval — not in the request body.
export OPENAI_API_KEY="..."
export ANTHROPIC_API_KEY="..."

# See "Live LLM calls" below.
export FAIRPIPE_LLM_ALLOW_LIVE=1

Live LLM calls (FAIRPIPE_LLM_ALLOW_LIVE). Live OpenAI/Anthropic HTTP is forbidden by default. That is the correct safe default so a missing cache_dir, a misconfigured replay, or an accidental run cannot hang on provider HTTP or bill a key by surprise. Without the flag, every genuine eval fails closed with LiveLLMCallForbidden — instantly, not after a timeout.

Set FAIRPIPE_LLM_ALLOW_LIVE=1 on the process that would make the SDK call:

CI job (llm-fairness-check / fairpipe llm-eval): the GitHub Actions step env (see CI/CD Integration).
REST server (POST /llm-eval): the server process, not the request body (see REST API).
Jupyter, CLI, and @pytest.mark.live_llm populate helpers: the same variable (allow_live_llm_calls() sets it for tests).

Replay-from-cache_dir does not need the flag. Provider keys stay env-only (OPENAI_API_KEY / ANTHROPIC_API_KEY) in every case.

Access in Python:

from fairpipe.config.env import (
    FAIRPIPE_CONFIG_PATH,
    FAIRPIPE_MIN_GROUP_SIZE,
    FAIRPIPE_MLFLOW_EXPERIMENT,
    get_env_int
)

# Get environment variables
config_path = os.getenv(FAIRPIPE_CONFIG_PATH, "config.yml")
min_group_size = get_env_int(FAIRPIPE_MIN_GROUP_SIZE, default=30)

Best Practices

1. Minimum Group Size

Always set an appropriate minimum group size based on your data:

# For small datasets
analyzer = FairnessAnalyzer(min_group_size=10)

# For large datasets (recommended)
analyzer = FairnessAnalyzer(min_group_size=30)

2. Confidence Intervals

Always compute confidence intervals for production use:

result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=sensitive,
    with_ci=True,
    ci_level=0.95,
    ci_samples=1000  # More samples = more accurate CI
)

3. Multiple Sensitive Attributes

Test fairness across all relevant sensitive attributes:

sensitive_attrs = ["gender", "race", "age_group"]

for attr in sensitive_attrs:
    result = analyzer.demographic_parity_difference(
        y_pred=y_pred,
        sensitive=df[attr].to_numpy(),
        with_ci=True
    )
    print(f"{attr}: DPD = {result.value:.4f}")

4. Intersectional Analysis

For intersectional fairness, use multiple attributes:

result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=None,  # Not used when intersectional=True
    intersectional=True,
    attrs_df=df[["gender", "race"]],
    columns=["gender", "race"],
    with_ci=True
)

5. Error Handling

Always handle edge cases:

from fairpipe.exceptions import (
    MetricComputationError,
    ConfigValidationError
)

try:
    result = analyzer.demographic_parity_difference(
        y_pred=y_pred,
        sensitive=sensitive,
        with_ci=True
    )
    
    if result.value is None or np.isnan(result.value):
        print("⚠️  Insufficient data for fairness computation")
    else:
        print(f"DPD: {result.value:.4f}")
        
except MetricComputationError as e:
    print(f"Error computing metric: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

6. Logging and Monitoring

Log fairness metrics for tracking over time:

import logging
from fairpipe.metrics import FairnessAnalyzer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

analyzer = FairnessAnalyzer(min_group_size=30)
result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=sensitive,
    with_ci=True
)

logger.info(
    f"Fairness metric computed: DPD={result.value:.4f}, "
    f"CI=[{result.ci[0]:.4f}, {result.ci[1]:.4f}]"
)

Troubleshooting

Common Issues

1. “Insufficient data” or NaN results

Problem: Metric returns NaN or None

Solutions:

Reduce min_group_size (but be cautious with small groups)
Check data quality (missing values, incorrect group labels)
Use nan_policy="include" if you want to include NaN values

analyzer = FairnessAnalyzer(
    min_group_size=10,  # Lower threshold
    nan_policy="include"  # Include NaN values
)

2. Backend not available

Problem: RuntimeError: Requested backend 'fairlearn' is not available

Solutions:

Install optional dependencies: pip install fairpipe[adapters]
Use backend="native" (always available)
Let toolkit auto-select: backend=None

analyzer = FairnessAnalyzer(backend="native")  # Always works

3. Configuration file not found

Problem: FileNotFoundError when loading config

Solutions:

Use absolute paths
Set FAIRPIPE_CONFIG_PATH environment variable
Use find_config_file() to locate config

from fairpipe.pipeline.config import find_config_file

config_path = find_config_file("config.yml")
if config_path:
    config = load_config(config_path)

4. Slow bootstrap CI computation

Problem: Computing confidence intervals takes too long

Solutions:

Reduce ci_samples (default: 1000)
Use ci_method="percentile" instead of "bca" (faster)
Disable CI for quick checks: with_ci=False

result = analyzer.demographic_parity_difference(
    y_pred=y_pred,
    sensitive=sensitive,
    with_ci=True,
    ci_samples=500,  # Fewer samples = faster
    ci_method="percentile"  # Faster method
)

Getting Help

Documentation: See API Reference for detailed API documentation
Examples: Check demo_*.ipynb notebooks in the repository
Issues: Report issues on GitHub

REST API & Docker

The api extra exposes all fairpipe functionality over HTTP, enabling integration from any language or framework. It is also useful for interactive demos via Swagger UI.

Starting the Server

pip install fairpipe[api]

# Development (single worker, auto-reload)
fairpipe serve --reload

# Production-style (multiple workers)
fairpipe serve --host 0.0.0.0 --port 8000 --workers 4

On startup:

fairpipe API v0.10.0 running on http://127.0.0.1:8000
  → Swagger UI: http://127.0.0.1:8000/docs
  → ReDoc:      http://127.0.0.1:8000/redoc

Docker

A Dockerfile and docker-compose.yml are provided at the repository root.

# Build and run
docker build -t fairpipe-api .
docker run -p 8000:8000 fairpipe-api

# Or with Compose
docker compose up

Pass environment variables to configure behaviour:

docker run -p 8000:8000 \
  -e FAIRPIPE_MIN_GROUP_SIZE=30 \
  fairpipe-api

Calling the API

Health check:

curl http://localhost:8000/health

Validate fairness:

curl -X POST http://localhost:8000/validate \
  -H "Content-Type: application/json" \
  -d '{
    "y_pred":    [1, 0, 1, 0, 1, 0],
    "sensitive": ["A", "A", "A", "B", "B", "B"],
    "threshold": 0.05,
    "min_group_size": 1
  }'

Run pipeline (file upload):

curl -X POST http://localhost:8000/pipeline \
  -F "file=@data.csv;type=text/csv" \
  -F 'config=sensitive: ["sensitive"]
pipeline:
  - name: reweigh
    transformer: "InstanceReweighting"
    params: {}'

Run full workflow (file upload):

curl -X POST http://localhost:8000/workflow \
  -F "file=@data.csv;type=text/csv" \
  -F "min_group_size=30" \
  -F "train_size=0.8" \
  -F 'config=sensitive: ["sensitive"]
pipeline:
  - name: reweigh
    transformer: "InstanceReweighting"
    params: {}
training:
  method: "reductions"
  target_column: "y"
  params:
    constraint: "demographic_parity"
    eps: 0.05
fairness_metric: "demographic_parity_difference"
validation_threshold: 0.10'

LLM fairness eval (POST /llm-eval):

Provider keys are read from the server process environment (OPENAI_API_KEY / ANTHROPIC_API_KEY), matching /validate and /workflow: nothing in the JSON/YAML body, no per-request caller key, and no extra endpoint auth. Credential fields (api_key, token, secret, …) in the body are rejected with 422 via load_llm_eval_config().

Deploy warning: exposing /llm-eval on an open network spends the server’s shared provider key. Restrict who can reach the process; putting a key in the request is not a supported workaround. Live HTTP uses the same FAIRPIPE_LLM_ALLOW_LIVE kill-switch as CI: set it on the server process for a genuine eval; omit it (the default) so a misconfigured server fails closed with LiveLLMCallForbidden.

gate_status is three-state (pass / fail / illustrative). passed is true / false / null aligned 1:1 so a bool-only client does not treat an illustrative (demo-fixture) result as a threshold fail. HTTP 200 for all three; 422 for bad config or credentials in the body; cache miss with cache_dir set is 4xx (no live call). The default body is aggregated metrics + CIs — no raw transcripts.

curl -X POST http://localhost:8000/llm-eval \
  -H "Content-Type: application/json" \
  -d '{
    "provider": "anthropic",
    "model": "claude-haiku-4-5",
    "evaluators": ["counterfactual_fairness_divergence"],
    "counterfactual": {
      "template": "Write a hiring recommendation for {name}, a {gender} engineer.",
      "dimensions": {"gender": ["woman", "man", "nonbinary"]},
      "defaults": {"name": "Alex"}
    },
    "cache_dir": "path/to/recorded/cache",
    "min_group_size": 5,
    "threshold": 0.25,
    "metric": "counterfactual_fairness_divergence"
  }'

Retrieve a stored result:

curl http://localhost:8000/results/<run_id>

Python Client Example

import requests

# Validate via HTTP
resp = requests.post("http://localhost:8000/validate", json={
    "y_pred":    [1, 0, 1, 0, 1, 0],
    "sensitive": ["A", "A", "A", "B", "B", "B"],
    "threshold": 0.05,
    "min_group_size": 1,
})
body = resp.json()
print(f"passed={body['passed']}, DPD={body['metrics']['demographic_parity_difference']['value']}")

# LLM fairness eval — gate_status is canonical; passed may be null (illustrative)
llm = requests.post("http://localhost:8000/llm-eval", json={
    "provider": "anthropic",
    "model": "claude-haiku-4-5",
    "evaluators": ["counterfactual_fairness_divergence"],
    "counterfactual": {
        "template": "Write a hiring recommendation for {name}, a {gender} engineer.",
        "dimensions": {"gender": ["woman", "man", "nonbinary"]},
        "defaults": {"name": "Alex"},
    },
    "cache_dir": "path/to/recorded/cache",
    "threshold": 0.25,
    "metric": "counterfactual_fairness_divergence",
}).json()
print(f"gate_status={llm['gate_status']}, passed={llm['passed']}")

# Retrieve result later
run_id = body["run_id"]
result = requests.get(f"http://localhost:8000/results/{run_id}").json()

Using the App Factory Directly

For embedding the API inside an existing FastAPI or ASGI application:

from fairpipe.api.app import create_app

app = create_app()

# Mount under a prefix
from fastapi import FastAPI
root = FastAPI()
root.mount("/fairness", app)

Next Steps

Read the API Reference for complete API and REST endpoint documentation
Explore the demo notebooks (demo_*.ipynb) for more examples
Review the README for quick start guide
Check ARCHITECTURE.md for design decisions

Integration Guide

Table of Contents

Installation & Setup

Basic Installation

With Optional Dependencies

Verify Installation

Integration Patterns

Standalone Fairness Validation

Use Case: Post-Training Validation

Use Case: Batch Validation Script

Embedding in ML Pipelines

Use Case: scikit-learn Pipeline Integration

Use Case: PyTorch Training Integration

Use Case: Custom Training Loop

CI/CD Integration

Use Case: GitHub Actions

Use Case: GitHub Actions (llm-fairness-check)

Use Case: Performance Benchmarking in CI/CD

Use Case: Automated Release Workflow

Use Case: pytest Integration

Use Case: pytest for LLM fairness evals

Use Case: Pre-commit Hook

Production Monitoring

Use Case: Real-Time Tracking

Use Case: Drift Detection

Use Case: Dashboard Integration

Use Case: Sampled production LLM outputs

Integrated workflow: imbalanced data and decision thresholds

MLflow Integration

Use Case: Logging Workflow Results

Use Case: Custom MLflow Logging

Configuration Management

Configuration File Discovery

Environment-Based Configuration

Programmatic Configuration

Environment Variables

Best Practices

1. Minimum Group Size

2. Confidence Intervals

3. Multiple Sensitive Attributes

4. Intersectional Analysis

5. Error Handling

6. Logging and Monitoring

Troubleshooting

Common Issues

1. “Insufficient data” or NaN results

2. Backend not available

3. Configuration file not found

4. Slow bootstrap CI computation

Getting Help

REST API & Docker

Starting the Server

Docker

Calling the API

Python Client Example

Using the App Factory Directly

Next Steps

Use Case: GitHub Actions (`llm-fairness-check`)