AI算法与编程深度实践：从卷积神经网络到强化学习的链路解析

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader

# 1. 定义CNN架构
class SimpleCNN(nn.Module):
    def __init__(self):
        super(SimpleCNN, self).__init__()
        # 卷积层: 输入1通道, 输出6通道, 核大小5x5
        self.conv1 = nn.Conv2d(1, 6, kernel_size=5, padding=2)  # 保持尺寸
        self.pool = nn.AvgPool2d(kernel_size=2, stride=2)       # 2x2下采样
        self.conv2 = nn.Conv2d(6, 16, kernel_size=5)            # 输出16通道
        self.fc1 = nn.Linear(16 * 5 * 5, 120)                   # 全连接层
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)                            # 10分类

    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))   # 32x32 -> 16x16
        x = self.pool(F.relu(self.conv2(x)))   # -> 8x8? 实际是 16*5*5
        x = x.view(-1, 16 * 5 * 5)             # 展平
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# 2. 数据加载与预处理
transform = transforms.Compose([
    transforms.Resize((32, 32)),   # 适配LeNet输入
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,))
])
train_dataset = datasets.MNIST('./data', train=True, download=True, transform=transform)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

# 3. 训练循环（含学习率调度与早停）
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = SimpleCNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=3, gamma=0.7)

def train(epochs=5):
    model.train()
    for epoch in range(epochs):
        running_loss = 0.0
        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)
            optimizer.zero_grad()
            outputs = model(images)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            running_loss += loss.item()
        scheduler.step()
        print(f'Epoch {epoch+1}, Loss: {running_loss/len(train_loader):.4f}, LR: {scheduler.get_last_lr()[0]:.6f}')

train(epochs=5)

import gym
import numpy as np
from collections import deque
import random

# 构建DQN模型（以CartPole环境为例）
class DQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden=128):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(state_dim, hidden)
        self.fc2 = nn.Linear(hidden, hidden)
        self.fc3 = nn.Linear(hidden, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        return self.fc3(x)

# 经验回放缓冲区
class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = map(np.stack, zip(*batch))
        return state, action, reward, next_state, done

    def __len__(self):
        return len(self.buffer)

# DQN训练主循环（含目标网络与ε-贪心）
def train_dqn(env_name='CartPole-v1', episodes=500, batch_size=64, gamma=0.99, epsilon_start=1.0, epsilon_end=0.01):
    env = gym.make(env_name)
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.n
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    policy_net = DQN(state_dim, action_dim).to(device)
    target_net = DQN(state_dim, action_dim).to(device)
    target_net.load_state_dict(policy_net.state_dict())
    target_net.eval()
    optimizer = optim.Adam(policy_net.parameters(), lr=1e-3)
    replay_buffer = ReplayBuffer(capacity=10000)
    epsilon = epsilon_start
    scores = []
    
    for episode in range(episodes):
        state = env.reset()
        state = torch.FloatTensor(state).to(device)
        total_reward = 0
        done = False
        while not done:
            # ε-贪心选择动作
            if np.random.random() < epsilon:
                action = env.action_space.sample()
            else:
                with torch.no_grad():
                    q_values = policy_net(state.unsqueeze(0))
                    action = q_values.max(1)[1].item()
            next_state, reward, done, _ = env.step(action)
            total_reward += reward
            next_state = torch.FloatTensor(next_state).to(device)
            # 存储经验
            replay_buffer.push(state.cpu().numpy(), action, reward, next_state.cpu().numpy(), done)
            state = next_state
            # 若缓冲区足够，采样训练
            if len(replay_buffer) >= batch_size:
                states, actions, rewards, next_states, dones = replay_buffer.sample(batch_size)
                states = torch.FloatTensor(states).to(device)
                actions = torch.LongTensor(actions).to(device)
                rewards = torch.FloatTensor(rewards).to(device)
                next_states = torch.FloatTensor(next_states).to(device)
                dones = torch.FloatTensor(dones).to(device)
                
                current_q = policy_net(states).gather(1, actions.unsqueeze(1))
                next_q = target_net(next_states).max(1)[0].detach()
                target_q = rewards + gamma * next_q * (1 - dones)
                loss = F.mse_loss(current_q.squeeze(), target_q)
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
        # 衰减ε
        epsilon = max(epsilon_end, epsilon * 0.995)
        scores.append(total_reward)
        # 每10轮更新目标网络
        if episode % 10 == 0:
            target_net.load_state_dict(policy_net.state_dict())
        print(f'Episode {episode+1}, Score: {total_reward}, Epsilon: {epsilon:.3f}')
        if np.mean(scores[-10:]) >= 195:  # 环境平均分阈值
            print('Solved!')
            break
    env.close()

train_dqn()

import matplotlib.pyplot as plt

# 线性噪声调度
def linear_beta_schedule(timesteps=1000, beta_start=1e-4, beta_end=0.02):
    return torch.linspace(beta_start, beta_end, timesteps)

# 前向加噪：从原图逐步加噪
def forward_diffusion(x0, t, betas, sqrt_alphas_bar):
    noise = torch.randn_like(x0)
    sqrt_alpha_bar = sqrt_alphas_bar[t]
    x_t = sqrt_alpha_bar * x0 + torch.sqrt(1 - sqrt_alpha_bar) * noise
    return x_t, noise

# 简单UNet（仅示意）
class SimpleUNet(nn.Module):
    def __init__(self, in_channels=3, out_channels=3):
        super(SimpleUNet, self).__init__()
        self.encoder = nn.Sequential(
            nn.Conv2d(in_channels, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, padding=1),
            nn.ReLU(),
        )
        self.decoder = nn.Sequential(
            nn.Conv2d(64, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.Conv2d(64, out_channels, kernel_size=3, padding=1),
        )
    def forward(self, x, t):
        # 实际需加入时间步t的嵌入
        return self.decoder(self.encoder(x))

# 训练循环（简化）
def train_diffusion(dataloader, timesteps=1000):
    betas = linear_beta_schedule(timesteps)
    alphas = 1. - betas
    alphas_bar = torch.cumprod(alphas, dim=0)
    sqrt_alphas_bar = torch.sqrt(alphas_bar)
    model = SimpleUNet().cuda()
    optimizer = optim.Adam(model.parameters(), lr=1e-4)
    mse = nn.MSELoss()
    for epoch in range(10):
        for batch in dataloader:
            x0 = batch.cuda()
            t = torch.randint(0, timesteps, (x0.size(0),), device='cuda')
            x_t, noise = forward_diffusion(x0, t, betas, sqrt_alphas_bar)
            predicted_noise = model(x_t, t)
            loss = mse(predicted_noise, noise)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        print(f'Epoch {epoch+1}, Loss: {loss.item():.4f}')