企业官网建设流程全解析

import torch import numpy as np from torch.utils.data import Dataset, DataLoader import os import torch import torch.nn as nn import matplotlib.pyplot as plt import matplotlib import numpy as np # ============ 添加这几行修复中文显示 ============ matplotlib.rcParams['font.sans-serif'] = ['SimHei'] # 或 'Microsoft YaHei' matplotlib.rcParams['axes.unicode_minus'] = False # ================================================ # 然后继续原来的类定义和可视化代码... # 设置随机种子，确保可重复性 torch.manual_seed(42) np.random.seed(42) # 定义序列长度列表 lengths = [10, 20, 50, 100, 150, 200] # 数据保存目录（可根据需要修改） data_dir = "./addition_data" os.makedirs(data_dir, exist_ok=True) def generate_addition_dataset(num_samples, seq_length, noise_prob=0.1): """ 生成数字求和任务数据集 参数: num_samples: 样本数量 seq_length: 序列长度 T noise_prob: 非前两个位置填充非零数字的概率（默认0.1，即10%） 返回: inputs: one-hot编码的输入序列，shape [num_samples, seq_length, 10] labels: 加和标签，shape [num_samples] """ # 前两个位置：均匀随机0~9 pos1 = np.random.randint(0, 10, size=num_samples) pos2 = np.random.randint(0, 10, size=num_samples) # 标签：pos1 + pos2 labels = pos1 + pos2 # shape [num_samples] # 初始化序列为全零 sequences = np.zeros((num_samples, seq_length), dtype=int) # 填充前两个位置 sequences[:, 0] = pos1 sequences[:, 1] = pos2 # 其余位置（从第2位到最后）加入噪声 if seq_length > 2: mask = np.random.rand(num_samples, seq_length - 2) < noise_prob noise_values = np.random.randint(1, 10, size=(num_samples, seq_length - 2)) # 噪声为1~9 sequences[:, 2:] = mask * noise_values # 转换为one-hot编码：每个数字变为10维独热向量 # 使用numpy实现，避免循环 one_hot = np.eye(10)[sequences] # shape [num_samples, seq_length, 10] return torch.tensor(one_hot, dtype=torch.float32), torch.tensor(labels, dtype=torch.long) # 生成并保存所有数据集 for T in lengths: print(f"正在生成序列长度 T={T} 的数据集...") # 训练集 train_inputs, train_labels = generate_addition_dataset(1000, T) torch.save({'inputs': train_inputs, 'labels': train_labels}, os.path.join(data_dir, f'train_T{T}.pt')) # 测试集 test_inputs, test_labels = generate_addition_dataset(200, T) torch.save({'inputs': test_inputs, 'labels': test_labels}, os.path.join(data_dir, f'test_T{T}.pt')) print(f"T={T} 完成：训练集10000条，测试集2000条，已保存。") print("所有数据集生成完毕！") print(f"数据保存在目录：{data_dir}") print("每个文件包含 'inputs' (shape: [N, T, 10]) 和 'labels' (shape: [N]) 两个张量。") # 示例：加载并验证某个数据集 example_T = 20 data = torch.load(os.path.join(data_dir, f'train_T{example_T}.pt')) print(f"\n验证示例（T={example_T} 训练集前3条）：") print("输入序列（前3个时间步的one-hot，转为数字显示）：") seq_digits = torch.argmax(data['inputs'][:3, :, :], dim=-1) print(seq_digits[:, :3]) print("对应标签（前两个数字之和）：", data['labels'][:3]) import torch import torch.nn as nn import matplotlib.pyplot as plt import numpy as np class SimpleRNN(nn.Module): def __init__(self, input_size=10, hidden_size=100, output_size=19): super(SimpleRNN, self).__init__() self.hidden_size = hidden_size # 参数定义 self.W_xh = nn.Parameter(torch.zeros(input_size, hidden_size)) self.W_hh = nn.Parameter(torch.zeros(hidden_size, hidden_size)) self.W_hy = nn.Parameter(torch.zeros(hidden_size, output_size)) self.b_h = nn.Parameter(torch.zeros(hidden_size)) self.b_y = nn.Parameter(torch.zeros(output_size)) # Xavier 初始化 nn.init.xavier_uniform_(self.W_xh) nn.init.xavier_uniform_(self.W_hh) nn.init.xavier_uniform_(self.W_hy) nn.init.zeros_(self.b_h) nn.init.zeros_(self.b_y) self.tanh = nn.Tanh() def forward(self, x): """ x: [batch_size, seq_len, input_size] 返回: logits [batch_size, output_size], hidden_states [batch_size, seq_len, hidden_size] """ batch_size, seq_len, _ = x.size() hidden = torch.zeros(batch_size, self.hidden_size, device=x.device) hidden_states = [] for t in range(seq_len): hidden = self.tanh( torch.matmul(x[:, t, :], self.W_xh) + torch.matmul(hidden, self.W_hh) + self.b_h ) hidden_states.append(hidden.unsqueeze(1)) # 方便后续stack logits = torch.matmul(hidden, self.W_hy) + self.b_y hidden_states = torch.cat(hidden_states, dim=1) # [batch, seq_len, hidden] return logits, hidden_states # ==================== 1. 实例化模型 ==================== model = SimpleRNN(input_size=10, hidden_size=100, output_size=19) # 参数量统计 total_params = sum(p.numel() for p in model.parameters()) print(f"模型总参数量: {total_params}") print(model) # ==================== 2. 权重分布可视化 ==================== plt.figure(figsize=(15, 4)) weights = [ ('W_xh (输入→隐藏)', model.W_xh.detach().cpu().numpy().flatten()), ('W_hh (隐藏→隐藏)', model.W_hh.detach().cpu().numpy().flatten()), ('W_hy (隐藏→输出)', model.W_hy.detach().cpu().numpy().flatten()) ] for i, (name, data) in enumerate(weights, 1): plt.subplot(1, 3, i) plt.hist(data, bins=50, color='skyblue', edgecolor='black', alpha=0.7) plt.title(name) plt.xlabel('权重值') plt.ylabel('频数') plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() # ==================== 3. 循环权重谱半径计算 ==================== with torch.no_grad(): eigenvalues = torch.linalg.eigvals(model.W_hh) spectral_radius = torch.max(torch.abs(eigenvalues)).item() print(f"\n循环权重 W_hh 的谱半径 (spectral radius): {spectral_radius:.4f}") # ==================== 4. 前向传播示例与隐藏状态可视化 ==================== # 生成随机输入（batch_size=8, seq_len=50） torch.manual_seed(123) batch_size = 8 seq_len = 50 x_example = torch.zeros(batch_size, seq_len, 10) # 前两个位置设置随机数字（one-hot） pos1 = torch.randint(0, 10, (batch_size,)) pos2 = torch.randint(0, 10, (batch_size,)) x_example[torch.arange(batch_size), 0, pos1] = 1.0 x_example[torch.arange(batch_size), 1, pos2] = 1.0 logits, hidden_states = model(x_example) print(f"\n前向传播验证：") print(f"输入形状: {x_example.shape}") print(f"输出 logits 形状: {logits.shape}") print(f"隐藏状态序列形状: {hidden_states.shape}") # 隐藏状态平均 L2 范数随时间步变化 hidden_norms = torch.norm(hidden_states, dim=-1).mean(dim=0).detach().cpu().numpy() plt.figure(figsize=(10, 5)) plt.plot(range(1, seq_len + 1), hidden_norms, marker='o', color='orange', linewidth=2, markersize=4) plt.title('隐藏状态平均 L2 范数随时间步变化 (seq_len=50)') plt.xlabel('时间步 t') plt.ylabel('平均 L2 范数') plt.grid(True, alpha=0.3) plt.show() print("\n前向传播示例完成。你可以直接截取以上两张图（权重分布 + 隐藏状态范数曲线）用于实验报告。") import torch import torch.nn as nn import torch.optim as optim import matplotlib.pyplot as plt import matplotlib import numpy as np import os # 修复中文显示（如果需要） matplotlib.rcParams['font.sans-serif'] = ['SimHei'] # 或 'Microsoft YaHei' matplotlib.rcParams['axes.unicode_minus'] = False # 假设已运行子任务1的数据生成代码，数据保存在 ./addition_data/ data_dir = "./addition_data" # 加载模型（与子任务2相同） class SimpleRNN(nn.Module): # （直接复制子任务2的模型类定义） def __init__(self, input_size=10, hidden_size=100, output_size=19): super(SimpleRNN, self).__init__() self.hidden_size = hidden_size self.W_xh = nn.Parameter(torch.zeros(input_size, hidden_size)) self.W_hh = nn.Parameter(torch.zeros(hidden_size, hidden_size)) self.W_hy = nn.Parameter(torch.zeros(hidden_size, output_size)) self.b_h = nn.Parameter(torch.zeros(hidden_size)) self.b_y = nn.Parameter(torch.zeros(output_size)) nn.init.xavier_uniform_(self.W_xh) nn.init.xavier_uniform_(self.W_hh) nn.init.xavier_uniform_(self.W_hy) nn.init.zeros_(self.b_h) nn.init.zeros_(self.b_y) self.tanh = nn.Tanh() def forward(self, x): batch_size, seq_len, _ = x.size() hidden = torch.zeros(batch_size, self.hidden_size, device=x.device) for t in range(seq_len): hidden = self.tanh( torch.matmul(x[:, t, :], self.W_xh) + torch.matmul(hidden, self.W_hh) + self.b_h ) logits = torch.matmul(hidden, self.W_hy) + self.b_y return logits # 训练函数 def train_on_length(T, epochs=100, lr=0.01): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # 加载数据 train_data = torch.load(os.path.join(data_dir, f'train_T{T}.pt')) test_data = torch.load(os.path.join(data_dir, f'test_T{T}.pt')) train_inputs, train_labels = train_data['inputs'].to(device), train_data['labels'].to(device) test_inputs, test_labels = test_data['inputs'].to(device), test_data['labels'].to(device) model = SimpleRNN().to(device) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=lr) train_losses = [] test_accs = [] for epoch in range(epochs): model.train() optimizer.zero_grad() logits = model(train_inputs) loss = criterion(logits, train_labels) loss.backward() optimizer.step() train_losses.append(loss.item()) # 测试准确率 model.eval() with torch.no_grad(): test_logits = model(test_inputs) pred = torch.argmax(test_logits, dim=1) acc = (pred == test_labels).float().mean().item() * 100 test_accs.append(acc) if (epoch + 1) % 20 == 0 or acc >= 99.9: print(f"T={T}, Epoch {epoch + 1}: Loss {loss.item():.4f}, Test Acc {acc:.2f}%") return train_losses, test_accs # 运行短序列训练（T=10, 20, 50） short_lengths = [10, 20, 50] results = {} for T in short_lengths: print(f"\n=== 训练序列长度 T={T} ===") losses, accs = train_on_length(T, epochs=100 if T <= 20 else 150, lr=0.01) results[T] = (losses, accs) # 可视化示例：T=20 losses, accs = results[20] plt.figure(figsize=(12, 5)) plt.subplot(1, 2, 1) plt.plot(losses, color='blue', linewidth=2) plt.title('训练损失随Epoch变化 (T=20)') plt.xlabel('Epoch') plt.ylabel('损失') plt.grid(True, alpha=0.3) plt.subplot(1, 2, 2) plt.plot(accs, color='green', linewidth=2) plt.title('测试准确率随Epoch变化 (T=20)') plt.xlabel('Epoch') plt.ylabel('准确率 (%)') plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() # 汇总短序列最终准确率 print("\n短序列最终测试准确率汇总：") for T in short_lengths: final_acc = results[T][1][-1] print(f"T={T}: {final_acc:.2f}%") import torch import torch.nn as nn import torch.optim as optim import matplotlib.pyplot as plt import matplotlib import numpy as np import os # 修复中文显示 matplotlib.rcParams['font.sans-serif'] = ['SimHei'] # 或 'Microsoft YaHei' matplotlib.rcParams['axes.unicode_minus'] = False data_dir = "./addition_data" # 模型定义（与之前完全一致） class SimpleRNN(nn.Module): def __init__(self, input_size=10, hidden_size=100, output_size=19): super(SimpleRNN, self).__init__() self.hidden_size = hidden_size self.W_xh = nn.Parameter(torch.zeros(input_size, hidden_size)) self.W_hh = nn.Parameter(torch.zeros(hidden_size, hidden_size)) self.W_hy = nn.Parameter(torch.zeros(hidden_size, output_size)) self.b_h = nn.Parameter(torch.zeros(hidden_size)) self.b_y = nn.Parameter(torch.zeros(output_size)) nn.init.xavier_uniform_(self.W_xh) nn.init.xavier_uniform_(self.W_hh) nn.init.xavier_uniform_(self.W_hy) nn.init.zeros_(self.b_h) nn.init.zeros_(self.b_y) self.tanh = nn.Tanh() def forward(self, x): batch_size, seq_len, _ = x.size() hidden = torch.zeros(batch_size, self.hidden_size, device=x.device) for t in range(seq_len): hidden = self.tanh( torch.matmul(x[:, t, :], self.W_xh) + torch.matmul(hidden, self.W_hh) + self.b_h ) logits = torch.matmul(hidden, self.W_hy) + self.b_y return logits # 训练函数（同子任务3） def train_on_length(T, epochs=200, lr=0.01): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") train_data = torch.load(os.path.join(data_dir, f'train_T{T}.pt')) test_data = torch.load(os.path.join(data_dir, f'test_T{T}.pt')) train_inputs, train_labels = train_data['inputs'].to(device), train_data['labels'].to(device) test_inputs, test_labels = test_data['inputs'].to(device), test_data['labels'].to(device) model = SimpleRNN().to(device) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=lr) train_losses = [] test_accs = [] for epoch in range(epochs): model.train() optimizer.zero_grad() logits = model(train_inputs) loss = criterion(logits, train_labels) loss.backward() optimizer.step() train_losses.append(loss.item()) model.eval() with torch.no_grad(): test_logits = model(test_inputs) pred = torch.argmax(test_logits, dim=1) acc = (pred == test_labels).float().mean().item() * 100 test_accs.append(acc) if (epoch + 1) % 20 == 0: print(f"T={T}, Epoch {epoch + 1}: Loss {loss.item():.4f}, Test Acc {acc:.2f}%") return train_losses, test_accs, model # 长序列训练 long_lengths = [100, 150, 200] results_long = {} for T in long_lengths: print(f"\n=== 训练长序列 T={T} ===") losses, accs, _ = train_on_length(T, epochs=200, lr=0.01) results_long[T] = (losses, accs) # 可视化示例：T=150 losses_150, accs_150 = results_long[150] plt.figure(figsize=(12, 5)) plt.subplot(1, 2, 1) plt.plot(losses_150, color='red', linewidth=2) plt.title('训练损失随Epoch变化 (T=150)') plt.xlabel('Epoch') plt.ylabel('损失') plt.grid(True, alpha=0.3) plt.subplot(1, 2, 2) plt.plot(accs_150, color='purple', linewidth=2) plt.title('测试准确率随Epoch变化 (T=150)') plt.xlabel('Epoch') plt.ylabel('准确率 (%)') plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() # 汇总所有序列长度准确率（包括短序列） all_lengths = [10, 20, 50, 100, 150, 200] final_accs = [] # 假设短序列结果已保存或重新计算，这里用示例值（实际运行后替换） short_accs = {10: 100.0, 20: 100.0, 50: 99.95} for T in all_lengths: if T in short_accs: final_accs.append(short_accs[T]) else: final_accs.append(results_long[T][1][-1]) plt.figure(figsize=(10, 6)) plt.bar(all_lengths, final_accs, color=['green'] * 3 + ['orange'] * 3) plt.title('不同序列长度下最终测试准确率对比') plt.xlabel('序列长度 T') plt.ylabel('测试准确率 (%)') plt.ylim(0, 110) for i, v in enumerate(final_accs): plt.text(all_lengths[i], v + 2, f"{v:.2f}%", ha='center') plt.grid(True, alpha=0.3, axis='y') plt.show() print("\n长序列最终测试准确率汇总：") for T in long_lengths: print(f"T={T}: {results_long[T][1][-1]:.2f}%") import torch import torch.nn as nn import torch.optim as optim import matplotlib.pyplot as plt import matplotlib import numpy as np import os # 修复中文显示 matplotlib.rcParams['font.sans-serif'] = ['SimHei'] matplotlib.rcParams['axes.unicode_minus'] = False data_dir = "./addition_data" # 模型定义（同前） class SimpleRNN(nn.Module): def __init__(self, input_size=10, hidden_size=100, output_size=19): super(SimpleRNN, self).__init__() self.hidden_size = hidden_size self.W_xh = nn.Parameter(torch.zeros(input_size, hidden_size)) self.W_hh = nn.Parameter(torch.zeros(hidden_size, hidden_size)) self.W_hy = nn.Parameter(torch.zeros(hidden_size, output_size)) self.b_h = nn.Parameter(torch.zeros(hidden_size)) self.b_y = nn.Parameter(torch.zeros(output_size)) nn.init.xavier_uniform_(self.W_xh) nn.init.xavier_uniform_(self.W_hh) nn.init.xavier_uniform_(self.W_hy) nn.init.zeros_(self.b_h) nn.init.zeros_(self.b_y) self.tanh = nn.Tanh() def forward(self, x): batch_size, seq_len, _ = x.size() hidden = torch.zeros(batch_size, self.hidden_size, device=x.device) for t in range(seq_len): hidden = self.tanh( torch.matmul(x[:, t, :], self.W_xh) + torch.matmul(hidden, self.W_hh) + self.b_h ) logits = torch.matmul(hidden, self.W_hy) + self.b_y return logits # 训练函数（带梯度范数记录和可选裁剪） def train_with_grad_monitor(T=200, epochs=50, lr=0.01, clip_norm=None): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") data = torch.load(os.path.join(data_dir, f'train_T{T}.pt')) inputs, labels = data['inputs'][:1000].to(device), data['labels'][:1000].to(device) # 用少量样本加速 model = SimpleRNN().to(device) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=lr) losses = [] grad_norms = [] for epoch in range(epochs): model.train() optimizer.zero_grad() logits = model(inputs) loss = criterion(logits, labels) loss.backward() # 记录梯度范数（裁剪前） total_norm = 0 for p in model.parameters(): if p.grad is not None: total_norm += p.grad.detach().data.norm(2).item() ** 2 total_norm = total_norm ** 0.5 grad_norms.append(total_norm) # 可选梯度裁剪 if clip_norm is not None: torch.nn.utils.clip_grad_norm_(model.parameters(), clip_norm) optimizer.step() losses.append(loss.item()) print(f"Epoch {epoch + 1}: Loss {loss.item():.4f}, Grad Norm {total_norm:.2e}") if np.isnan(loss.item()): print("损失变为NaN，梯度爆炸！") break return losses, grad_norms # 运行两次：无裁剪（观察爆炸）和有裁剪 print("=== 无梯度裁剪（T=200） ===") losses_no_clip, norms_no_clip = train_with_grad_monitor(T=200, epochs=50, clip_norm=None) print("\n=== 施加梯度裁剪（阈值5.0） ===") losses_clip, norms_clip = train_with_grad_monitor(T=200, epochs=50, clip_norm=5.0) # 可视化对比 plt.figure(figsize=(15, 5)) plt.subplot(1, 3, 1) plt.plot(losses_no_clip, color='red', label='无裁剪') plt.plot(losses_clip, color='blue', label='裁剪阈值5.0') plt.title('训练损失对比 (T=200)') plt.xlabel('Epoch') plt.ylabel('损失') plt.legend() plt.grid(True, alpha=0.3) plt.subplot(1, 3, 2) plt.plot(norms_no_clip, color='red', marker='o') plt.title('梯度范数（无裁剪）') plt.xlabel('Epoch') plt.ylabel('梯度L2范数') plt.yscale('log') plt.grid(True, alpha=0.3) plt.subplot(1, 3, 3) plt.plot(norms_clip, color='blue', marker='o') plt.title('梯度范数（裁剪后）') plt.xlabel('Epoch') plt.ylabel('梯度L2范数') plt.grid(True, alpha=0.3) plt.tight_layout() plt.show()