
JAX
用 JAX 实现自动微分、vmap 与 Flax 线性回归训练
从 JAX 数组与不可变性讲起,逐步演示 jax.grad 求梯度、argnums 处理多参数、用 .at[] 做函数式更新、手写 SGD 线性回归训练循环、vmap 自动向量化,以及用 Optax 与 Flax Linen 搭建编码器/解码器并统计参数量。
JAX 是一个可微的 NumPy:它的数组接口与 NumPy 高度相似,同一段代码可以不加修改地跑在 CPU、GPU 和 TPU 上,同时提供自动微分、自动向量化和即时编译能力。它源自 autograd,采用纯函数式编程范式——函数不产生副作用,状态由调用者显式传入传出。这篇教程面向已经用过 PyTorch、TensorFlow 或 NumPy 的开发者,目标是带你从数组、梯度一路走到手写训练循环、vmap 向量化,最后用 Optax 与 Flax 搭出一个小型编码器/解码器网络。如果你更想先确认站内有没有配套的在线运行环境,可以看 JAX 工具页。
准备工作
开始之前需要确认几件事。
- Python 环境:需要一个可用的 Python 解释器与 pip。素材中的示例在 Python 3.9 环境下运行过,但 JAX 对具体小版本的要求请以官网当前信息为准。
- 安装 JAX:至少需要
jax与jaxlib两个包。CPU 版本安装最简单;要使用 GPU 或 TPU,需要按官网说明选择对应的加速器版本,安装命令与可用平台以官网当前信息为准。 - 可选依赖:本文示例还会用到
numpy、matplotlib(画散点图与拟合直线)、optax(优化器)和flax(神经网络层)。这些都不是 JAX 核心的一部分,需要单独安装。 - 前置知识:熟悉 NumPy 的数组操作、了解梯度下降的基本形式、见过 PyTorch 的
nn.Module写法,会让后面的内容顺畅很多。
导入方式很直接:
import jax
import jax.numpy as jnp
如果机器上没有找到 GPU 或 TPU,JAX 会打印一条警告并回退到 CPU 执行。这条警告本身不影响结果,只是提示当前后端。想看到更详细的日志,可以按提示设置环境变量 TF_CPP_MIN_LOG_LEVEL=0 后重新运行。
操作步骤
第一步:创建并理解 JAX 数组
JAX 中对应 np.array、tf.Tensor、torch.Tensor 的类型就是 JAX 数组。创建一个向量:
x = jnp.arange(10)
print(x)
# [0 1 2 3 4 5 6 7 8 9]
这段代码在 CPU、GPU、TPU 上可以逐字运行,不需要改动。与 NumPy 数组最关键的区别是:JAX 数组永远是不可变的。任何“就地修改”的想法都要换成返回新数组的写法,这一点会贯穿后面所有步骤。
第二步:测量点积与后端差异
用两个长度为一千万的向量做点积,可以直观看到后端差异:
y = jnp.arange(int(1e7))
%timeit jnp.dot(y, y).block_until_ready()
注意末尾的 block_until_ready()。JAX 的运算默认是异步派发的,如果不等待结果就计时,测到的只是“派发耗时”而不是“计算耗时”。加上它才能得到真实数字。素材中给出的参考量级是:CPU 上每次循环约 10.6 ms,GPU 上约 362 µs。具体数值取决于硬件与版本,只作量级参考。
第三步:用 jax.grad 求梯度
对一个用 Python 写的数值函数 f,把它包进 jax.grad,会返回一个新的 Python 函数,这个新函数计算原函数的梯度。数学上即 (∇f)(x)ᵢ = ∂f/∂xᵢ。
取输入数组 [1, 2, 3, 4],定义 f(x) = Σxᵢ²,其值为 1+4+9+16 = 30,梯度为 Σ2xᵢ:
def sum_of_squares(x):
return jnp.sum(x ** 2)
sum_of_squares_dx = jax.grad(sum_of_squares)
x = jnp.asarray([1., 2., 3., 4.])
print(sum_of_squares(x)) # 30.0
print(sum_of_squares_dx(x)) # [2. 4. 6. 8.]
这里要特别注意与 TensorFlow、PyTorch 的习惯差异:JAX 不通过损失张量本身求梯度,没有 loss.backward() 这样的调用。梯度是对函数求的,得到的是另一个函数。
第四步:处理多参数函数与 argnums
当函数有多个输入参数时,jax.grad() 默认只对第一个参数求梯度。定义 f(x, y) = Σ(xᵢ − yᵢ)²,则对 x 的梯度是 2(x − y),对 y 的梯度是 2(y − x):
def sum_squared_error(x, y):
return jnp.sum((x - y) ** 2)
sum_squared_error_dx = jax.grad(sum_squared_error)
y = jnp.asarray([1.1, 2.1, 3.1, 4.1])
print(sum_squared_error_dx(x, y))
# [-0.20000005 -0.19999981 -0.19999981 -0.19999981]
可以和解析解对照:2 * (x - y) 与 2 * (y - x) 的结果与上面一致(浮点误差导致的末位差异属正常现象)。
如果要对其他参数、或同时对多个参数求梯度,用 argnums 指定参数下标,传元组即可返回多个梯度:
jax.grad(sum_squared_error, argnums=(0, 1))(x, y)
# (Array([-0.20000005, -0.19999981, -0.19999981, -0.19999981], dtype=float32),
# Array([0.20000005, 0.19999981, 0.19999981, 0.19999981], dtype=float32))
第五步:用 .at[] 替代就地赋值
副作用指函数产生了不出现在返回值里的效果,就地修改数组就是典型例子。在 NumPy 里这样写是合法的:
import numpy as np
x = np.array([1, 2, 3])
def in_place_modify(x):
x[0] = 123
return None
in_place_modify(x)
x # array([123, 2, 3])
把这个函数套到 JAX 数组上会直接抛出 TypeError,提示该对象不支持元素赋值,并建议改用 x.at[idx].set(y) 或其他 .at[] 方法。正确写法是:
x = jnp.array([5, 4, 3, 2])
# x[1] = 10 -> 抛出 TypeError
x = x.at[1].set(10)
print(x) # [5 10 3 2]
记住这条替换规则:凡是 x[idx] = y,都改写成 x = x.at[idx].set(y),或者使用 .at[] 下的其他方法。
第六步:手写线性回归训练循环
先造数据。真实关系为 y = w_true·x + b_true + ε,这里取 w_true = 3、b_true = −1,噪声标准差 0.1,样本 100 个:
import numpy as np
import matplotlib.pyplot as plt
xs = np.random.normal(size=(100,))
noise = np.random.normal(scale=0.1, size=(100,))
ys = xs * 3 - 1 + noise
plt.scatter(xs, ys)
模型为 ŷ(x; θ) = wx + b,把两个参数放进同一个数组 theta = [w, b]:
def model(theta, x):
"""Computes wx + b on a batch of input x."""
w, b = theta
return w * x + b
损失函数取均方误差 J(x, y; θ) = (ŷ − y)²:
def loss_fn(theta, x, y):
prediction = model(theta, x)
return jnp.mean((prediction - y) ** 2)
更新规则就是最朴素的 SGD:θ_new = θ − lr·∇θJ。因为 JAX 的梯度是对函数求的,更新函数可以写得非常短:
def update(theta, x, y, lr=0.1):
return theta - lr * jax.grad(loss_fn)(theta, x, y)
theta = jnp.array([1., 1.])
for _ in range(1000):
theta = update(theta, xs, ys)
plt.scatter(xs, ys)
plt.plot(xs, model(theta, xs))
w, b = theta
print(f"w: {w:.2f}, b: {b:.2f}")
# w: 3.00, b: -1.00
训练 1000 步后参数收敛到接近真实值。注意整个循环里没有任何“优化器对象”,参数就是普通数组,每次迭代由 update 返回新数组——这正是纯函数式风格在训练中的体现。
第七步:用 vmap 做自动向量化
写损失函数时经常要处理“单样本”逻辑,再套到 batch 上。手动维护 batch 维度容易出错,jax.vmap 可以把一个处理单样本的函数自动变成处理批量数据的函数。例如给二元交叉熵加 logits 的实现加上装饰器:
@jax.vmap
def binary_cross_entropy_with_logits(logits, labels):
logits = nn.log_sigmoid(logits)
return -jnp.sum(labels * logits + (1. - labels) * jnp.log(-jnp.expm1(logits)))
这样函数体里只写单个样本的数学表达式,批量维度由 vmap 负责,代码里不必到处携带 batch 维度。
第八步:用 Optax 替换手写更新
Optax 是配合 JAX 训练模型常用的优化库,同样以函数式方式使用:先构造优化器,初始化状态,每次迭代拿到更新量与新状态,再把更新量应用到参数上。
import optax
tx = optax.adam(learning_rate=0.03)
opt_state = tx.init(variables)
updates, opt_state = tx.update(grads, opt_state)
variables = optax.apply_updates(variables, updates)
Optax 覆盖的范围很广,从最基础的 SGD(optax.sgd(learning_rate, momentum))到 AdamW、Lion 等较新的变体都有。具体可用的优化器列表与参数默认值以官网当前信息为准。
第九步:用 Flax Linen 定义网络
Flax 在 JAX 生态里扮演 PyTorch 中 torch.nn、TensorFlow 中 tf.keras 的角色,提供基础神经网络层。导入方式:
import flax
import flax.linen as nn
下面用编码器/解码器结构演示。与 PyTorch 不同,这里不在 __init__ 里定义参数、也不单独写 forward,而是用一个 __call__ 函数以函数式风格封装全部逻辑:
class Encoder(nn.Module):
latents: int
@nn.compact
def __call__(self, x):
x = nn.Dense(500, name='fc1')(x)
x = nn.relu(x)
mean_x = nn.Dense(self.latents, name='fc2_mean')(x)
logvar_x = nn.Dense(self.latents, name='fc2_logvar')(x)
return mean_x, logvar_x
class Decoder(nn.Module):
@nn.compact
def __call__(self, z):
z = nn.Dense(500, name='fc1')(z)
z = nn.relu(z)
z = nn.Dense(784, name='fc2')(z)
return z
Linen 中定义子模块和变量有两种方式:
- 显式方式(setup):在
setup方法里把子模块或变量赋给self.<attr>,然后在类的任意前向方法中使用。这与 PyTorch 定义模块的方式接近。 - 内联方式(nn.compact):在标注了
@nn.compact的单个前向方法里直接写网络逻辑,子模块和变量就定义在使用它们的位置旁边。
内联方式的好处是:定义与使用挨在一起,不必上下翻找;遇到条件分支或循环时才定义子模块,可以减少重复代码;代码形态也更接近数学记号,比如 self.param('W', ...) @ x + self.param('b', ...) 读起来就像 y = Wx + b。
作为对照,用 setup 写一个 VAE 组合模块:
class VAE(nn.Module):
latents: int = 20
def setup(self):
self.encoder = Encoder(self.latents)
self.decoder = Decoder()
def __call__(self, x, z_rng):
mean, logvar = self.encoder(x)
z = reparameterize(z_rng, mean, logvar)
recon_x = self.decoder(z)
return recon_x, mean, logvar
def generate(self, z):
return nn.sigmoid(self.decoder(z))
第十步:用 tabulate 统计参数量
Flax 提供了一个很实用的能力:为任意网络生成参数摘要,入口是 flax.linen.Module 的 tabulate() 方法。以 28×28 的 MNIST 图像(即 x ∈ ℝ⁷⁸⁴)、潜变量 z ∈ ℝ¹⁰ 为例,上面那个简单编码器的参数量在 40 万左右,属于很小的网络:
encoder = Encoder(latents=10)
print(encoder.tabulate(jax.random.PRNGKey(0), jnp.ones((128, 784))))
输出是一张表格,逐层列出 path、module、inputs、outputs 与 params。可以看到 fc1 这一层是 Dense,输入 float32[128,784],输出 float32[128,500],参数为 bias float32[500] 与 kernel float32[784,500],合计 392,500 个(约 1.6 MB);fc2_mean 与 fc2_logvar 各自是 bias float32[10] 加 kernel float32[500,10],各 5,010 个(约 20.0 KB)。注意 tabulate 需要传入一个 PRNG key 和一份形状正确的假输入,用来推导各层形状。
一个完整示例
把前面的片段串起来,下面是一个从头到尾可运行的最小线性回归训练脚本,只依赖 jax、jax.numpy 与 numpy:
import numpy as np
import jax
import jax.numpy as jnp
# 1. 造数据:y = 3x - 1 + 噪声
rng = np.random.default_rng(0)
xs = rng.normal(size=(100,))
noise = rng.normal(scale=0.1, size=(100,))
ys = xs * 3 - 1 + noise
# 2. 模型:theta = [w, b]
def model(theta, x):
w, b = theta
return w * x + b
# 3. 损失:均方误差
def loss_fn(theta, x, y):
prediction = model(theta, x)
return jnp.mean((prediction - y) ** 2)
# 4. 单步 SGD 更新
def update(theta, x, y, lr=0.1):
return theta - lr * jax.grad(loss_fn)(theta, x, y)
# 5. 训练
theta = jnp.array([1., 1.])
for step in range(1000):
theta = update(theta, xs, ys)
w, b = theta
print(f"w: {w:.2f}, b: {b:.2f}")
print(f"final loss: {loss_fn(theta, xs, ys):.6f}")
运行后应看到 w 接近 3.00、b 接近 −1.00。如果想把优化器换成 Adam,只需把 update 里的手写公式替换为 Optax 的 tx.init / tx.update / optax.apply_updates 三步,参数数组本身不需要改结构。
注意事项
- 数组不可变:任何
x[idx] = y都会抛TypeError,必须写成x = x.at[idx].set(y)。这是从 NumPy 迁移过来最容易踩的坑。 - 不要写副作用:函数不应修改外部状态或入参,所有状态通过参数传入、通过返回值传出。JAX 的变换(grad、vmap、jit)都建立在这个前提上。
- 梯度是对函数求的:没有
loss.backward()。多参数函数默认只对第一个参数求梯度,其余参数要用argnums显式指定。 - 计时前先同步:JAX 运算异步派发,测性能时务必在表达式后加
.block_until_ready(),否则数字没有意义。 - 后端回退:找不到 GPU/TPU 时会打印警告并回退到 CPU,代码仍可运行,但速度差异明显。
- 浮点误差:梯度结果与解析解对照时可能出现末位差异(如 −0.20000005 对 −0.2),属正常现象。
- tabulate 需要假输入:调用
tabulate时必须提供 PRNG key 和形状正确的输入数组,否则无法推导各层形状。 - 版本与平台信息:安装方式、可用加速器、Optax 与 Flax 的接口细节都可能随版本变化,请以官网当前信息为准。