多层感知机(MLP)：从数学原理到实战应用

import numpy as np
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

# 加载数据
iris = load_iris()
X = iris.data
y = iris.target

# 数据标准化 - 对模型训练非常重要
scaler = StandardScaler()
X = scaler.fit_transform(X)

# 将标签转换为独热编码形式
y_onehot = np.eye(3)[y]

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y_onehot, test_size=0.2, random_state=42)

class MLP:
    def __init__(self, input_size, hidden_size, output_size):
        # 初始化权重和偏置
        self.W1 = np.random.randn(input_size, hidden_size) * 0.01
        self.b1 = np.zeros((1, hidden_size))
        self.W2 = np.random.randn(hidden_size, output_size) * 0.01
        self.b2 = np.zeros((1, output_size))
        
    def relu(self, x):
        return np.maximum(0, x)
    
    def relu_derivative(self, x):
        return np.where(x > 0, 1, 0)
    
    def softmax(self, x):
        exp_x = np.exp(x - np.max(x, axis=1, keepdims=True))  # 防止数值溢出
        return exp_x / np.sum(exp_x, axis=1, keepdims=True)
    
    def forward(self, X):
        # 前向传播
        self.z1 = np.dot(X, self.W1) + self.b1
        self.a1 = self.relu(self.z1)
        self.z2 = np.dot(self.a1, self.W2) + self.b2
        self.a2 = self.softmax(self.z2)
        return self.a2
    
    def backward(self, X, y, learning_rate):
        # 反向传播
        m = X.shape[0]
        
        # 输出层误差
        dz2 = self.a2 - y
        dW2 = (1/m) * np.dot(self.a1.T, dz2)
        db2 = (1/m) * np.sum(dz2, axis=0, keepdims=True)
        
        # 隐藏层误差
        dz1 = np.dot(dz2, self.W2.T) * self.relu_derivative(self.z1)
        dW1 = (1/m) * np.dot(X.T, dz1)
        db1 = (1/m) * np.sum(dz1, axis=0, keepdims=True)
        
        # 更新参数
        self.W2 -= learning_rate * dW2
        self.b2 -= learning_rate * db2
        self.W1 -= learning_rate * dW1
        self.b1 -= learning_rate * db1

# 定义交叉熵损失函数
def cross_entropy_loss(y_pred, y_true):
    m = y_true.shape[0]
    epsilon = 1e-8  # 防止对数中的零值
    y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
    log_likelihood = -np.sum(y_true * np.log(y_pred)) / m
    return log_likelihood

# 创建MLP
mlp = MLP(input_size=4, hidden_size=10, output_size=3)

# 训练参数
epochs = 5000
lr = 0.01

# 训练循环
for epoch in range(epochs):
    # 前向传播
    output = mlp.forward(X_train)
    
    # 计算损失 (交叉熵)
    loss = cross_entropy_loss(output, y_train)
    
    # 反向传播
    mlp.backward(X_train, y_train, lr)
    
    # 每500轮打印损失
    if epoch % 500 == 0:
        print(f"Epoch {epoch}, Loss: {loss:.4f}")

# 评估
test_output = mlp.forward(X_test)
predictions = np.argmax(test_output, axis=1)
true_labels = np.argmax(y_test, axis=1)
accuracy = np.mean(predictions == true_labels)
print(f"Test Accuracy: {accuracy*100:.2f}%")

import numpy as np
from sklearn.datasets import fetch_california_housing  # 注：load_boston已被弃用，使用california_housing替代
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

# 加载数据
housing = fetch_california_housing()
X = housing.data
y = housing.target.reshape(-1, 1)

# 标准化特征和目标值
X_scaler = StandardScaler()
y_scaler = StandardScaler()
X = X_scaler.fit_transform(X)
y = y_scaler.fit_transform(y)

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
y_test_original = y_scaler.inverse_transform(y_test)  # 保存原始比例的测试集标签，用于后续评估

class RegressionMLP(MLP):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__(input_size, hidden_size, output_size)
    
    def forward(self, X):
        # 前向传播
        self.z1 = np.dot(X, self.W1) + self.b1
        self.a1 = self.relu(self.z1)
        self.z2 = np.dot(self.a1, self.W2) + self.b2
        return self.z2  # 线性输出，不使用激活函数
    
    def backward(self, X, y, learning_rate):
        # 反向传播
        m = X.shape[0]
        
        # 输出层误差 (使用MSE损失)
        dz2 = (self.z2 - y) / m
        dW2 = np.dot(self.a1.T, dz2)
        db2 = np.sum(dz2, axis=0, keepdims=True)
        
        # 隐藏层误差
        dz1 = np.dot(dz2, self.W2.T) * self.relu_derivative(self.z1)
        dW1 = np.dot(X.T, dz1)
        db1 = np.sum(dz1, axis=0, keepdims=True)
        
        # 更新参数
        self.W2 -= learning_rate * dW2
        self.b2 -= learning_rate * db2
        self.W1 -= learning_rate * dW1
        self.b1 -= learning_rate * db1

# 训练模型
mlp = RegressionMLP(input_size=X.shape[1], hidden_size=20, output_size=1)

# 训练参数
epochs = 10000
lr = 0.001

# 训练循环
for epoch in range(epochs):
    # 前向传播
    pred = mlp.forward(X_train)
    
    # 计算损失 (MSE)
    loss = np.mean((pred - y_train)**2)
    
    # 反向传播
    mlp.backward(X_train, y_train, lr)
    
    # 每1000轮打印损失
    if epoch % 1000 == 0:
        print(f"Epoch {epoch}, MSE Loss: {loss:.4f}")

# 预测测试集
test_pred = mlp.forward(X_test)

# 将预测结果转换回原始比例
test_pred = y_scaler.inverse_transform(test_pred)

# 评估模型性能
from sklearn.metrics import r2_score, mean_squared_error
r2 = r2_score(y_test_original, test_pred)
rmse = np.sqrt(mean_squared_error(y_test_original, test_pred))
print(f"R-squared: {r2:.4f}")
print(f"RMSE: {rmse:.4f}")

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.metrics import AUC

# 构建处理不平衡数据的MLP模型
def build_fraud_detection_model(input_dim=30):
    model = Sequential([
        Dense(64, input_dim=input_dim, activation='relu'),
        Dropout(0.5),  # 防止过拟合
        Dense(32, activation='relu'),
        Dropout(0.3),
        Dense(1, activation='sigmoid')  # 二分类问题
    ])
    
    # 焦点损失函数 - 专为处理不平衡数据设计
    def focal_loss(y_true, y_pred, alpha=0.25, gamma=2):
        """
        焦点损失：对难以分类的样本给予更高的权重
        - alpha: 平衡正负样本的权重
        - gamma: 调整难易样本的权重
        """
        pt = tf.where(tf.equal(y_true, 1), y_pred, 1 - y_pred)
        focal_weight = alpha * tf.pow(1 - pt, gamma)
        return -focal_weight * tf.math.log(pt)
    
    # 编译模型
    model.compile(
        optimizer=Adam(learning_rate=0.001),
        loss=focal_loss,
        metrics=[AUC()]  # AUC是不平衡数据的良好评估指标
    )
    
    return model

# 模型训练示例
# model = build_fraud_detection_model()
# history = model.fit(
#     X_train, y_train,
#     validation_data=(X_val, y_val),
#     class_weight={0: 1, 1: 50},  # 类别权重
#     batch_size=256,
#     epochs=20
# )

import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Flatten, Concatenate, Dense

def build_recommendation_model(num_users, num_items, embedding_size=16):
    # 输入层
    user_input = Input(shape=(1,), name='user_input')
    item_input = Input(shape=(1,), name='item_input')
    
    # 嵌入层 - 将用户和物品ID转换为密集向量
    user_embedding = Embedding(
        input_dim=num_users + 1,  # +1 用于处理未知用户
        output_dim=embedding_size,
        name='user_embedding'
    )(user_input)
    
    item_embedding = Embedding(
        input_dim=num_items + 1,  # +1 用于处理未知物品
        output_dim=embedding_size,
        name='item_embedding'
    )(item_input)
    
    # 展平嵌入向量
    user_vector = Flatten()(user_embedding)
    item_vector = Flatten()(item_embedding)
    
    # 合并特征
    merged = Concatenate()([user_vector, item_vector])
    
    # MLP层 - 学习用户-物品交互
    x = Dense(128, activation='relu')(merged)
    x = Dense(64, activation='relu')(x)
    x = Dense(32, activation='relu')(x)
    
    # 输出层 - 预测评分或点击概率
    output = Dense(1, activation='linear')(x)
    
    # 构建模型
    model = Model(inputs=[user_input, item_input], outputs=output)
    model.compile(optimizer='adam', loss='mean_squared_error')
    
    return model

# 模型使用示例
# model = build_recommendation_model(num_users=10000, num_items=5000)
# model.fit([user_ids, item_ids], ratings, epochs=10, batch_size=64)

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from tensorflow.keras.initializers import GlorotUniform, HeNormal, LecunNormal

# Xavier/Glorot初始化
model_xavier = Sequential([
    Dense(128, input_dim=20, activation='tanh', 
          kernel_initializer=GlorotUniform()),
    Dense(64, activation='tanh', 
          kernel_initializer=GlorotUniform()),
    Dense(1, activation='sigmoid')
])

# He初始化
model_he = Sequential([
    Dense(128, input_dim=20, activation='relu', 
          kernel_initializer=HeNormal()),
    Dense(64, activation='relu', 
          kernel_initializer=HeNormal()),
    Dense(1, activation='sigmoid')
])

# LeCun初始化
model_lecun = Sequential([
    Dense(128, input_dim=20, activation='selu', 
          kernel_initializer=LecunNormal()),
    Dense(64, activation='selu', 
          kernel_initializer=LecunNormal()),
    Dense(1, activation='sigmoid')
])

from tensorflow.keras.layers import BatchNormalization, Dropout
from tensorflow.keras.regularizers import l1, l2, l1_l2

def build_regularized_model(input_dim, output_dim):
    model = Sequential([
        # L2权重正则化 - 惩罚大的权重值
        Dense(128, input_dim=input_dim, activation='relu',
              kernel_regularizer=l2(0.01)),
              
        # Dropout - 随机关闭一定比例的神经元
        Dropout(0.3),
        
        # 批归一化 - 标准化每一层的输入
        BatchNormalization(),
        
        # 组合L1和L2正则化（弹性网络）
        Dense(64, activation='relu',
              kernel_regularizer=l1_l2(l1=0.001, l2=0.01)),
        
        Dropout(0.2),
        
        Dense(output_dim, activation='softmax')
    ])
    
    return model

import numpy as np
import matplotlib.pyplot as plt
from tensorflow.keras.callbacks import LearningRateScheduler

# 学习率阶梯衰减
def step_decay(epoch, initial_lr=0.01, drop_factor=0.5, epochs_drop=10):
    """每隔epochs_drop轮将学习率降低为原来的drop_factor倍"""
    return initial_lr * np.power(drop_factor, np.floor(epoch / epochs_drop))

# 指数衰减
def exponential_decay(epoch, initial_lr=0.01, decay_rate=0.95):
    """学习率按指数衰减"""
    return initial_lr * np.power(decay_rate, epoch)

# 余弦退火
def cosine_annealing(epoch, initial_lr=0.01, min_lr=0.0001, total_epochs=50):
    """余弦退火学习率调度"""
    return min_lr + 0.5 * (initial_lr - min_lr) * (1 + np.cos(epoch / total_epochs * np.pi))

# 可视化不同的学习率调度
epochs = range(50)
plt.figure(figsize=(12, 6))
plt.plot(epochs, [step_decay(e) for e in epochs], 'r-', label='Step Decay')
plt.plot(epochs, [exponential_decay(e) for e in epochs], 'g-', label='Exponential Decay')
plt.plot(epochs, [cosine_annealing(e) for e in epochs], 'b-', label='Cosine Annealing')
plt.xlabel('Epochs')
plt.ylabel('Learning Rate')
plt.title('Learning Rate Schedules')
plt.legend()
plt.grid(True)
# plt.show()  # 在实际环境中显示图表

# 在模型训练中使用学习率调度
lr_scheduler = LearningRateScheduler(cosine_annealing)
# model.fit(X_train, y_train, epochs=50, callbacks=[lr_scheduler])

import tensorflow as tf
from tensorflow.keras.layers import Layer, Dense, MultiHeadAttention, LayerNormalization
from tensorflow.keras.models import Sequential

class TransformerBlock(Layer):
    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):
        super().__init__()
        
        # 多头注意力机制
        self.att = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        
        # 前馈网络 (MLP)
        self.ffn = Sequential([
            Dense(ff_dim, activation="relu"),  # 扩展维度
            Dense(embed_dim)                   # 恢复原始维度
        ])
        
        # 层归一化
        self.layernorm1 = LayerNormalization(epsilon=1e-6)
        self.layernorm2 = LayerNormalization(epsilon=1e-6)
        
        # Dropout
        self.dropout1 = tf.keras.layers.Dropout(rate)
        self.dropout2 = tf.keras.layers.Dropout(rate)
    
    def call(self, inputs, training=True):
        # 自注意力机制 + 残差连接
        attn_output = self.att(inputs, inputs)
        attn_output = self.dropout1(attn_output, training=training)
        out1 = self.layernorm1(inputs + attn_output)
        
        # 前馈网络 (MLP) + 残差连接
        ffn_output = self.ffn(out1)
        ffn_output = self.dropout2(ffn_output, training=training)
        return self.layernorm2(out1 + ffn_output)

import tensorflow as tf
from tensorflow.keras.layers import Layer, Dense
from tensorflow.keras.models import Sequential

class GraphNeuralNetworkLayer(Layer):
    def __init__(self, output_dim):
        super().__init__()
        
        # 节点特征变换的MLP
        self.feature_mlp = Sequential([
            Dense(128, activation='relu'),
            Dense(output_dim)
        ])
        
        # 边特征处理的MLP（可选）
        self.edge_mlp = Sequential([
            Dense(64, activation='relu'),
            Dense(output_dim)
        ])
    
    def call(self, inputs, adj_matrix, edge_features=None):
        # 消息传递：聚合邻居信息
        # adj_matrix是邻接矩阵，表示节点之间的连接关系
        aggregated = tf.matmul(adj_matrix, inputs)
        
        # 如果有边特征，处理边信息
        if edge_features is not None:
            edge_transformed = self.edge_mlp(edge_features)
            # 将边特征与聚合的节点特征结合
            combined = tf.concat([aggregated, edge_transformed], axis=-1)
            return self.feature_mlp(combined)
        
        # 使用MLP变换聚合后的特征
        return self.feature_mlp(aggregated)

import autokeras as ak

# 使用AutoML自动搜索最佳MLP架构
classifier = ak.StructuredDataClassifier(
    max_trials=10,  # 尝试10种不同架构
    overwrite=True
)
classifier.fit(X_train, y_train, epochs=10)

def attention_mlp(inputs, hidden_units):
    # 自注意力机制
    query = Dense(hidden_units)(inputs)
    key = Dense(hidden_units)(inputs)
    value = Dense(hidden_units)(inputs)
    
    # 计算注意力权重
    attention_scores = tf.matmul(query, key, transpose_b=True)
    attention_scores = attention_scores / tf.math.sqrt(tf.cast(hidden_units, tf.float32))
    attention_weights = tf.nn.softmax(attention_scores, axis=-1)
    
    # 应用注意力权重
    context = tf.matmul(attention_weights, value)
    
    # MLP处理
    output = Dense(hidden_units, activation='relu')(context)
    return output