| name | AI Risk Assessment |
| description | Identifying, assessing, and mitigating risks in AI systems including bias, safety, privacy, security, and ethical concerns. |
AI Risk Assessment
Overview
AI Risk Assessment is the systematic process of identifying potential harms from AI systems, evaluating their likelihood and impact, and implementing mitigations. This is essential for responsible AI deployment and regulatory compliance.
Core Principle: "Identify risks before they become incidents. Prevention is cheaper than remediation."
1. Types of AI Risks
| Risk Category | Description | Example |
|---|
| Safety | Physical harm to people | Autonomous vehicle crash |
| Bias & Fairness | Discrimination against groups | Loan denial based on race |
| Privacy | Unauthorized data exposure | Model leaking training data |
| Security | Malicious attacks | Adversarial examples fooling classifier |
| Reliability | System failures | Model crashes on edge cases |
| Ethical | Harmful content/behavior | LLM generating hate speech |
| Reputational | Brand damage | AI making embarrassing mistakes publicly |
| Legal | Regulatory violations | GDPR non-compliance |
2. Risk Assessment Framework
Step 1: Identify Risks
class RiskIdentification:
"""Systematic risk identification"""
@staticmethod
def identify_risks(ai_system: dict) -> List[Risk]:
risks = []
if ai_system['domain'] in ['autonomous_vehicles', 'medical', 'industrial']:
risks.append(Risk(
category='safety',
description='Physical harm from incorrect decisions',
likelihood='medium',
impact='critical'
))
if ai_system['affects_people']:
risks.append(Risk(
category='bias',
description='Discrimination against protected groups',
likelihood='high',
impact='high'
))
if ai_system['uses_personal_data']:
risks.append(Risk(
category='privacy',
description='PII exposure or model inversion',
likelihood='medium',
impact='high'
))
risks.append(Risk(
category='security',
description='Adversarial attacks or prompt injection',
likelihood='medium',
impact='medium'
))
risks
Step 2: Assess Likelihood and Impact
class RiskMatrix:
"""Risk matrix for prioritization"""
LIKELIHOOD = {
'rare': 1,
'unlikely': 2,
'possible': 3,
'likely': 4,
'certain': 5
}
IMPACT = {
'negligible': 1,
'minor': 2,
'moderate': 3,
'major': 4,
'critical': 5
}
@staticmethod
def calculate_risk_score(likelihood: str, impact: str) -> int:
"""Calculate risk score (1-25)"""
return RiskMatrix.LIKELIHOOD[likelihood] * RiskMatrix.IMPACT[impact]
@staticmethod
def get_risk_level(score: int) -> str:
"""Categorize risk level"""
if score >= 15:
return 'critical'
elif score >= 10:
return 'high'
elif score >= 5:
return
:
Step 3: Prioritize Risks
def prioritize_risks(risks: List[Risk]) -> List[Risk]:
"""Prioritize risks by score"""
for risk in risks:
risk.score = RiskMatrix.calculate_risk_score(
risk.likelihood,
risk.impact
)
risk.level = RiskMatrix.get_risk_level(risk.score)
return sorted(risks, key=lambda r: r.score, reverse=True)
3. Bias and Fairness Assessment
Fairness Metrics
from fairlearn.metrics import MetricFrame, demographic_parity_difference, equalized_odds_difference
def assess_fairness(y_true, y_pred, sensitive_features):
"""Assess model fairness across groups"""
dp_diff = demographic_parity_difference(
y_true, y_pred, sensitive_features=sensitive_features
)
eo_diff = equalized_odds_difference(
y_true, y_pred, sensitive_features=sensitive_features
)
mf = MetricFrame(
metrics={'accuracy': accuracy_score, 'precision': precision_score},
y_true=y_true,
y_pred=y_pred,
sensitive_features=sensitive_features
)
fairness_risk = {
'demographic_parity_diff': dp_diff,
'equalized_odds_diff': eo_diff,
'per_group_metrics': mf.by_group.to_dict(),
'risk_level': 'high' if abs(dp_diff) > 0.1 or abs(eo_diff) > 0.1 else 'low'
}
return fairness_risk
Bias Testing Across Groups
def test_bias_across_demographics(model, test_data):
"""Test for bias across demographic groups"""
demographics = ['gender', 'race', 'age_group']
bias_report = {}
for demo in demographics:
groups = test_data[demo].unique()
for group in groups:
group_data = test_data[test_data[demo] == group]
predictions = model.predict(group_data)
accuracy = accuracy_score(group_data['label'], predictions)
approval_rate = predictions.mean()
bias_report[f"{demo}_{group}"] = {
'accuracy': accuracy,
'approval_rate': approval_rate,
'sample_size': len(group_data)
}
for demo in demographics:
groups = [k for k in bias_report.keys() if k.startswith(demo)]
approval_rates = [bias_report[g]['approval_rate'] for g in groups]
disparate_impact = min(approval_rates) / max(approval_rates)
if disparate_impact < 0.8:
logger.warning(
f"Disparate impact detected for {demo}: {disparate_impact:.2f}"
)
return bias_report
4. Safety Assessment
Failure Mode Analysis
class FailureModeAnalysis:
"""Identify what can go wrong"""
@staticmethod
def identify_failure_modes(ai_system: dict) -> List[FailureMode]:
failure_modes = []
failure_modes.append(FailureMode(
name='false_positive',
description='Model incorrectly predicts positive class',
consequence=ai_system.get('false_positive_consequence', 'User inconvenience'),
mitigation='Increase precision threshold'
))
failure_modes.append(FailureMode(
name='false_negative',
description='Model misses positive cases',
consequence=ai_system.get('false_negative_consequence', 'Missed opportunity'),
mitigation='Lower decision threshold'
))
failure_modes.append(FailureMode(
name='ood_input',
description='Input outside training distribution',
consequence='Unpredictable behavior',
mitigation='OOD detection and rejection'
))
failure_modes.append(FailureMode(
name='model_drift',
description='Model performance degrades over time',
consequence='Increasing error rate',
mitigation='Continuous monitoring and retraining'
))
return failure_modes
Red Teaming
def red_team_testing(model, test_cases):
"""Adversarial testing to find weaknesses"""
vulnerabilities = []
for test_case in test_cases['edge_cases']:
prediction = model.predict(test_case['input'])
if prediction != test_case['expected']:
vulnerabilities.append({
'type': 'edge_case_failure',
'input': test_case['input'],
'expected': test_case['expected'],
'actual': prediction
})
for adv_example in test_cases['adversarial']:
prediction = model.predict(adv_example['input'])
if prediction == adv_example['target_class']:
vulnerabilities.append({
'type': 'adversarial_success',
'input': adv_example['input'],
'fooled_into': prediction
})
return vulnerabilities
5. Privacy Risk Assessment
PII Exposure Risk
def assess_pii_exposure_risk(training_data, model):
"""Assess risk of PII leakage"""
risks = []
pii_fields = ['email', 'phone', 'ssn', 'name', 'address']
pii_found = [field for field in pii_fields if field in training_data.columns]
if pii_found:
risks.append({
'risk': 'pii_in_training_data',
'severity': 'high',
'fields': pii_found,
'mitigation': 'Remove or anonymize PII before training'
})
membership_attack_success = test_membership_inference(model, training_data)
if membership_attack_success > 0.6:
risks.append({
'risk': 'membership_inference_vulnerable',
'severity': 'medium',
'success_rate': membership_attack_success,
'mitigation': 'Apply differential privacy'
})
return risks
Model Inversion Attack Test
def test_model_inversion(model, target_features):
"""Test if model can be inverted to reconstruct training data"""
reconstructed = attempt_reconstruction(model, target_features)
similarity = calculate_similarity(reconstructed, target_features)
if similarity > 0.7:
return {
'vulnerable': True,
'similarity': similarity,
'risk_level': 'high',
'mitigation': 'Add noise to outputs or use differential privacy'
}
return {'vulnerable': False}
6. Security Risk Assessment
Adversarial Attack Vectors
class AdversarialRiskAssessment:
"""Assess vulnerability to adversarial attacks"""
@staticmethod
def test_adversarial_robustness(model, test_data, epsilon=0.1):
"""Test robustness against adversarial examples"""
from art.attacks.evasion import FastGradientMethod
from art.estimators.classification import SklearnClassifier
classifier = SklearnClassifier(model=model)
attack = FastGradientMethod(estimator=classifier, eps=epsilon)
x_adv = attack.generate(x=test_data)
original_accuracy = model.score(test_data, test_labels)
adversarial_accuracy = model.score(x_adv, test_labels)
robustness_score = adversarial_accuracy / original_accuracy
return {
'original_accuracy': original_accuracy,
'adversarial_accuracy': adversarial_accuracy,
'robustness_score': robustness_score,
'risk_level': 'high' if robustness_score < 0.7 else 'medium'
}
Prompt Injection Testing (LLMs)
def test_prompt_injection(llm, test_prompts):
"""Test LLM vulnerability to prompt injection"""
vulnerabilities = []
for prompt in test_prompts:
response = llm.generate(prompt['injection_attempt'])
if prompt['target_behavior'] in response.lower():
vulnerabilities.append({
'prompt': prompt['injection_attempt'],
'succeeded': True,
'response': response
})
vulnerability_rate = len(vulnerabilities) / len(test_prompts)
return {
'vulnerability_rate': vulnerability_rate,
'risk_level': 'critical' if vulnerability_rate > 0.3 else 'medium',
'examples': vulnerabilities[:5]
}
7. Ethical Risk Assessment
Harmful Content Generation
def assess_harmful_content_risk(llm):
"""Assess risk of generating harmful content"""
test_categories = [
'hate_speech',
'violence',
'illegal_activity',
'misinformation',
'self_harm'
]
risks = {}
for category in test_categories:
prompts = load_test_prompts(category)
harmful_responses = 0
for prompt in prompts:
response = llm.generate(prompt)
if is_harmful(response, category):
harmful_responses += 1
risk_rate = harmful_responses / len(prompts)
risks[category] = {
'risk_rate': risk_rate,
'severity': 'critical' if risk_rate > 0.1 else 'medium'
}
return risks
8. Risk Documentation
Risk Register
class RiskRegister:
"""Central registry of identified risks"""
def __init__(self):
self.risks = []
def add_risk(
self,
category: str,
description: str,
likelihood: str,
impact: str,
mitigation: str,
owner: str
):
risk = {
'id': generate_risk_id(),
'category': category,
'description': description,
'likelihood': likelihood,
'impact': impact,
'score': RiskMatrix.calculate_risk_score(likelihood, impact),
'level': RiskMatrix.get_risk_level(
RiskMatrix.calculate_risk_score(likelihood, impact)
),
'mitigation': mitigation,
'owner': owner,
'status': 'identified',
'created_at': datetime.now()
}
self.risks.append(risk)
return risk['id']
def get_critical_risks(self):
"""Get all critical risks"""
return [r for r in self.risks if r['level'] == 'critical']
def ():
pandas pd
df = pd.DataFrame(.risks)
df.to_csv(filename, index=)
Risk Matrix Visualization
import matplotlib.pyplot as plt
import numpy as np
def visualize_risk_matrix(risks):
"""Visualize risks on likelihood x impact matrix"""
fig, ax = plt.subplots(figsize=(10, 8))
for risk in risks:
x = RiskMatrix.LIKELIHOOD[risk['likelihood']]
y = RiskMatrix.IMPACT[risk['impact']]
color = {
'critical': 'red',
'high': 'orange',
'medium': 'yellow',
'low': 'green'
}[risk['level']]
ax.scatter(x, y, s=200, c=color, alpha=0.6)
ax.annotate(risk['category'], (x, y), fontsize=8)
ax.set_xlabel('Likelihood')
ax.set_ylabel('Impact')
ax.set_title('AI Risk Matrix')
ax.set_xticks(range(1, 6))
ax.set_yticks(range(1, 6))
ax.set_xticklabels(['Rare', 'Unlikely', 'Possible', 'Likely', 'Certain'])
ax.set_yticklabels(['Negligible', 'Minor', 'Moderate', 'Major', ])
ax.grid(, alpha=)
plt.tight_layout()
plt.savefig()
9. Risk Mitigation Strategies
Guardrails (Input/Output Filters)
class AIGuardrails:
"""Input and output filtering for safety"""
@staticmethod
def filter_input(user_input: str) -> dict:
"""Filter potentially harmful inputs"""
if contains_injection_pattern(user_input):
return {
'allowed': False,
'reason': 'Potential prompt injection detected'
}
if contains_pii(user_input):
return {
'allowed': False,
'reason': 'PII detected in input'
}
return {'allowed': True}
@staticmethod
def filter_output(ai_output: str) -> dict:
"""Filter potentially harmful outputs"""
if is_harmful(ai_output):
return {
'allowed': False,
'reason': 'Harmful content detected',
'filtered_output': '[Content filtered for safety]'
}
contains_pii(ai_output):
{
: ,
: ,
: redact_pii(ai_output)
}
{: , : ai_output}
10. Continuous Risk Monitoring
class RiskMonitoring:
"""Continuous monitoring of AI risks"""
@staticmethod
def monitor_model_performance():
"""Monitor for performance degradation"""
current_accuracy = calculate_current_accuracy()
baseline_accuracy = get_baseline_accuracy()
if current_accuracy < baseline_accuracy * 0.95:
alert(
severity='warning',
message=f'Model accuracy dropped to {current_accuracy:.1%}'
)
@staticmethod
def monitor_fairness_drift():
"""Monitor for fairness metric changes"""
current_fairness = calculate_fairness_metrics()
baseline_fairness = get_baseline_fairness()
for metric, value in current_fairness.items():
if abs(value - baseline_fairness[metric]) > 0.1:
alert(
severity='warning',
message=f'Fairness metric {metric} drifted: {value:.2f}'
)
11. AI Risk Assessment Checklist
Related Skills
44-ai-governance/model-bias-fairness
44-ai-governance/ai-data-privacy
44-ai-governance/model-risk-management
44-ai-governance/ai-ethics-compliance