AI鲁棒性：从稳定到可靠

# 典型的分布漂移现象
train_samples = ["这个产品很好用", "服务态度不错"]
online_samples = ["这货太拉了", "服了这波操作", "产品体验差评"]

# 不够彻底的边界处理
def preprocess(text):
    if not text:
        return ""
    text = text[:512]  # 简单截断，可能截断到关键信息
    return text

# 对异常标签缺乏容错
predictions = model.predict([0, 1, 2, 3, 4, 5, 6])  # 训练时没见过 6
# 可能抛出 IndexError 或者返回垃圾结果

import numpy as np
import random

def add_text_noise(text, noise_level=0.1):
    """给文本加噪声"""
    chars = list(text)
    for i in range(len(chars)):
        if random.random() < noise_level:
            chars[i] = random.choice(['!', '@', '#', '*', '~'])
    return ''.join(chars)

# 训练时构造多种异常样本
augmented_samples = []
for sample in original_samples:
    augmented_samples.append(add_text_noise(sample, 0.05))
    augmented_samples.append(add_text_noise(sample, 0.1))
    augmented_samples.append(sample[::-1])  # 反转

# 模拟不同输入风格
styles = [
    "正式风格", "口语化", "网络用语", "方言表达", "表情符号丰富"
]

def apply_style(text, style):
    """应用不同输入风格"""
    if style == "网络用语":
        text = text.replace("很好", "yyds")
        text = text.replace("不行", "拉胯")
    elif style == "表情符号丰富":
        text = text + "😊👍🎉"
    return text

def validate_input(text):
    """输入校验"""
    if not text or not isinstance(text, str):
        return False
    if len(text.strip()) == 0:
        return False
    if len(text) > 10000:  # 超长输入
        return False
    # 检查是否包含可疑内容
    suspicious_patterns = ['<script', 'javascript:', 'eval(']
    for pattern in suspicious_patterns:
        if pattern in text.lower():
            return False
    return True

def clean_input(text):
    """输入清洗"""
    # 移除 HTML 标签
    import re
    text = re.sub(r'<[^>]+>', '', text)
    # 规范化空白字符
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def safe_predict(model, inputs):
    """安全的预测封装"""
    try:
        if not isinstance(inputs, list):
            inputs = [inputs]

        # 过滤无效输入
        valid_inputs = []
        valid_indices = []
        for i, inp in enumerate(inputs):
            if validate_input(inp):
                valid_inputs.append(clean_input(inp))
                valid_indices.append(i)

        if not valid_inputs:
            return [None] * len(inputs)

        # 预测
        predictions = model.predict(valid_inputs)

        # 恢复原始顺序
        result = [None] * len(inputs)
        for i, idx in enumerate(valid_indices):
            result[idx] = predictions[i]

        return result
    except Exception as e:
        # 出错时返回默认值，而不是崩溃
        return [None] * len(inputs)

def predict_with_fallback(model, inputs):
    """带降级策略的预测"""
    try:
        # 先尝试主模型
        predictions = safe_predict(model, inputs)
        if all(p is not None for p in predictions):
            return predictions, "main_model"

        # 有失败，启用降级方案
        return fallback_predict(inputs), "fallback"

    except Exception as e:
        # 主模型完全失败，使用降级
        return fallback_predict(inputs), "emergency_fallback"

def fallback_predict(inputs):
    """降级预测（可以用更简单但更鲁棒的模型）"""
    # 这里用规则或者简化模型
    return [simple_rule_based(inp) for inp in inputs]

import tensorflow as tf

def huber_loss(y_true, y_pred, delta=1.0):
    """Huber Loss 对异常值更鲁棒"""
    error = y_true - y_pred
    abs_error = tf.abs(error)
    quadratic = tf.minimum(abs_error, delta)
    linear = abs_error - quadratic
    return tf.reduce_mean(0.5 * quadratic**2 + delta * linear)

# 使用 Label Smoothing 防止过拟合
def label_smoothing_loss(y_true, y_pred, smoothing=0.1):
    """Label Smoothing"""
    num_classes = tf.shape(y_pred)[-1]
    y_true_smoothed = y_true * (1 - smoothing) + smoothing / num_classes
    return tf.keras.losses.categorical_crossentropy(y_true_smoothed, y_pred)

import tensorflow as tf

def adversarial_training_step(model, x, y, epsilon=0.01):
    """对抗训练步骤"""
    with tf.GradientTape() as tape:
        tape.watch(x)
        predictions = model(x)
        loss = tf.keras.losses.sparse_categorical_crossentropy(y, predictions)

    # 计算梯度并生成对抗样本
    gradients = tape.gradient(loss, x)
    adversarial_x = x + epsilon * tf.sign(gradients)

    # 用对抗样本训练
    with tf.GradientTape() as tape:
        predictions = model(adversarial_x)
        loss = tf.keras.losses.sparse_categorical_crossentropy(y, predictions)

    gradients = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(gradients, model.trainable_variables))

class ModelMonitor:
    def __init__(self):
        self.prediction_count = 0
        self.error_count = 0
        self.confidence_history = []

    def log_prediction(self, confidence, is_error=False):
        """记录预测"""
        self.prediction_count += 1
        if is_error:
            self.error_count += 1
        self.confidence_history.append(confidence)

    def get_error_rate(self):
        """获取错误率"""
        if self.prediction_count == 0:
            return 0
        return self.error_count / self.prediction_count

    def get_avg_confidence(self):
        """获取平均置信度"""
        if not self.confidence_history:
            return 0
        return sum(self.confidence_history) / len(self.confidence_history)

class AnomalyDetector:
    def __init__(self, threshold=0.05):
        self.threshold = threshold
        self.normal_stats = {}

    def fit(self, normal_samples):
        """拟合正常样本的统计特征"""
        # 可以用简单统计，也可以用更复杂的方法
        self.normal_stats['length_mean'] = np.mean([len(s) for s in normal_samples])
        self.normal_stats['length_std'] = np.std([len(s) for s in normal_samples])

    def is_anomaly(self, sample):
        """检测是否异常"""
        # 基于长度异常检测
        length = len(sample)
        z_score = abs(length - self.normal_stats['length_mean']) / self.normal_stats['length_std']
        if z_score > 3:  # 3倍标准差之外认为是异常
            return True
        return False