一、数据一个一个来，意味着什么

设定是这样的：第 t 步来一个样本 x_t，你先给出预测，然后才拿到真标签 y_t， 拿到之后允许改一次权重，改完这个样本就丢掉。 内存里只能留一个当前权重，不能留历史样本。 这个限制听起来像工程细节，其实是数学上的硬约束。 批处理方法的每一次更新都要把全部样本过一遍，在线方法没有这个条件， 所以它的每一步只能看一个样本——能看到的信息少了，理论上限就低了。 衡量指标因此换成两个： 一是累计错误数，就是整个过程一共判错多少次。 二是后悔值，等于「自己的累计错误」减去「事后挑一个最好的固定权重会犯的错误」。 后悔值衡量的是「因为不知道未来而多付的代价」，它才是在线学习真正要比的东西。 注意「事后挑一个最好的固定权重」这件事，现实中做不到——那需要先看完所有数据。 它只是一个基准，用来给在线算法打分。 本节的 Demo 里，这个基准用批量铰链损失训练得到，它把 400 个样本反复看了 400 遍。 还要提前说清一件事，免得后面看着别扭：在线学习的错误率不会一路降到很低。 数据本身带噪声，就算拿到真权重，也会有一批样本注定判错。 实测 400 步跑完错误率停在 0.383， 而事后最优固定权重是 0.265——这中间的 0.118 就是在线的天花板。

二、那一步是怎么算出来的

想法只有两句： 第一，希望更新之后这个样本被分对，而且不只分对， 还要留出一点余量——要求 y_t 乘上 w 与 x_t 的内积，至少等于 1。 第二，在满足这个要求的所有权重里，挑离原来那个最近的一个。 写成式子就是：在约束 y_t (w · x_t) 大于等于 1 之下，让 w 与旧权重的欧氏距离最小。 这是一个带一个线性约束的二次规划，解出来是一个闭式：

m   = y_t * (w_t . x_t)          # 当前间隔
tau = (1 - m) / (x_t . x_t)      # 要走的步长
w   = w_t + tau * y_t * x_t      # 只沿 x_t 方向动

为什么是除以 x_t 自己的长度平方？ 因为你只沿 x_t 方向走，走多远才能让内积增加 1，取决于 x_t 有多长。 x_t 长一倍，同样走一步对内积的贡献就大一倍，所以步长要按长度的平方反比缩。 这也解释了一个现象：特征没标准化的时候，长的那一列会主导步长，短的那一列几乎推不动。 几何上更清楚：把权重沿着 x_t 的方向推到「分界面刚好越过这个样本 1 个单位」的位置， 垂直于 x_t 的那个分量完全不动。 所以 PA 的每次更新都是「只在自己被质疑的那个方向上退让，别的一概不改」。 两个词就是这么来的： 被动，指间隔已经够 1 的时候一步都不走。 攻击，指间隔不够的时候要动，而且动到刚好够为止，不多走一丝。 Demo 第一段把这件事演到极致：同一个方向的样本连来八次。

三、三种数据下的对局

同一份数据流，三种噪声水平，各跑 400 步：

四、步长封顶：唯一要调的旋钮

不封顶的 PA 有一个致命弱点：遇到一个翻错的标签，它也会老老实实去满足那个假约束。 如果这个错标签的样本恰好很长，步长会被放大成一大步，一次就把好不容易学到的方向带偏。 补救的办法是给步长加个上限，这就是 PA-I：

PA      : tau = (1 - m) / (x . x)
PA-I    : tau = min(C, (1 - m) / (x . x))
PA-II   : tau = (1 - m) / ((x . x) + 1 / (2C))

PA-I 是硬性截断，PA-II 是软化——把 1 除以 2C 这一项加进分母，样本很长的时候效果和截断差不多， 样本很短的时候又不会过度惩罚。 翻错两成标签、十个种子平均下来：

五、错误率不会一路降到零

很多人第一次跑在线算法，会盯着错误率曲线看它降不降，然后怀疑自己写错了。

六、动手 Demo：间隔够了就一步不动，不够就只走一小步

这段代码做四件事：先让同一个方向的样本连来八次，看步长怎么从 1 掉到 0； 再在三种噪声水平下和批量铰链损失对局；然后把步长上限 C 从 0.1 扫到 10； 最后看累计错误率到底降不降。 重点看两处：第一段里第二次之后的步长为什么全是 0.0000； 以及封顶表里 C 等于 10 时数字和「不封顶」一模一样。

import numpy as np


def stream(n=400, d=5, seed=1, noise=0.6, flip=0.0):
    """逐样本到来的二分类数据流：真权重只看前两个坐标。"""
    rg = np.random.default_rng(seed)
    w = np.array([1.0, -0.6, 0.0, 0.0, 0.0])
    w = w / np.linalg.norm(w)
    X, Y = [], []
    for t in range(n):
        x = rg.normal(0.0, 1.0, d)
        y = 1.0 if float(w @ x) + rg.normal(0.0, noise) > 0 else -1.0
        if flip > 0.0 and rg.uniform() < flip:
            y = -y
        X.append(x)
        Y.append(y)
    return np.array(X), np.array(Y)


def pa(X, Y, C=1e6, mode='I'):
    """被动攻击：间隔不够才动，动就动让间隔恰好等于 1 的那一小步。"""
    n, d = X.shape
    w = np.zeros(d)
    err, taus, hist = 0, [], []
    for t in range(n):
        m = Y[t] * float(w @ X[t])
        if m <= 0.0:
            err += 1
        hist.append(err)
        tau = 0.0
        if m < 1.0:
            xx = float(X[t] @ X[t])
            if mode == 'II':
                tau = (1.0 - m) / (xx + 1.0 / (2.0 * C))
            else:
                tau = min(C, (1.0 - m) / xx)
            w = w + tau * Y[t] * X[t]
        taus.append(tau)
    return err, np.array(taus), np.array(hist)


def batch_hinge(X, Y, lam=1e-3, iters=400, lr=0.5):
    """批量铰链损失加 l2 正则，当作「看得见全部数据」的离线基准。"""
    n, d = X.shape
    w = np.zeros(d)
    for _ in range(iters):
        g = lam * w
        for t in range(n):
            if Y[t] * float(w @ X[t]) < 1.0:
                g = g - Y[t] * X[t] / n
        w = w - lr * g
    return w


def err_of(w, X, Y):
    return int(np.sum((X @ w) * Y < 0))


print('[1] 同一个方向的样本连来八次，看步长怎么变')
w = np.zeros(2)
x = np.array([1.0, 0.0])
for k in range(1, 9):
    m = float(w @ x)
    tau = 0.0 if m >= 1.0 else (1.0 - m) / float(x @ x)
    w = w + tau * x
    print('  第 %d 次  间隔 %+.4f  步长 %.4f  w = [%.4f %.4f]'
          % (k, m, tau, w[0], w[1]))

print('')
print('[2] 三种数据下单次对局：PA 对批量铰链')
for tag, kw in (('可分', dict(noise=0.2)), ('有噪', dict(noise=1.2)),
                ('翻转20%', dict(noise=0.6, flip=0.2))):
    X, Y = stream(400, 5, 1, **kw)
    e, taus, hist = pa(X, Y)
    wb = batch_hinge(X, Y)
    print('  %-9s PA %3d  批量铰链 %3d  差 %3d  更新 %3d/400  步长中位 %.4f'
          % (tag, e, err_of(wb, X, Y), e - err_of(wb, X, Y),
             int((taus > 0).sum()), np.median(taus)))

print('')
print('[3] 十个种子平均：步长封顶放在哪里（翻 20% 标签）')
for nm, kw in (('PA', dict(C=1e6, mode='I')),
               ('PA-I C=0.1', dict(C=0.1, mode='I')),
               ('PA-I C=1', dict(C=1.0, mode='I')),
               ('PA-II C=0.1', dict(C=0.1, mode='II')),
               ('PA-II C=1', dict(C=1.0, mode='II')),
               ('PA-II C=10', dict(C=10.0, mode='II'))):
    v = []
    for sd in range(1, 11):
        X, Y = stream(400, 5, sd, noise=0.6, flip=0.2)
        v.append(pa(X, Y, **kw)[0])
    print('  %-12s 平均 %6.1f  最好 %3d  最差 %3d' % (nm, np.mean(v), min(v), max(v)))

print('')
print('[4] 累计错误走到哪一步（有噪，seed=1）')
X, Y = stream(400, 5, 1, noise=1.2)
e, taus, hist = pa(X, Y)
for t in (50, 100, 200, 300, 400):
    print('  t=%3d  累计错误 %3d  错误率 %.3f' % (t, hist[t - 1], hist[t - 1] / (t + 0.0)))
wb = batch_hinge(X, Y)
print('  事后最优固定权重的错误 %d，PA 多错 %d'
      % (err_of(wb, X, Y), e - err_of(wb, X, Y)))

实跑输出：

--------------------------------------------------------------------
被动攻击：间隔够了就一步不动，不够就只走一小步
--------------------------------------------------------------------
[1] 同一个方向的样本连来八次，看步长怎么变
  第 1 次  间隔 +0.0000  步长 1.0000  w = [1.0000 0.0000]
  第 2 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 3 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 4 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 5 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 6 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 7 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]
  第 8 次  间隔 +1.0000  步长 0.0000  w = [1.0000 0.0000]

[2] 三种数据下单次对局：PA 对批量铰链
  可分        PA  37  批量铰链  23  差  14  更新 100/400  步长中位 0.0000
  有噪        PA 153  批量铰链 106  差  47  更新 249/400  步长中位 0.1350
  翻转20%     PA 158  批量铰链 115  差  43  更新 252/400  步长中位 0.1480

[3] 十个种子平均：步长封顶放在哪里（翻 20% 标签）
  PA           平均  166.7  最好 153  最差 185
  PA-I C=0.1   平均  142.1  最好 128  最差 149
  PA-I C=1     平均  165.2  最好 158  最差 176
  PA-II C=0.1  平均  154.7  最好 145  最差 162
  PA-II C=1    平均  164.2  最好 153  最差 184
  PA-II C=10   平均  166.7  最好 153  最差 185

[4] 累计错误走到哪一步（有噪，seed=1）
  t= 50  累计错误  21  错误率 0.420
  t=100  累计错误  42  错误率 0.420
  t=200  累计错误  70  错误率 0.350
  t=300  累计错误 115  错误率 0.383
  t=400  累计错误 153  错误率 0.383
  事后最优固定权重的错误 106，PA 多错 47

第一段是整节的核心，八个样本全是同一个方向 x 等于 [1, 0]。 第一步间隔是 0，步长 1.0000，权重一步变成 [1, 0]。 第二步间隔恰好 1.0000，条件「小于 1」不成立，于是步长 0.0000——之后七次一模一样。 如果换成感知机，这个故事会完全不同：感知机只看「分对没有」，间隔 1.0000 早就分对了， 也不会动。 两者在这里恰好一致，但换成间隔 0.3 的样本就分道扬镳——感知机不动，PA 会补上那 0.7 的余量。 第二段三种数据，PA 分别是 37、153、158，批量铰链是 23、106、115。 差距 14、47、43。 注意可分数据上 PA 只更新了 100 次，步长中位数 0.0000——四百次里三百次一声不吭， 这就是「被动」。 第三段是这一节最实用的一张表。 不封顶 166.7，封在 0.1 是 142.1，封在 1 是 165.2，封在 10 是 166.7。 C 等于 10 和不封顶完全同数，说明这个上限一次都没被触发过——封顶要封到「能真的截住」才有意义。 第四段那五个数 21、42、70、115、153 对应的错误率是 0.420、0.420、0.350、0.383、0.383。 它没有一路下降，最后停在 0.383。 而事后最优固定权重只错 106 个，PA 多错 47 个——这个 47 就是后悔值。 代码里有一处值得多说一句：错误是按「更新之前」那次预测统计的。 如果先更新再判错，刚被纠正的样本会显示成「预测正确」， 实测的错误数会直接变成 0——这是实现在线算法最容易踩的坑。

30 秒实验

1. 把 stream 的 noise 从 1.2 改成 3.0，重跑第二段。 看 PA 和批量铰链各自错多少，再看那个差值变大还是变小。 噪声越大，两者越接近——因为噪声占了大头，看不看得见未来就不那么重要了。
2. 把 pa 里的 if m < 1.0 改成 if m < 0.5。 这等于把「够用的余量」从 1 降到 0.5，更新次数会明显减少。 看错误数是升还是降——这一步决定了余量这个超参数该往哪边调。
3. 把 batch_hinge 的 iters 从 400 改成 1，让它只扫一遍。 这样它和 PA 的信息量就一样了，再看谁错得少。 这个对照能回答一个关键问题：PA 的差距到底来自「只扫一遍」，还是来自算法本身。

常见误区

1. 「PA 是感知机的改进版，所以错误数一定更少」。 它改进的是「更新多少」，不是「一定能赢」。 在可分数据上实测 PA 错 37 个、事后最优错 23 个——在线算法输给批处理是常态，赢才是意外。
2. 「步长上限 C 越大越好，越大越接近理论最优」。 实测正好相反：C 等于 0.1 时平均错 142.1，C 等于 1 时 165.2， C 等于 10 时和不封顶完全一样（166.7）。 C 大等于允许单次走很远，而错标签给出的方向本来就是错的。
3. 「在线学习跑得越久越准，错误率会一路降到很低」。 错误率停在 0.383，而事后最优固定权重是 0.265——中间那 0.118 是噪声和信息量共同定下的地板， 跑多久都下不去。 看在线曲线要用滑动窗口，累计平均会掩盖模型正在变差。
4. 「在线学习就是单遍的随机梯度下降」。 SGD 仍然假设样本同分布、可以多轮扫，只是每轮顺序打乱； 在线学习是真正单遍、不可回头，而且分布可能中途变化。 PA 的步长由约束解出来，和 SGD 的固定学习率也不是一回事。

PA 有一个隐含的假设：所有方向都同样不确定，所以它每次只沿 x 走， 从来不考虑「这个方向我已经很熟了」。 下一节给它配一个协方差矩阵，让它知道自己对哪个方向有把握、对哪个方向没底。

• ☐ （本节完，下一节 15.2 15.2）

相关笔记

• 适应正则化学习
• 使用 Hinge 损失最小化学习来解释
• 0/1 损失和间隔
• 🤖 图解机器学习 · 索引