AI全栈测试工程指南：构建可信赖的智能系统

┌─────────────────────────────────────────┐
│   UI / API 层  (业务场景 + 安全测试)     │
├─────────────────────────────────────────┤
│   推理服务层  (延迟、吞吐、内存泄漏)      │
├─────────────────────────────────────────┤
│   模型层  (精度、校准、鲁棒性、公平性)    │
├─────────────────────────────────────────┤
│   特征工程层  (空值、分布漂移、异常值)    │
├─────────────────────────────────────────┤
│   数据管道层  (完整性、一致性、回填)      │
└─────────────────────────────────────────┘

import great_expectations as ge

# 假设从Kafka读取的原始数据为DataFrame
df = ge.dataset.PandasDataset(raw_data)

# 定义期望套件
expectations = [
    df.expect_column_values_to_not_be_null("user_id"),
    df.expect_column_values_to_be_between("age", 0, 120),
    df.expect_column_values_to_be_in_set("country", ["US", "CN", "EU", "Other"]),
    df.expect_column_pair_values_to_be_equal("order_amount", "taxed_amount", 
                                             parse_strings_as_datetimes=False)
]

# 运行检查并生成数据文档
validation_result = df.validate(result_format="SUMMARY")
assert validation_result["success"] is True, "数据质量门禁失败"

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """计算两个分布之间的PSI"""
    expected_percents, bin_edges = np.histogram(expected, bins=bins, density=True)
    actual_percents, _ = np.histogram(actual, bins=bin_edges, density=True)
    
    # 平滑处理，避免除以零
    expected_percents = np.clip(expected_percents, 1e-8, 1)
    actual_percents = np.clip(actual_percents, 1e-8, 1)
    
    psi = np.sum((actual_percents - expected_percents) * 
                 np.log(actual_percents / expected_percents))
    return psi

# 训练集特征分布作为基准
train_feature = train_data['credit_score']
daily_feature = today_data['credit_score']

psi_value = calculate_psi(train_feature, daily_feature)
assert psi_value < 0.1, f"特征漂移超过阈值，PSI={psi_value:.3f}"

from art.attacks.evasion import FastGradientMethod
from art.estimators.classification import TensorFlowV2Classifier
import tensorflow as tf

# 加载训练好的TF模型
model = tf.keras.models.load_model("resnet50_face.h5")
classifier = TensorFlowV2Classifier(model=model, 
                                    nb_classes=1000, 
                                    input_shape=(224, 224, 3),
                                    clip_values=(0, 255))

# 生成FGSM对抗样本（epsilon=0.05）
attack = FastGradientMethod(estimator=classifier, eps=0.05)
adversarial_batch = attack.generate(x_test_batch)

# 评估对抗精度
orig_preds = np.argmax(model.predict(x_test_batch), axis=1)
adv_preds = np.argmax(model.predict(adversarial_batch), axis=1)
robust_acc = np.mean(orig_preds == adv_preds)

# 门禁：对抗精度不得低于原始精度的80%
assert robust_acc >= 0.8 * original_acc, "鲁棒性不足，拒绝上线"

import numpy as np
from sklearn.calibration import calibration_curve

def compute_ece(y_true, y_prob, n_bins=10):
    """计算期望校准误差"""
    bin_boundaries = np.linspace(0, 1, n_bins + 1)
    ece = 0.0
    for i in range(n_bins):
        bin_low, bin_high = bin_boundaries[i], bin_boundaries[i+1]
        in_bin = (y_prob >= bin_low) & (y_prob < bin_high)
        if np.sum(in_bin) > 0:
            avg_acc = np.mean(y_true[in_bin])
            avg_conf = np.mean(y_prob[in_bin])
            ece += (np.sum(in_bin) / len(y_true)) * np.abs(avg_acc - avg_conf)
    return ece

# 二分类示例
ece_train = compute_ece(y_train, model.predict_proba(X_train)[:,1])
assert ece_train < 0.05, f"模型校准不佳，ECE={ece_train:.4f}"

# locustfile.py
from locust import HttpUser, task, between
import numpy as np

class ModelUser(HttpUser):
    wait_time = between(0.1, 0.5)
    
    @task
    def predict(self):
        # 构造随机特征向量（与训练一致）
        payload = {
            "features": np.random.randn(128).tolist()
        }
        with self.client.post("/v1/predict", json=payload, catch_response=True) as resp:
            if resp.status_code == 200:
                latency = resp.elapsed.total_seconds() * 1000  # ms
                # 自定义断言：P99必须 < 200ms（由Grafana聚合）
                if latency > 200:
                    resp.failure(f"高延迟: {latency:.2f}ms")
            else:
                resp.failure("推理失败")

from pynvml import *
nvmlInit()
handle = nvmlDeviceGetHandleByIndex(0)

def check_gpu_memory_leak(func, iterations=1000):
    mem_before = nvmlDeviceGetMemoryInfo(handle).used
    for _ in range(iterations):
        _ = func()
    mem_after = nvmlDeviceGetMemoryInfo(handle).used
    leak = (mem_after - mem_before) / (1024**2)  # MB
    assert leak < 100, f"显存泄漏 {leak:.2f} MB"

import requests
import json
from deepdiff import DeepDiff

# 从Kafka回放历史请求
for request_sample in replayed_traffic:
    old_resp = requests.post("http://old-model/predict", json=request_sample)
    new_resp = requests.post("http://new-model/predict", json=request_sample)
    
    diff = DeepDiff(old_resp.json(), new_resp.json(), ignore_order=True)
    if diff:
        # 只允许数值误差在 ±0.01 以内
        for key, value in diff.get('values_changed', {}).items():
            old_val = value['old_value']
            new_val = value['new_value']
            if isinstance(old_val, float) and isinstance(new_val, float):
                assert abs(old_val - new_val) < 0.01, f"输出差异过大: {key}"

import shap
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_sample)

# 业务规则：收入特征（索引5）的SHAP值必须为正
assert np.all(shap_values[:, 5] > -0.1), "收入特征的SHAP贡献出现负值，违反业务直觉"

pipeline {
    stages {
        stage('Data Validation') {
            steps { sh 'great_expectations --v3-api run checkpoint data_quality' }
        }
        stage('Model Evaluation') {
            steps { 
                sh 'python model_eval_suite.py --threshold-file thresholds.yaml'
                // 生成报告并上传至MLflow
                sh 'mlflow run . --entry-point evaluate'
            }
        }
        stage('Integration & Performance') {
            parallel {
                stage('Contract Test') { sh 'pytest tests/contract/' }
                stage('Load Test') { sh 'locust -f loadtest.py --headless -u 50' }
            }
        }
        stage('Canary Deployment') {
            // 金丝雀发布后，运行20分钟影子流量对比
            sh 'kubectl apply -f canary.yaml'
            sh 'python shadow_traffic_analyzer.py --duration 1200'
        }
    }
    post {
        failure {
            // 自动回滚并发送告警
            sh 'kubectl rollout undo deployment/model'
            slackSend(message: "🚨 AI模型上线失败，已回滚")
        }
    }
}