第 20 章 · 向量化运算：告别 for 循环

这一章要解决什么：把"逐支电芯算 SOH、逐个判断状态"的 for 循环，改成一行向量化代码。 打掉一个认知误区：* 不是矩阵乘法 —— 它是逐元素相乘。

本章你需要完成的代码：demos/ch20-vectorization.py

20.1 一行 vs 循环

import numpy as np

caps = np.array([2.50, 2.49, 2.47, 2.45, 2.43, 2.41, 2.39, 2.36, 2.34, 2.32])
cap_rated = 2.5

# 向量化：一行
pct = caps / cap_rated * 100
print(pct.round(2))
# [100.   99.6  98.8  98.   97.2  96.4  95.6  94.4  93.6  92.8]

# 等价的 for 循环：四行
loop_result = []
for c in caps:
    loop_result.append(round(c / cap_rated * 100, 2))

两者结果一致（可以用 np.allclose 验证），但向量化版本：

1. 更短：一行 vs 四行；
2. 更快：底层是编译好的 C 循环，跳过了 Python 逐元素的类型检查和函数调用；
3. 更容易读懂意图："整列除以额定容量再乘 100"，而不是"建个空列表，然后一个个……"

点开：到底快多少？（实测数据）

本课程示例的实测（在普通笔记本上跑 demos/ch20-vectorization.py）：

数据量 1,000,000
Python for 循环 :    48.6 ms
NumPy 向量化    :     1.9 ms
加速比          : 26 倍

快 26 倍。而且这个比值随数据量增大还会拉开：

数据量	for 循环	向量化	加速比
1 万	~0.5 ms	~0.02 ms	~25×
100 万	~50 ms	~2 ms	~26×
1 亿	~5 秒	~0.2 秒	~26×

为什么差距这么稳定：因为两者做的"算术次数"是一样的， 差距来自每个元素的开销：

Python for 循环的每个元素：
  ① 从列表取出对象（指针解引用）
  ② 检查类型（是 float 吗？）
  ③ 调用 __truediv__ 方法
  ④ 创建新的 float 对象（堆分配！）
  ⑤ 调用列表的 append

NumPy 的做法：
  ① 一整块内存交给 C 循环
  ② 循环体是 CPU 指令级别的加法/乘法（还能被 SIMD 向量化加速）
  ③ 结果直接写进预先分配好的内存

这就是"为什么 AI 必用张量"的核心原因： 矩阵乘法是 (1000, 1000) × (1000, 1000) 这种规模， 如果按元素走 Python 循环，一次前向传播要跑几分钟；用 BLAS 库的矩阵乘法只要几十毫秒。

20.2 四则运算都是"逐元素"的

a = np.array([1.0, 2.0, 3.0])
b = np.array([10.0, 20.0, 30.0])

print(a + b)     # [11. 22. 33.]
print(a - b)     # [ -9. -18. -27.]
print(a * b)     # [10. 40. 90.]     ← 逐元素乘！
print(a / b)     # [0.1 0.1 0.1]
print(a ** 2)    # [1. 4. 9.]
print(a @ b)     # 140.0             ← 点积（矩阵乘法）

点开：* 和 @ 的区别（最容易搞错的一个）

数学里的 $AB$（矩阵乘法）在 NumPy 里要写 a @ b，不是 a * b。

A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

print(A * B)         # 逐元素乘（Hadamard 积）
# [[ 5 12]
#  [21 32]]

print(A @ B)         # 矩阵乘法
# [[19 22]
#  [43 50]]

print(np.matmul(A, B))   # 完全等价于 A @ B

手算验证矩阵乘法：

A @ B 的第 (0,0) 个元素 = 1×5 + 2×7 = 5 + 14 = 19   ✅
A * B 的第 (0,0) 个元素 = 1×5 = 5                    ✅

为什么这个坑很危险： 在三维数组上，A * B 和 A @ B 都能算出结果、都不报错，但结果完全不同。 如果你在实现某个公式时搞混了，可能要等到结果明显异常才发现。

在 AI 里的对应：

操作	NumPy	PyTorch
逐元素乘	a * b	a * b
矩阵乘法	a @ b / np.matmul	a @ b / torch.matmul
批量矩阵乘	np.einsum	torch.bmm / torch.einsum

@ 的由来：Python 3.5 专门为矩阵乘法引入了这个运算符（PEP 465）， 因为 * 已经被"逐元素"占用了 —— 而逐元素在数值计算里同样常用。

一个记忆点：@ 看起来像"矩阵的网格"，所以它是矩阵乘法。

20.3 NumPy 标量的打印（2.x 的新变化）

print(round(np.float64(100.0), 1))      # 100.0
print(float(round(np.float64(100.0), 1)))   # 100.0

看起来一样？ 在交互式环境（REPL）里会有区别：

>>> import numpy as np
>>> np.float64(100.0)
np.float64(100.0)          # NumPy 2.x 会在 repr 里带前缀
>>> float(np.float64(100.0))
100.0

什么时候需要注意：写日志、拼字符串、生成报告时， 如果直接把 NumPy 标量插进 f-string，可能得到 np.float64(100.0) 这种输出。

修复：用 float() / int() 包一层，或者用 f-string 的格式符（{x:.2f} 会自动转）。

x = np.float64(99.5)
print(f"{x}")          # 99.5         ← f-string 默认 str()，通常没问题
print(f"{x:.1f}")      # 99.5         ← 格式符更安全
print(f"{x!r}")        # np.float64(99.5)   ← !r 用 repr，会暴露前缀

本书示例里凡是要打印数组元素，都用 .round(n) 或 float() —— 这样输出干净、可读。

20.4 比较也是"整列一次做完"

pct = caps / cap_rated * 100

mask = pct < 95
print(mask)
# [False False False False False False False  True  True  True]

print(mask.sum())       # 3        个数（True 当 1）
print(mask.mean())      # 0.3      占比
print(pct[mask])        # [94.4 93.6 92.8]     取满足条件的值

关键认知：pct < 95 返回的是一个布尔数组，不是单个 True/False。

所以这句话是错的：

if pct < 95:            # ❌
# ValueError: The truth value of an array with more than one element is ambiguous

想判断"整列都满足"，用 .all()；"至少一个满足"，用 .any()：

print((pct < 100).all())     # True   （所有元素都 < 100）
print((pct < 95).any())      # True   （至少有一个 < 95）

这三个方法（sum / mean / all / any）是布尔数组的标准工具箱， 后面在 Pandas 里会用得更多。

20.5 np.where：向量化的 if/else

for + if 的向量化版本：

status = np.where(pct < 95, "预警", "健康")
print(status)
# ['健康' '健康' '健康' '健康' '健康' '健康' '健康' '预警' '预警' '预警']

签名：np.where(条件, 条件为真时的值, 条件为假时的值)

三档就用嵌套（和 if/elif/else 一样，顺序要从严格到宽松）：

three = np.where(pct < 90, "待退役",
                 np.where(pct < 95, "预警", "健康"))

层数多了用 np.select：

three = np.select(
    [pct < 90, pct < 95],            # 条件列表（按顺序判断）
    ["待退役", "预警"],                # 对应的值
    default="健康",                   # 都不满足时的值
)

点开：np.where 的两个陷阱

陷阱一：两个分支都会被算出来。

result = np.where(x > 0, np.log(x), 0)
#                            ^^^^^^^^ 对 x <= 0 的元素，log 会算出 -inf 或 nan
#                                     虽然最后没被选中，但过程中已经算了（并会打印警告）

修复：先用 np.errstate 忽略警告，或者先构造好输入：

safe_x = np.where(x > 0, x, 1)        # 把不合法的值换成 1
result = np.where(x > 0, np.log(safe_x), 0)

陷阱二：顺序写反不报错，只给错结果。

# ❌ 先判 < 95，0.78 也满足它，永远轮不到"待退役"
wrong = np.select([pct < 95, pct < 90], ["预警", "待退役"], default="健康")

# ✅ 从严格到宽松
right = np.select([pct < 90, pct < 95], ["待退役", "预警"], default="健康")

这和 Python 的 if/elif 是同一个坑（第 08 章）： elif 只在前面的条件都不成立时才求值；np.select 只取第一个为 True 的条件。

验证方法：拿几个边界值试一下。

test = np.array([0.78, 0.84, 0.92])
print(np.select([test < 90, test < 95], ["待退役", "预警"], default="健康"))
# ['待退役' '预警' '健康']    ✅

20.6 原地修改：+ vs +=

x = np.array([1.0, 2.0, 3.0])

y = x / 2
print(x, y)      # [1. 2. 3.] [0.5 1.  1.5]   ← x 没变（产生新数组）

x /= 2
print(x)         # [0.5 1.  1.5]              ← 原地修改了 x

规则和第 19 章的视图是一套逻辑：

写法	效果
y = x / 2	产生新数组，x 不变
x /= 2	原地修改 x（等价于 x = x / 2 但复用内存）
y = x	y 和 x 是同一个对象（第 03 章讲的"标签"）

为什么要有原地版本：对 10 GB 的数组，x = x / 2 需要同时驻留两份（旧的 + 新的）， x /= 2 只需要一份。显存/内存紧张时这是救命的。

对应的函数版本：很多 NumPy 函数有 out= 参数。

a = np.ones(3)
np.add(a, 1, out=a)          # 结果写回 a，不新建数组
print(a)                     # [2. 2. 2.]

在 AI 里：PyTorch 用下划线后缀表示原地版本（add_ / mul_ / zero_）， 比 out= 参数更显式、更不容易漏。

20.7 但这不是"处处向量化"

向量化的代价：中间结果会额外占内存。

result = a * b + c * d       # 中间产生了 a*b 和 c*d 两个临时数组

对超大数组，这个"临时数组"可能就是几百 MB。

三种应对：

# ① 用 out= 复用一个数组
out = np.empty_like(a)
np.multiply(a, b, out=out)
np.add(out, c, out=out)

# ② 分块处理
for i in range(0, len(a), chunk_size):
    a[i:i + chunk_size] = a[i:i + chunk_size] * b[i:i + chunk_size]

# ③ 用 numexpr 之类的库（编译时优化表达式）

但更重要的原则：

能用向量化就用，读不懂就退回循环。正确 > 快。

如果一个向量化表达式你写完之后自己要花 30 秒才看懂， 那它的价值就大打折扣了 —— 三个月后的你，和一个要接手你代码的同事，都会感谢那个"啰嗦但清晰"的 for 循环。

判断标准：

情况	建议
简单的整列算术（a / b * 100）	✅ 向量化
元素间有依赖（斐波那契、累计递推）	❌ 用循环（或 np.cumsum 这类专用函数）
需要 try/except 逐元素处理	❌ 用循环
需要在循环里打印/写文件	❌ 用循环
复杂的多分支逻辑（> 2 层）	⚠️ np.select 或回到循环

20.8 完整示例与输出

cd code
python demos/ch20-vectorization.py

真实输出（节选）：

=== ② 四则运算都是「逐元素」的 ===
  a * b  = [10. 40. 90.]   ← 逐元素乘，不是矩阵乘法！
  a @ b  = 140.0   ← 点积/矩阵乘法用 @ 或 np.matmul
  这是最容易搞错的地方：数学的 AB 要写 a @ b，a * b 是「对应位置相乘」。

=== ④ 比较也会「整列一次做完」 ===
  pct < 95  → [False False False False False False False  True  True  True]
  这是布尔数组，不是 True/False 单一值（所以不能写 if pct < 95:）
  满足条件的个数：mask.sum() = 3   ← True 当 1，False 当 0

=== ⑤ np.where：向量化的 if/else ===
  np.where(pct < 95, '预警', '健康')
  → ['健康' '健康' '健康' '健康' '健康' '健康' '健康' '预警' '预警' '预警']

=== ⑦ 到底快多少（实测） ===
  数据量 1,000,000
  Python for 循环 :    48.6 ms
  NumPy 向量化    :     1.9 ms
  加速比          : 26 倍

本章练习

练习 1：把 for 循环改成向量化

给定 10 支电芯的 SOH，用向量化（不许写 for）完成：

1. 整列生成状态（< 0.94 → 待退役，否则健康）；
2. 数出待退役支数与占比；
3. 把 SOH 换算成百分比、保留 1 位小数。

再用 for 循环写一遍，验证两者结果一致。

参考答案

import numpy as np

soh = np.array([1.0, 0.996, 0.988, 0.98, 0.972, 0.964, 0.956, 0.944, 0.936, 0.928])

# ① 向量化的 if/else
status = np.where(soh < 0.94, "待退役", "健康")
print(status)
# ['健康' '健康' '健康' '健康' '健康' '健康' '健康' '健康' '待退役' '待退役']

# ② 计数与占比：布尔数组的两个"魔法"
n_eol = (soh < 0.94).sum()
print(n_eol)                        # 2
print(f"{(soh < 0.94).mean():.1%}") # 20.0%
# 等价写法：(status == "待退役").sum()

# ③ 百分比
pct = (soh * 100).round(1)
print(pct)   # [100.  99.6  98.8  98.   97.2  96.4  95.6  94.4  93.6  92.8]

# 用 for 验证
status_loop = np.array(["待退役" if s < 0.94 else "健康" for s in soh])
print(np.array_equal(status, status_loop))     # True

三个要点：

1. np.where(条件, 真值, 假值) 就是向量化版的 if/else， 两个分支都会被算出来，最后按条件选一个。
2. 布尔数组的 .sum() 是计数、.mean() 是占比 —— 这是全书最常用的两个小技巧。
3. np.array_equal / np.allclose 用来验证两种写法一致。 养成"改写成向量化后，用旧写法验证一遍"的习惯。

加分：三档

three = np.select([soh < 0.94, soh < 0.97], ["待退役", "预警"], default="健康")
print(three)
# ['健康' '健康' '健康' '健康' '预警' '预警' '预警' '待退役' '待退役' '待退役']

注意条件要从严格到宽松排列 —— 和 if/elif 是同一个坑。

完整练习：exercises/ex20-01-vectorization

练习 2：这对代码结果一样吗？

import numpy as np

a = np.array([1.0, 2.0, 3.0])
b = np.array([10.0, 20.0, 30.0])

print(a * b)
print(a @ b)

要求：先手算两个结果，再运行验证。然后回答：一维数组的 @ 是什么运算？

参考答案

print(a * b)     # [10. 40. 90.]       逐元素相乘
print(a @ b)     # 140.0               点积（内积）

手算过程：

a * b  = [1×10, 2×20, 3×30] = [10, 40, 90]

a @ b  = 1×10 + 2×20 + 3×30
       = 10 + 40 + 90
       = 140

一维数组的 @ 就是点积（内积） —— 对应位置相乘再求和。

升到二维：

A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

print(A * B)
# [[ 5 12]
#  [21 32]]             逐元素

print(A @ B)
# [[19 22]
#  [43 50]]             矩阵乘法

手算矩阵乘法：

A @ B 的第 0 行第 0 列 = 1×5 + 2×7 = 19
A @ B 的第 0 行第 1 列 = 1×6 + 2×8 = 22
A @ B 的第 1 行第 0 列 = 3×5 + 4×7 = 43
A @ B 的第 1 行第 1 列 = 3×6 + 4×8 = 50

为什么这个区别在 AI 里至关重要：

神经网络的一层就是 y = x @ W + b：

• x 形状 (batch, in_features)
• W 形状 (in_features, out_features)
• x @ W 形状 (batch, out_features)

如果误写成 x * W，形状不匹配会直接报错（这种情况反而安全）； 但如果某些维度恰好相等（比如 (3, 3) * (3, 3)），就会静默算错。

结论：写任何矩阵运算时，先确认 * 和 @ 该用哪个， 再用一个小的手算例子验证。这是 AI 工程里值得养成的习惯。

顺带：PyTorch 里同样有 *（逐元素）和 @（矩阵乘），语义完全一致。 额外的还有 torch.bmm（批量矩阵乘）、torch.einsum（爱因斯坦求和，能表达任意收缩）。

练习 3：什么时候该用循环

下面四个任务，哪些适合向量化，哪些应该用 for 循环？说明理由。

1. 把 100 万个容量值从 Ah 换算成 mAh；
2. 按顺序模拟电池退化，每一圈的状态依赖上一圈；
3. 逐个检查 100 个文件，能读的就处理、读不了的就记录下来；
4. 把一列 SOH 按 < 0.85 / < 0.90 / 其余 分成三档。

参考答案

任务	选择	理由
1. 单位换算	向量化	caps * 1000 一行搞定，纯逐元素、无依赖
2. 递推模拟	for 循环	元素间有依赖 —— 第 n 圈依赖第 n-1 圈，无法并行
3. 逐文件容错处理	for 循环	需要 try/except 和 I/O 副作用
4. 分档	向量化	np.select 一行搞定

判断标准（按重要性排序）：

① 元素之间有依赖吗？ → 有 → 必须循环

# 容量递减是递推的：下一圈 = 上一圈 × 衰减因子
cap = 2.5
for i in range(1000):
    cap = cap * 0.9995            # 每步依赖上一步，无法向量化
    caps.append(cap)

这类问题在 NumPy 里有专用函数（不用手写循环）：

np.cumsum(arr)          # 累加
np.cumprod(arr)         # 累乘（电池衰减就是它）
np.diff(arr)            # 相邻差

② 需要 try/except 或 I/O 吗？ → 需要 → 必须循环

③ 逻辑超过两层分支吗？ → 是 → 建议循环（可读性优先）

④ 都没有？ → 向量化

一个真实的权衡例子：

# 向量化版本：一行，但需要理解 np.select 的参数顺序
df["health"] = np.select([df["soh"] < 0.85, df["soh"] < 0.90],
                         ["待退役", "预警"], default="健康")

# 循环版本：五行，但任何人一眼能看懂
health = []
for s in df["soh"]:
    if s < 0.85:
        health.append("待退役")
    elif s < 0.90:
        health.append("预警")
    else:
        health.append("健康")
df["health"] = health

对 1.4 万行数据，两者的耗时差异不到 10 毫秒 —— 此时可读性应该赢。

向量化真正不可替代的场景是"数据量大到循环跑不动"： 上百万行、图像张量、矩阵乘法。在这些场景下没有选择。

所以完整的判断链是：

元素有依赖吗？ → 有 → 循环（或找专用函数）
  ↓ 没有
数据量大到循环受不了吗（> 百万级）？ → 是 → 向量化
  ↓ 不是
向量化写法我能一眼看懂吗？ → 能 → 向量化
  ↓ 不能
用循环（可读性优先）

面向 AI 开发：向量化就是"GPU 编程"的思想基础

这一章讲的是 CPU 上的向量化，但思路完全相同于 GPU 编程。

GPU 为什么快：它有几千个核心，可以同时算几千个数。

CPU：8~16 个强核心，适合"串行 + 复杂分支"
GPU：几千个弱核心，适合"同一个操作，作用在几万个数据上"

而"同一个操作作用在一整块数据上"，正是向量化在做的事。

# 这一行代码在 GPU 上会变成：几千个核心各算一部分
y = x * 2 + 1

这就是为什么"写 for 循环"在深度学习里是致命的：

# ❌ 每个样本单独过一次模型，GPU 利用率接近 0
for i in range(len(dataset)):
    y = model(dataset[i])

# ✅ 一次喂一批，GPU 跑满
y = model(batch)          # batch 形状 (32, ...)

"batch 维度"存在的根本原因就是向量化： 把 32 个样本堆成一个 (32, ...) 的张量，让一次矩阵乘法处理完 32 个样本 —— 而不是循环 32 次。

再深一层：为什么"逐元素运算"能并行

y[i] = x[i] * 2 + 1        # 每个 i 的计算互相独立

互相独立是关键 —— 意味着可以任意调度到不同核心上，没有先后顺序要求。 GPU 的调度器就是干这个的。

而"有依赖"的运算就不能这么干：

y[i] = y[i-1] * 0.999       # y[5] 依赖 y[4]，依赖 y[3]…… 只能串行

这类计算在 GPU 上要靠"扫描算法"（parallel prefix sum）来加速， 复杂得多，也是 np.cumsum 这类函数比手写循环快得多的原因。

三个能直接迁移到 AI 的习惯：

① 看到嵌套循环就想"能不能变成矩阵乘法"。

# 手工算 1000 个样本两两之间的欧氏距离：O(n²) 次循环
for i in range(1000):
    for j in range(1000):
        dist[i, j] = np.sqrt(((X[i] - X[j]) ** 2).sum())

# 向量化：一次矩阵运算
# ||a - b||² = ||a||² + ||b||² - 2·a·b
d2 = (X ** 2).sum(1)[:, None] + (X ** 2).sum(1)[None, :] - 2 * X @ X.T
dist = np.sqrt(np.maximum(d2, 0))

第二版用了广播（第 21 章）和矩阵乘法，在 1000 个样本上快 100 倍以上。

② 检查你的 batch 维度是不是"用上了"。

如果模型接受的输入是 (32, 784)，而你在外面套了个循环一次喂 (1, 784)， 那你浪费了 32 倍的吞吐。这是 AI 工程里最常见的一类性能问题。

③ num_workers / pin_memory 这类参数，本质也是"让数据准备别拖后腿"。 向量化解决了"算得慢"，数据加载的并行化解决"喂得慢"——两者都是为了让 GPU 不空转。

---

下一章：第 21 章 · 广播机制：不同形状怎么一起算