1.1.1案例介绍
今天我们将一起探索如何使用华为的 MindSpore 深度学习框架，从零开始构建一个变分自编码器（Variational Autoencoder, VAE），并用它来生成独特的数字艺术作品。
这篇文章不仅会提供完整的代码，还会逐行解释其背后的原理和功能，让你不仅知其然，更知其所以然。
什么是变分自编码器 (VAE)？
在深入代码之前，我们先简单了解一下 VAE。VAE 是一种强大的生成模型，它由两部分组成：
编码器 (Encoder)：它的任务是学习将输入数据（比如一张图片）压缩成一个潜在空间（Latent Space）中的分布。这个分布通常用均值（mu）和方差（log_var）来描述。
解码器 (Decoder)：它的任务是从这个潜在空间中随机采样一个点，并将其解码（还原）成与原始输入数据相似的输出。
训练完成后，我们就可以直接从潜在空间中随机采样，然后用解码器生成全新的、从未见过的图片。这就是我们实现 “AI 艺术生成” 的核心原理。
1.1.2环境设置
首先，我们需要导入所有必要的库，并对 MindSpore 进行一些初始设置。
import mindspore as ms
import mindspore.nn as nn
import mindspore.ops as ops
from mindspore import Tensor, context
import numpy as np
import matplotlib.pyplot as plt
import os
# 设置CPU环境。如果你有GPU并已配置好，可以改为"GPU"以获得更快的训练速度
# GRAPH_MODE表示使用静态图模式，这是MindSpore默认且高效的执行模式
context.set_context(mode=context.GRAPH_MODE, device_target="CPU")
# 打印MindSpore版本和项目标题，确认环境配置正确
print(f"MindSpore版本: {ms.__version__}")
print("qianduanjidi-AI艺术生成器")
# 创建一个名为 "qianduanjidi" 的目录，用于存放所有生成的文件（模型、图片等）
# exist_ok=True 表示如果目录已存在，则不会报错
demo_dir = "qianduanjidi"
os.makedirs(demo_dir, exist_ok=True)
# 设置随机种子。这是一个好习惯，可以确保你的实验结果是可复现的
ms.set_seed(42)
代码分析：
我们导入了 mindspore 及其核心模块 nn (神经网络) 和 ops (算子)。
numpy 用于数值计算，matplotlib.pyplot 用于绘图和可视化。
os 模块用于与操作系统交互，比如创建目录和读写文件。
context.set_context 是 MindSpore 的核心配置入口，这里我们指定了使用 CPU 和静态图模式。
os.makedirs 创建了我们的工作目录。
ms.set_seed 固定了随机数生成器的种子，使得每次运行代码时，初始化的权重和随机数据都一样。
1.1.3构建 VAE 模型
接下来，我们将定义 VAE 的核心结构。
class SimpleVAE(nn.Cell):
"""变分自编码器"""
def __init__(self, image_size=784, h_dim=400, z_dim=20):
super(SimpleVAE, self).__init__()
# 编码器 (Encoder)
# 它是一个简单的神经网络，输入是展平的图片 (784像素)
# 输出是潜在空间分布的参数：mu和log_var (各20个维度)
self.encoder = nn.SequentialCell(
nn.Dense(image_size, h_dim), # 全连接层：784 -> 400
nn.ReLU(), # 激活函数，增加非线性
nn.Dense(h_dim, z_dim * 2) # 全连接层：400 -> 40 (20个mu + 20个log_var)
)
# 解码器 (Decoder)
# 它从潜在空间的一个点 (z_dim=20) 开始，尝试还原出原始图片
self.decoder = nn.SequentialCell(
nn.Dense(z_dim, h_dim), # 全连接层：20 -> 400
nn.ReLU(), # 激活函数
nn.Dense(h_dim, image_size), # 全连接层：400 -> 784
nn.Sigmoid() # Sigmoid激活，确保输出值在0到1之间（像灰度图的像素值）
)
def encode(self, x):
"""将输入x编码为mu和log_var"""
h = self.encoder(x)
mu, log_var = h[:, :20], h[:, 20:] # 将输出的40个维度切分成两部分
return mu, log_var
def reparameterize(self, mu, log_var):
"""
重参数化技巧 (Reparameterization Trick)
这是VAE的关键。为了让梯度能够顺畅地从解码器流回编码器，
我们不直接从 N(mu, sigma^2) 中采样，而是：
1. 从标准正态分布 N(0, 1) 中采样一个 eps
2. 计算 z = mu + eps * sigma (其中 sigma = exp(log_var / 2))
这样，采样过程就变得可微分了。
"""
std = ops.Exp()(log_var * 0.5)
eps = ops.StandardNormal()(std.shape)
return mu + eps * std
def decode(self, z):
"""将潜在向量z解码为图片"""
return self.decoder(z)
def construct(self, x):
"""
MindSpore模型的核心执行逻辑。
当你调用模型实例时，这个方法会被执行。
"""
mu, log_var = self.encode(x)
z = self.reparameterize(mu, log_var)
return self.decode(z), mu, log_var
代码分析：
我们定义了一个继承自 nn.Cell 的 SimpleVAE 类，这是 MindSpore 中所有神经网络模型的基类。
init 方法中定义了模型的层。encoder 和 decoder 都是 nn.SequentialCell，它可以将多个层按顺序组合起来。
encode, reparameterize, decode 方法将 VAE 的前向传播过程分解为清晰的步骤。
重参数化技巧是 VAE 最重要的创新，它解决了随机采样过程不可微分的问题，使得整个模型可以被端到端地训练。
construct 方法是模型的 “入口”。当我们把数据送入模型时，MindSpore 会自动调用这个方法。它返回了三个值：重建的图片、均值 mu 和对数方差 log_var。
1.1.4定义训练逻辑
MindSpore 中，我们通常将训练逻辑封装在另一个 Cell 中，称为 “训练器”。
class VAETrainer(nn.Cell):
"""
训练器Cell，负责定义单步训练的逻辑。
"""
def __init__(self, network, optimizer):
super(VAETrainer, self).__init__(auto_prefix=False)
self.network = network # 我们的VAE模型
self.optimizer = optimizer # 优化器，如Adam
self.weights = self.optimizer.parameters # 需要被优化的模型参数
self.grad = ops.GradOperation(get_by_list=True) # 用于计算梯度的算子
def construct(self, x):
# 1. 前向传播：将输入x送入VAE模型
reconstructed, mu, log_var = self.network(x)
# 2. 计算损失函数 (Loss Function)
# VAE的损失由两部分组成：重建损失和KL散度损失
# a. 重建损失 (Reconstruction Loss)：衡量重建的图片与原始图片的相似度
recon_loss = ops.reduce_mean((reconstructed - x) ** 2) # MSE损失
# b. KL散度损失 (KL Divergence Loss)：衡量编码器输出的分布与标准正态分布的差异
# 这部分损失起到了正则化的作用，防止模型过拟合，并鼓励潜在空间具有良好的结构
kl_loss = -0.5 * ops.reduce_mean(1 + log_var - mu**2 - ops.exp(log_var))
# 总损失 = 重建损失 + KL损失 (通常会给KL损失一个较小的权重)
total_loss = recon_loss + 0.001 * kl_loss
# 3. 反向传播：计算梯度
grads = self.grad(self.network, self.weights)(x)
# 4. 更新参数：使用优化器根据梯度更新模型权重
self.optimizer(grads)
# 返回单步的总损失
return total_loss
代码分析：
VAETrainer 同样继承自 nn.Cell。它接收一个模型 (network) 和一个优化器 (optimizer) 作为输入。
construct 方法定义了单批次数据的训练流程：
前向传播：得到模型输出。
计算损失：VAE 的损失是重建损失和KL 散度损失的总和。
反向传播：ops.GradOperation 是一个非常强大的工具，它会自动计算 self.network 在输入 x 下的梯度。
参数更新：调用 self.optimizer(grads) 来更新模型的权重。
1.1.5创建合成数据集
为了简化演示，我们不使用真实的 MNIST 数据集，而是生成一些简单的合成数据 —— 随机的圆形。
def create_simple_dataset(num_samples=600):
"""创建一个由简单圆形组成的合成数据集"""
print("创建合成数据集...")
images = []
for i in range(num_samples):
# 创建一个28x28的全零矩阵，代表一张黑色图片
img = np.zeros((28, 28), dtype=np.float32)
# 随机选择圆形的中心点
center_x, center_y = np.random.randint(8, 20), np.random.randint(8, 20)
# 在图片上绘制一个圆形
for x in range(28):
for y in range(28):
# 计算点 (x,y) 到圆心的距离
dist = np.sqrt((x - center_x)**2 + (y - center_y)**2)
# 如果距离小于6，则该点为白色（值为0.8到1.0之间的随机数）
if dist < 6:
img[x, y] = 0.8 + np.random.rand() * 0.2
# 将28x28的矩阵展平成一个784维的向量，并添加到列表中
images.append(img.flatten())
# 将列表转换为NumPy数组并返回
return np.array(images)
代码分析：
这个函数非常直观，它通过双重循环在一个 28x28 的网格上绘制圆形。
每个样本都是一个代表 “数字 0” 的抽象圆形，位置和亮度略有不同。
最后，img.flatten() 将二维图像变成一维向量，这是我们之前定义的 Dense 层所期望的输入格式。
1.1.6主训练与生成流程
现在，我们将所有部分组装起来，编写主训练循环和后续的生成、评估代码。
def train_simple_vae():
"""训练VAE模型的主函数"""
print("开始训练VAE...")
# 1. 实例化模型和优化器
vae = SimpleVAE()
optimizer = nn.Adam(vae.trainable_params(), learning_rate=1e-3)
# 2. 实例化训练器
trainer = VAETrainer(vae, optimizer)
# 3. 获取数据
images = create_simple_dataset()
# 4. 训练循环
epochs = 20 # 训练轮数
losses = [] # 用于记录每一轮的平均损失
for epoch in range(epochs):
total_loss = 0
batch_size = 64
batch_count = 0
# 遍历数据集，按批次进行训练
for i in range(0, len(images), batch_size):
# 提取一个批次的数据
batch_data = images[i:min(i+batch_size, len(images))]
# 将NumPy数组转换为MindSpore的Tensor
batch_tensor = Tensor(batch_data, ms.float32)
# 执行单步训练，并获取损失
loss = trainer(batch_tensor)
# 累积损失
total_loss += loss.asnumpy()
batch_count += 1
# 计算并记录本轮的平均损失
avg_loss = total_loss / batch_count
losses.append(avg_loss)
# 每5轮打印一次损失信息
if (epoch + 1) % 5 == 0:
print(f"Epoch [{epoch+1}/{epochs}], Loss: {avg_loss:.4f}")
print("训练完成!")
# 返回训练好的模型和损失历史
return vae, losses
# ... (此处省略 save_model_simple, generate_and_save_images, plot_training_loss,
# demo_model_loading, list_demo_files 等辅助函数的定义，它们的代码在文末完整版本中) ...
def main():
"""程序的主入口"""
print("=" * 50)
print("MindSpore AI艺术生成器")
print("=" * 50)
# 1. 训练模型
vae, losses = train_simple_vae()
# 2. 保存训练好的模型
save_model_simple(vae, 'vae_model.ckpt')
# 3. 使用训练好的模型生成艺术图像
generate_and_save_images(vae)
# 4. 绘制并保存训练损失曲线
plot_training_loss(losses)
# 5. 演示如何加载已保存的模型并进行生成
demo_model_loading()
# 6. 列出所有生成的文件
list_demo_files()
print("\n🎉 所有任务完成!")
print(f"📁 请查看 '{demo_dir}' 文件夹中的结果文件")
# 当直接运行此脚本时，调用main函数
if __name__ == "__main__":
main()
代码分析：
train_simple_vae 函数组织了完整的训练流程：实例化模型、准备数据、然后在一个 for 循环中迭代训练。
在每一轮（epoch）中，我们又会按批次（batch）处理数据。这是深度学习训练的标准模式。
trainer(batch_tensor) 这一行是关键，它触发了我们在 VAETrainer 的 construct 方法中定义的所有计算： 前向传播、损失计算、反向传播和参数更新。
main 函数是整个程序的调度中心，它按顺序调用了训练、保存、生成、可视化等所有功能。
1.1.7运行结果
程序运行结束后，会在当前目录下创建一个名为 qianduanjidi 的文件夹，里面包含了：
vae_model.ckpt: 保存的模型权重文件。
generated_art.png: 由训练好的模型生成的 16 幅艺术图像。
training_loss.png: 训练过程中的损失变化曲线。
generated_art_from_loaded_model.png: 加载模型后生成的另一组图像，用于验证模型加载功能。
你会看到，生成的图像是一些与我们训练数据（圆形）相似但又不完全相同的抽象图案。这正是 VAE 创造力的体现！
完整代码：
import mindspore as ms
import mindspore.nn as nn
import mindspore.ops as ops
from mindspore import Tensor, context
import numpy as np
import matplotlib.pyplot as plt
import os
# 设置CPU环境
context.set_context(mode=context.GRAPH_MODE, device_target="CPU")
print(f"MindSpore版本: {ms.__version__}")
print("qianduanjidi-AI艺术生成器")
# 创建输出目录
demo_dir = "qianduanjidi"
os.makedirs(demo_dir, exist_ok=True)
# 设置随机种子
ms.set_seed(42)
class SimpleVAE(nn.Cell):
"""变分自编码器"""
def __init__(self, image_size=784, h_dim=400, z_dim=20):
super(SimpleVAE, self).__init__()
# 编码器
self.encoder = nn.SequentialCell(
nn.Dense(image_size, h_dim),
nn.ReLU(),
nn.Dense(h_dim, z_dim * 2) # 同时输出mu和log_var
)
# 解码器
self.decoder = nn.SequentialCell(
nn.Dense(z_dim, h_dim),
nn.ReLU(),
nn.Dense(h_dim, image_size),
nn.Sigmoid()
)
def encode(self, x):
h = self.encoder(x)
mu, log_var = h[:, :20], h[:, 20:]
return mu, log_var
def reparameterize(self, mu, log_var):
std = ops.Exp()(log_var * 0.5)
eps = ops.StandardNormal()(std.shape)
return mu + eps * std
def decode(self, z):
return self.decoder(z)
def construct(self, x):
mu, log_var = self.encode(x)
z = self.reparameterize(mu, log_var)
return self.decode(z), mu, log_var
class VAETrainer(nn.Cell):
"""训练器"""
def __init__(self, network, optimizer):
super(VAETrainer, self).__init__(auto_prefix=False)
self.network = network
self.optimizer = optimizer
self.weights = self.optimizer.parameters
self.grad = ops.GradOperation(get_by_list=True)
def construct(self, x):
# 前向传播
reconstructed, mu, log_var = self.network(x)
# 计算损失
recon_loss = ops.reduce_mean((reconstructed - x) ** 2)
kl_loss = -0.5 * ops.reduce_mean(1 + log_var - mu**2 - ops.exp(log_var))
loss = recon_loss + 0.001 * kl_loss
# 反向传播
grads = self.grad(self.network, self.weights)(x)
self.optimizer(grads)
return loss
def create_simple_dataset(num_samples=600):
"""创建合成数据集"""
print("创建合成数据集...")
images = []
for i in range(num_samples):
img = np.zeros((28, 28), dtype=np.float32)
center_x, center_y = np.random.randint(8, 20), np.random.randint(8, 20)
# 创建圆形数字
for x in range(28):
for y in range(28):
dist = np.sqrt((x - center_x)**2 + (y - center_y)**2)
if dist < 6:
img[x, y] = 0.8 + np.random.rand() * 0.2
images.append(img.flatten())
return np.array(images)
def train_simple_vae():
"""训练VAE"""
print("开始训练VAE...")
# 创建模型和优化器
vae = SimpleVAE()
optimizer = nn.Adam(vae.trainable_params(), learning_rate=1e-3)
trainer = VAETrainer(vae, optimizer)
# 创建数据
images = create_simple_dataset()
# 训练循环
epochs = 20
losses = []
for epoch in range(epochs):
total_loss = 0
batch_size = 64
batch_count = 0
for i in range(0, len(images), batch_size):
batch_data = images[i:min(i+batch_size, len(images))]
batch_tensor = Tensor(batch_data, ms.float32)
# 训练一步
loss = trainer(batch_tensor)
total_loss += loss.asnumpy()
batch_count += 1
avg_loss = total_loss / batch_count
losses.append(avg_loss)
if (epoch + 1) % 5 == 0:
print(f"Epoch [{epoch+1}/{epochs}], Loss: {avg_loss:.4f}")
print("训练完成!")
return vae, losses
def save_model_simple(model, filename):
"""保存模型到qianduanjidi文件夹"""
filepath = os.path.join(demo_dir, filename)
ms.save_checkpoint(model, filepath)
print(f"✅ 模型保存到: {filepath}")
def generate_and_save_images(model, num_images=16, filename='generated_art.png'):
"""生成并保存图像"""
print("生成艺术作品...")
# 将模型设置为评估模式
model.set_train(False)
# 从标准正态分布中随机采样num_images个潜在向量
z = ops.StandardNormal()((num_images, 20))
# 使用解码器生成图像
generated = model.decode(z).asnumpy()
# 绘制图像
fig, axes = plt.subplots(4, 4, figsize=(8, 8))
for i, ax in enumerate(axes.flat):
if i < num_images:
ax.imshow(generated[i].reshape(28, 28), cmap='viridis')
ax.axis('off')
ax.set_title(f'Art {i+1}')
plt.tight_layout()
# 保存图像
image_path = os.path.join(demo_dir, filename)
plt.savefig(image_path, dpi=150, bbox_inches='tight')
plt.show()
print(f"✅ 艺术作品保存到: {image_path}")
return generated
def plot_training_loss(losses):
"""绘制训练损失曲线"""
plt.figure(figsize=(8, 4))
plt.plot(losses, 'b-', linewidth=2)
plt.title('训练损失曲线')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.grid(True, alpha=0.3)
loss_path = os.path.join(demo_dir, 'training_loss.png')
plt.savefig(loss_path, dpi=150, bbox_inches='tight')
plt.show()
print(f"✅ 损失曲线保存到: {loss_path}")
def demo_model_loading():
"""演示模型加载"""
print("\n演示模型加载...")
# 创建新的模型实例（结构必须与保存时一致）
new_vae = SimpleVAE()
# 加载权重
model_path = os.path.join(demo_dir, 'vae_model.ckpt')
try:
param_dict = ms.load_checkpoint(model_path)
ms.load_param_into_net(new_vae, param_dict)
print("✅ 模型加载成功!")
# 用加载的模型生成新图像
new_images = generate_and_save_images(new_vae, 8, 'generated_art_from_loaded_model.png')
return True
except Exception as e:
print(f"❌ 模型加载失败: {e}")
return False
def list_demo_files():
"""列出qianduanjidi文件夹中的文件"""
print(f"\n📁 {demo_dir} 文件夹内容:")
if os.path.exists(demo_dir):
for file in os.listdir(demo_dir):
filepath = os.path.join(demo_dir, file)
size = os.path.getsize(filepath)
print(f" {file} ({size} bytes)")
else:
print(" 文件夹不存在")
def main():
"""主函数"""
print("=" * 50)
print("MindSpore AI艺术生成器")
print("=" * 50)
# 1. 训练模型
vae, losses = train_simple_vae()
# 2. 保存模型
save_model_simple(vae, 'vae_model.ckpt')
# 3. 生成艺术作品
generated_images = generate_and_save_images(vae)
# 4. 绘制损失曲线
plot_training_loss(losses)
# 5. 演示模型加载
demo_model_loading()
# 6. 显示文件列表
list_demo_files()
print("\n🎉 所有任务完成!")
print(f"📁 请查看 '{demo_dir}' 文件夹中的结果文件")
# 运行程序
if __name__ == "__main__":
main()