Phase 6: Capstone — Ship a Responsible AI Audit Pipeline · 50 min · Python · scikit-learn · Fairlearn
The Problem
Over the past 5 phases, you've built individual modules: a bias audit, SHAP/LIME explainability, a model card generator, a transparency report, a privacy impact assessment, an algorithmic impact assessment, an EU AI Act classifier, a risk register, an incident response playbook, and a monitoring dashboard. But these are scattered scripts — each runs independently, each produces its own output files.
In the real world, a RAI engineer runs one command and gets a complete audit package. This lesson integrates everything into a single, cohesive toolkit with a CLI interface.
The toolkit architecture
rai_audit_toolkit/
├── __init__.py
├── cli.py # CLI entry point
├── pipeline.py # Orchestrates the full audit
├── modules/
│ ├── bias_audit.py # Phase 2: Fairlearn bias audit
│ ├── explainability.py # Phase 3: SHAP + LIME explanations
│ ├── model_card.py # Phase 3: Model card generation
│ ├── privacy.py # Phase 4: PIA + membership inference
│ ├── governance.py # Phase 5: AIA + EU AI Act classifier
│ ├── risk_register.py # Phase 5: Risk register
│ ├── monitoring.py # Phase 5: Drift detection
│ └── report.py # Phase 3: Transparency report generator
└── templates/
└── report_template.html # Jinja2 template for final report
Building the pipeline orchestrator
"""
rai_audit_toolkit.py
Integrated Responsible AI audit toolkit — orchestrates all modules.
Run: python rai_audit_toolkit.py --model model.pkl --data data.csv --output audit/
"""
import json
import os
import argparse
import numpy as np
import pandas as pd
from datetime import datetime
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
class RAIAuditToolkit:
"""
Orchestrates a complete Responsible AI audit pipeline.
Integrates bias audit, explainability, model card, privacy,
governance, risk register, and transparency report generation.
"""
def __init__(self, output_dir="audit_output"):
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
self.results = {
"audit_date": datetime.now().isoformat(),
"modules_run": [],
"artifacts": [],
}
def run_bias_audit(self, model, X, y, sensitive_features):
"""Run bias audit using Fairlearn metrics."""
from sklearn.metrics import accuracy_score
y_pred = model.predict(X)
try:
from fairlearn.metrics import (
demographic_parity_difference,
equalized_odds_difference,
MetricFrame,
selection_rate,
)
fairness_metrics = {}
for attr_name, attr_values in sensitive_features.items():
dp = demographic_parity_difference(y, y_pred, sensitive_features=attr_values)
eo = equalized_odds_difference(y, y_pred, sensitive_features=attr_values)
# Disparate impact ratio
rates = MetricFrame(
metrics=selection_rate, y_true=y, y_pred=y_pred,
sensitive_features=attr_values
)
di_ratio = min(rates.by_group) / max(rates.by_group) if max(rates.by_group) > 0 else 0
fairness_metrics[attr_name] = {
"demographic_parity_difference": round(float(dp), 4),
"equalized_odds_difference": round(float(eo), 4),
"disparate_impact_ratio": round(float(di_ratio), 4),
"passes_80_percent_rule": di_ratio >= 0.80,
"passes_equalized_odds": eo < 0.05,
}
result = {
"module": "bias_audit",
"fairness_metrics": fairness_metrics,
"overall_accuracy": round(float(accuracy_score(y, y_pred)), 4),
}
except ImportError:
result = {
"module": "bias_audit",
"error": "Fairlearn not installed. Install with: pip install fairlearn",
"fairness_metrics": {},
}
self._save_result("bias_audit", result)
return result
def run_explainability(self, model, X_train, X_explain, feature_names):
"""Run SHAP explainability analysis."""
try:
import shap
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_explain)
# Handle binary classification (list of arrays)
if isinstance(shap_values, list):
shap_values_arr = shap_values[1] if len(shap_values) > 1 else shap_values[0]
else:
shap_values_arr = shap_values
global_importance = np.abs(shap_values_arr).mean(axis=0)
ranked = sorted(zip(feature_names, global_importance), key=lambda x: x[1], reverse=True)
result = {
"module": "explainability",
"global_feature_importance_ranked": [(f, round(float(v), 4)) for f, v in ranked],
"n_explanations": len(X_explain),
}
# Save SHAP plot
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
sorted_idx = np.argsort(global_importance)
plt.barh(range(len(feature_names)), global_importance[sorted_idx])
plt.yticks(range(len(feature_names)), [feature_names[i] for i in sorted_idx])
plt.xlabel("Mean |SHAP|")
plt.title("Global Feature Importance (SHAP)")
plt.tight_layout()
plt.savefig(os.path.join(self.output_dir, "shap_summary.png"), dpi=150)
plt.close()
except ImportError:
result = {
"module": "explainability",
"error": "SHAP not installed. Install with: pip install shap",
"global_feature_importance_ranked": [],
}
self._save_result("explainability", result)
return result
def run_model_card(self, model_info, training_info, evaluation_info,
fairness_results, explainability_results, ethical_considerations):
"""Generate a model card from audit results."""
from datetime import datetime
card = {
"model_details": {
"name": model_info.get("name", "Unknown"),
"version": model_info.get("version", "1.0"),
"owner": model_info.get("owner", "Unknown"),
"model_type": model_info.get("type", "Unknown"),
"date": datetime.now().strftime("%Y-%m-%d"),
},
"intended_use": {
"primary_use_cases": model_info.get("use_cases", []),
"out_of_scope_uses": model_info.get("out_of_scope", []),
},
"training_data": training_info,
"evaluation_data": evaluation_info,
"metrics": {
"accuracy": evaluation_info.get("accuracy"),
"fairness": fairness_results.get("fairness_metrics", {}),
},
"explainability": {
"global_feature_importance": explainability_results.get("global_feature_importance_ranked", [])[:10],
},
"ethical_considerations": ethical_considerations,
"generated_date": datetime.now().isoformat(),
}
self._save_result("model_card", card)
return card
def run_governance(self, system_name, system_purpose, aia_scores, eu_keywords):
"""Run AIA and EU AI Act classification."""
# Simplified AIA
total = sum(aia_scores.values())
max_score = len(aia_scores) * 4
score_pct = total / max_score if max_score > 0 else 0
impact_level = "Very High" if score_pct >= 0.75 else "High" if score_pct >= 0.5 else "Moderate" if score_pct >= 0.25 else "Low"
aia_result = {
"module": "algorithmic_impact_assessment",
"impact_level": impact_level,
"score": f"{total}/{max_score}",
}
# Simplified EU AI Act classification
high_risk_keywords = ["credit", "loan", "employment", "hiring", "education", "medical", "biometric", "law enforcement"]
is_high_risk = any(kw in eu_keywords.lower() for kw in high_risk_keywords)
eu_result = {
"module": "eu_ai_act_classification",
"risk_tier": "High-risk" if is_high_risk else "Minimal-risk",
"obligations_count": 10 if is_high_risk else 0,
}
self._save_result("aia", aia_result)
self._save_result("eu_ai_act", eu_result)
return {"aia": aia_result, "eu_ai_act": eu_result}
def run_risk_register(self, fairness_results, privacy_risk="medium"):
"""Generate a risk register from audit findings."""
risks = []
# Bias risk
for attr, fm in fairness_results.get("fairness_metrics", {}).items():
if not fm.get("passes_80_percent_rule"):
risks.append({
"id": f"RISK-{len(risks)+1:03d}",
"description": f"Disparate impact ratio {fm['disparate_impact_ratio']} for {attr} fails 80% rule",
"category": "Bias",
"severity": "high",
"likelihood": "high",
"mitigation": "Apply bias mitigation (reweighing or threshold optimization)",
"status": "open",
})
# Privacy risk
risks.append({
"id": f"RISK-{len(risks)+1:03d}",
"description": "Model may be vulnerable to membership inference attacks",
"category": "Privacy",
"severity": privacy_risk,
"likelihood": "medium",
"mitigation": "Model regularization, differential privacy for v2",
"status": "open",
})
result = {
"module": "risk_register",
"risks": risks,
"total_risks": len(risks),
}
self._save_result("risk_register", result)
return result
def generate_transparency_report(self, model_name, accuracy, fairness_results,
shap_results, model_card, governance_results,
risk_register):
"""Generate the final HTML transparency report."""
import base64
def encode_img(path):
try:
with open(path, "rb") as f:
return base64.b64encode(f.read()).decode()
except FileNotFoundError:
return None
shap_plot = encode_img(os.path.join(self.output_dir, "shap_summary.png"))
all_pass = all(
fm.get("passes_80_percent_rule") and fm.get("passes_equalized_odds")
for fm in fairness_results.get("fairness_metrics", {}).values()
)
html = f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8">
<title>RAI Audit Report — {model_name}</title>
<style>
body {{ font-family: -apple-system, sans-serif; max-width: 900px; margin: 0 auto; padding: 40px; }}
h1 {{ border-bottom: 3px solid #16213e; }} h2 {{ color: #16213e; margin-top: 40px; }}
table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
th {{ background: #16213e; color: white; }}
.pass {{ color: #28a745; font-weight: bold; }} .fail {{ color: #dc3545; font-weight: bold; }}
.summary {{ background: #f0f4ff; border-left: 4px solid #16213e; padding: 20px; margin: 20px 0; }}
.plot {{ text-align: center; margin: 20px 0; }} .plot img {{ max-width: 100%; }}
.badge {{ padding: 4px 12px; border-radius: 4px; font-weight: bold; }}
.badge-pass {{ background: #d4edda; color: #155724; }} .badge-fail {{ background: #f8d7da; color: #721c24; }}
</style></head><body>
<h1>Responsible AI Audit Report</h1>
<p><strong>Model:</strong> {model_name} | <strong>Date:</strong> {datetime.now().strftime('%Y-%m-%d')}</p>
<div class="summary">
<h2>Executive Summary</h2>
<p><strong>Overall:</strong> <span class="badge {'badge-pass' if all_pass else 'badge-fail'}">{'PASS' if all_pass else 'FAIL'}</span></p>
<p><strong>Accuracy:</strong> {accuracy:.2%} | <strong>Risks:</strong> {risk_register['total_risks']} |
<strong>AIA:</strong> {governance_results['aia']['impact_level']} | <strong>EU AI Act:</strong> {governance_results['eu_ai_act']['risk_tier']}</p>
</div>
<h2>1. Bias Audit</h2>
<table><tr><th>Attribute</th><th>DI Ratio</th><th>EO Diff</th><th>80% Rule</th><th>EO Pass</th></tr>
"""
for attr, fm in fairness_results.get("fairness_metrics", {}).items():
html += f"<tr><td>{attr}</td><td>{fm['disparate_impact_ratio']}</td><td>{fm['equalized_odds_difference']}</td>"
html += f"<td class=\"{'pass' if fm['passes_80_percent_rule'] else 'fail'}\">{'PASS' if fm['passes_80_percent_rule'] else 'FAIL'}</td>"
html += f"<td class=\"{'pass' if fm['passes_equalized_odds'] else 'fail'}\">{'PASS' if fm['passes_equalized_odds'] else 'FAIL'}</td></tr>"
html += "</table>\n<h2>2. Explainability (SHAP)</h2>\n<table><tr><th>Rank</th><th>Feature</th><th>Mean |SHAP|</th></tr>\n"
for i, (feat, val) in enumerate(shap_results.get("global_feature_importance_ranked", [])[:10], 1):
html += f"<tr><td>{i}</td><td>{feat}</td><td>{val}</td></tr>"
html += "</table>\n"
if shap_plot:
html += f'<div class="plot"><img src="data:image/png;base64,{shap_plot}" alt="SHAP Summary"></div>\n'
html += f"""
<h2>3. Governance</h2>
<table><tr><th>Assessment</th><th>Result</th></tr>
<tr><td>Algorithmic Impact Assessment</td><td>{governance_results['aia']['impact_level']} ({governance_results['aia']['score']})</td></tr>
<tr><td>EU AI Act Classification</td><td>{governance_results['eu_ai_act']['risk_tier']}</td></tr>
</table>
<h2>4. Risk Register</h2>
<table><tr><th>ID</th><th>Description</th><th>Category</th><th>Severity</th><th>Status</th></tr>
"""
for risk in risk_register.get("risks", []):
html += f"<tr><td>{risk['id']}</td><td>{risk['description']}</td><td>{risk['category']}</td><td>{risk['severity']}</td><td>{risk['status']}</td></tr>"
html += f"""
</table>
<h2>5. Framework Compliance</h2>
<table><tr><th>Framework</th><th>Control</th><th>Evidence</th></tr>
<tr><td>NIST AI RMF</td><td>MEASURE 2.7 — Bias testing</td><td class="pass">Section 1</td></tr>
<tr><td>EU AI Act</td><td>Art. 10 — Data governance</td><td class="pass">Section 1</td></tr>
<tr><td>EU AI Act</td><td>Art. 13 — Transparency</td><td class="pass">Section 2 & 3</td></tr>
<tr><td>EU AI Act</td><td>Annex IV — Technical documentation</td><td class="pass">Model card</td></tr>
<tr><td>ISO 42001</td><td>Clause 9.1 — Monitoring</td><td class="pass">This report</td></tr>
</table>
<p style="margin-top:60px;color:#666;font-size:0.9em;">Generated by RAI Audit Toolkit | {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
</body></html>"""
report_path = os.path.join(self.output_dir, "audit_report.html")
with open(report_path, "w") as f:
f.write(html)
self.results["artifacts"].append("audit_report.html")
return report_path
def _save_result(self, name, result):
path = os.path.join(self.output_dir, f"{name}.json")
with open(path, "w") as f:
json.dump(result, f, indent=2)
self.results["modules_run"].append(name)
self.results["artifacts"].append(f"{name}.json")
def get_summary(self):
return self.results
if __name__ == "__main__":
# Demo: Run the complete audit pipeline
np.random.seed(42)
n = 3000
# Generate synthetic loan data
data = pd.DataFrame({
"income": np.random.normal(55000, 20000, n).clip(15000, 200000),
"credit_score": np.random.normal(700, 80, n).clip(300, 850),
"debt_ratio": np.random.uniform(0.1, 0.5, n),
"employment_years": np.random.exponential(5, n).clip(0, 40),
})
# Create a sensitive attribute with bias
race = np.random.choice(["Group_A", "Group_B"], n, p=[0.6, 0.4])
# Introduce bias: Group_B has systematically lower approval rates
base_prob = (data["credit_score"] > 650) & (data["debt_ratio"] < 0.4)
bias_mask = (race == "Group_B") & (np.random.random(n) < 0.15)
data["approved"] = (base_prob & ~bias_mask).astype(int)
data["race"] = race
features = ["income", "credit_score", "debt_ratio", "employment_years"]
X = data[features]
y = data["approved"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# Train model
model = RandomForestClassifier(n_estimators=100, max_depth=8, random_state=42)
model.fit(X_train, y_train)
# Run the audit
toolkit = RAIAuditToolkit(output_dir="audit_output")
print("1. Running bias audit...")
bias = toolkit.run_bias_audit(model, X_test, y_test, {"race": data.loc[X_test.index, "race"]})
print("2. Running explainability...")
explain = toolkit.run_explainability(model, X_train, X_test[:100], features)
print("3. Generating model card...")
card = toolkit.run_model_card(
model_info={"name": "LoanApproval-RF-v1", "version": "1.0", "owner": "RAI Team", "type": "RandomForest"},
training_info={"description": "Synthetic loan data", "size": str(len(X_train)), "features": features},
evaluation_info={"description": "30% test split", "accuracy": float(model.score(X_test, y_test))},
fairness_results=bias,
explainability_results=explain,
ethical_considerations={"bias_findings": ["DI ratio below 80% rule for race"]},
)
print("4. Running governance assessment...")
governance = toolkit.run_governance(
"LoanApproval-RF-v1", "Automated loan approval",
{"decision_making": 3, "affected_population": 3, "impact_scale": 3, "impact_duration": 3, "rights_impact": 3},
"credit scoring loan approval",
)
print("5. Building risk register...")
risk_reg = toolkit.run_risk_register(bias)
print("6. Generating transparency report...")
report_path = toolkit.generate_transparency_report(
"LoanApproval-RF-v1", float(model.score(X_test, y_test)),
bias, explain, card, governance, risk_reg,
)
summary = toolkit.get_summary()
print(f"\n=== Audit Complete ===")
print(f"Modules run: {', '.join(summary['modules_run'])}")
print(f"Artifacts: {', '.join(summary['artifacts'])}")
print(f"Report: {report_path}")
The RAIAuditToolkit.run_bias_audit and run_explainability methods both wrap their core logic in a try/except ImportError block that returns an error dict instead of crashing when Fairlearn or SHAP isn't installed. A teammate suggests removing these try/except blocks "to keep the code simple, since we always install these dependencies anyway." Should you agree?
The whole point of an integrated toolkit is that it needs to survive running in environments you don't fully control — a teammate's fresh virtual environment, a CI runner with a minimal install, or a production container that only has the dependencies someone remembered to add to requirements.txt. If run_explainability crashed the whole script the moment it hit a missing import shap, you would lose the bias audit results, the model card, and everything else that had already run successfully before that point — even though those parts had nothing to do with SHAP. Returning a structured error dict for the missing dependency, exactly like explain_lime.py does in Phase 3, lets the rest of the pipeline keep running and gives whoever's running it a clear next step ("pip install shap") instead of a stack trace. This is standard defensive engineering for any tool meant to be run by other people, not just its author.