AI Ethics and Responsible AI Development

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import pandas as pd
from aequitas.group import Group

# Analyze for bias
df = pd.read_csv('predictions.csv')
g = Group()
xtab, _ = g.get_crosstabs(df, score_col='score', label_col='label')

# Check disparate impact
from aequitas.fairness import Fairness
f = Fairness()
fair = f.get_group_value_fairness(xtab)

1
2
3
4
5
6
7
8
9

# Demographic parity
def demographic_parity(y_true, y_pred, sensitive_feature):
    groups = np.unique(sensitive_feature)
    rates = []
    for group in groups:
        mask = sensitive_feature == group
        rate = y_pred[mask].mean()
        rates.append(rate)
    return np.std(rates)  # Lower is better

1
2
3
4

from diffprivlib.models import GaussianNB

clf = GaussianNB(epsilon=1.0)  # Privacy budget
clf.fit(X_train, y_train)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Train on distributed data without centralizing
class FederatedLearning:
    def __init__(self, clients):
        self.clients = clients
        self.global_model = Model()
    
    def train_round(self):
        client_models = []
        for client in self.clients:
            # Each client trains locally
            model = client.train(self.global_model)
            client_models.append(model)
        
        # Aggregate models
        self.global_model = self.aggregate(client_models)

1
2
3
4
5
6
7

import shap

explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# Visualize
shap.summary_plot(shap_values, X_test)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from lime.lime_tabular import LimeTabularExplainer

explainer = LimeTabularExplainer(
    X_train,
    feature_names=feature_names,
    class_names=['class0', 'class1']
)

explanation = explainer.explain_instance(
    X_test[0],
    model.predict_proba
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

Model Card:
  Model Details:
    name: Credit Risk Classifier
    version: 1.0
    date: 2024-03-27
  
  Intended Use:
    purpose: Assess credit risk
    users: Financial institutions
    out_of_scope: Not for employment decisions
  
  Training Data:
    size: 100,000 samples
    timeframe: 2020-2023
    demographics: Age 18-80, global
  
  Performance:
    overall_accuracy: 0.85
    fairness_metrics:
      demographic_parity: 0.02
  
  Limitations:
    - May not generalize to new markets
    - Requires annual retraining

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def filter_toxic_content(text):
    from transformers import pipeline
    
    classifier = pipeline("text-classification",
                         model="unitary/toxic-bert")
    result = classifier(text)
    
    if result[0]['score'] > 0.8:
        return "Content filtered"
    return text

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from functools import wraps
import time

def rate_limit(max_per_minute):
    def decorator(func):
        calls = []
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [c for c in calls if c > now - 60]
            
            if len(calls) >= max_per_minute:
                raise Exception("Rate limit exceeded")
            
            calls.append(now)
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

@rate_limit(max_per_minute=60)
def generate_text(prompt):
    return model.generate(prompt)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class AISystem:
    """
    Document system thoroughly:
    - Purpose and capabilities
    - Training data sources
    - Model architecture
    - Performance metrics
    - Known limitations
    - Ethical considerations
    - Update history
    """
    pass

1
2
3
4
5
6
7
8
9

# Test robustness
from art.attacks.evasion import FastGradientMethod
from art.estimators.classification import SklearnClassifier

classifier = SklearnClassifier(model=model)
attack = FastGradientMethod(estimator=classifier, eps=0.2)

x_test_adv = attack.generate(x=X_test)
accuracy_adv = model.score(x_test_adv, y_test)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Test edge cases
edge_cases = [
    {"input": "", "expected": "error"},
    {"input": "x" * 10000, "expected": "truncated"},
    {"input": "special_chars_!@#", "expected": "handled"}
]

for case in edge_cases:
    result = system.process(case["input"])
    assert result.status == case["expected"]