测量逐条与批量预测的延迟、分位数和吞吐量

同一个模型，逐条预测与一次处理一批数据，往往会给出差别很大的“每条耗时”。要让性能数字有意义，先说明计时范围：测的是一次请求，还是一批请求摊到每个样本的成本？是否包括预处理、网络和排队？分位数来自哪些重复观测？

scikit-learn 的 Prediction Latency 示例提供了一个从生成数据到绘图的完整框架。本文按该示例的全部技术环节整理：数据准备、模型训练、逐条和批量计时、特征维数扫描、短窗口吞吐量。2026-10-05 实读页面显示 1.9.1。下文给出明确标注差异的计时修订版；本次只做静态审核，没有运行基准，也不提供虚构的性能数字。

先建立可比较的输入

原例用 make_regression 生成合成回归数据，默认配置为 1,000 条训练样本、100 条测试样本、100 个特征，噪声为 0.1。先划分训练集和测试集，再对训练集打乱顺序。StandardScaler 只在训练特征上拟合，并把相同变换应用于测试特征；目标 y 也用另一套 scaler 同样处理。这样避免让测试集统计量进入训练预处理。

这里的计时不包括数据生成、划分、缩放或拟合，只围绕已经训练好的模型的 predict。输入还保留了测试标签，但原性能示例没有把预测误差作为比较条件。因此它展示的是调用成本，不能单独得出哪种模型最适合业务；实际选型必须同时比较相同任务上的预测质量。

原配置包含三个回归器：带 elasticnet 惩罚的 SGDRegressor，参数为 alpha=0.01、l1_ratio=0.25、tol=1e-4；默认 RandomForestRegressor；以及 RBF 核的 SVR。原图标签分别显示非零系数个数、森林中的估计器数量和支持向量数量，帮助读者把模型大小与成本联系起来。

逐条模式：一次 predict 就是一条观测

逐条模式遍历测试集，用 X_test[[i], :] 取得形状为 (1, n_features) 的二维数组，再单独调用 predict。双层方括号保留样本维，避免把一维向量误传给需要二维输入的估计器。每个样本得到一条耗时记录。

原例把取出样本的操作放在计时之前，因此这个值包含 predict 内部的输入检查和计算，但不含循环外的数据准备，也不含切片。100 个测试样本会产生 100 条观测，箱线图显示这些观测的分布。这里的 p90 是这组单样本调用耗时的第 90 百分位，不应自动等同于线上服务的 p90。

批量模式：整批耗时除以样本数

批量模式重复调用 predict(X_test)，原例默认重复 30 次。每次计时得到整批 100 个样本的耗时，然后除以 100，形成“每样本摊销成本”。它有助于观察批处理减少固定调用开销、利用向量化或底层计算库的效果。

注意，图里一个批量观测点对应一次完整批处理，不对应其中某个样本真实等待了多久。因此批量摊销 p90 的含义是“30 次整批调用的单位样本摊销时间的 p90”，不能把它宣传为单个在线请求的尾延迟。如果业务需要凑够一批才预测，还必须把组批等待时间计入服务指标。

保持时间单位和图表含义一致

两个基础计时函数都返回以秒为单位的数组。原例在绘图前乘以 1e6 转成微秒，并分别生成逐条与批量箱线图，纵轴是每样本预测时间。箱线图中的中位数、四分位范围、须和离群点能够提示分布偏斜及调度噪声，但它们并不代替重复实验或系统层面的诊断。

不要只抄一张图的纵轴读数；至少一起记录硬件、操作系统、Python 与库版本、CPU 线程设置、模型参数、数据维度、批大小和是否预热。底层 BLAS/OpenMP 的线程数不一定由估计器的 n_jobs 控制，所以把随机森林设成单线程，也不代表整段 Python 程序只有一个计算线程。

修订后的完整计时示例

下面保留原例的完整实验结构，改写辅助函数以便说明计时边界。与原文相比：用单调高精度的 time.perf_counter() 替换 time.time()；为合成数据和随机估计器补上固定种子；增加计时前预热；随机森林显式设 n_jobs=1；吞吐窗口由 0.1 秒改为 5 秒，并用实际经过时间作分母。图表保留原来的三类比较，简化纯样式代码。新增主入口保护，导入本文件时不会自动开始基准。

这些变化会改变实验条件与结果，不能把本修订的未来输出和源站图表当作直接可比的同一实验。若目标是严格复现原页面，请使用原页面提供的下载代码。本段尚未执行。

# Based on scikit-learn developers' Prediction Latency example.
# SPDX-License-Identifier: BSD-3-Clause
# Editorial adaptation: monotonic timing, fixed seeds, warm-up,
# single-thread forest and actual elapsed-time throughput.
import gc
import time
import numpy as np
import matplotlib.pyplot as plt
from sklearn.base import clone
from sklearn.datasets import make_regression
from sklearn.ensemble import RandomForestRegressor
from sklearn.linear_model import Ridge, SGDRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVR
from sklearn.utils import shuffle

def generate_dataset(n_train=1000, n_test=100, n_features=100):
    X, y = make_regression(
        n_samples=n_train + n_test, n_features=n_features,
        noise=0.1, random_state=13
    )
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, train_size=n_train, test_size=n_test, random_state=13
    )
    X_train, y_train = shuffle(X_train, y_train, random_state=13)
    xs, ys = StandardScaler(), StandardScaler()
    X_train = xs.fit_transform(X_train)
    X_test = xs.transform(X_test)
    y_train = ys.fit_transform(y_train[:, None])[:, 0]
    y_test = ys.transform(y_test[:, None])[:, 0]
    return X_train, y_train, X_test, y_test

def atomic_times(estimator, X_test):
    estimator.predict(X_test[:1])  # warm-up, outside timing
    gc.collect()
    times = np.empty(len(X_test))
    for i in range(len(X_test)):
        row = X_test[[i], :]       # slicing is outside timing
        start = time.perf_counter()
        estimator.predict(row)
        times[i] = time.perf_counter() - start
    return times

def bulk_times(estimator, X_test, repeats=30):
    estimator.predict(X_test)      # warm-up, outside timing
    gc.collect()
    times = np.empty(repeats)
    for i in range(repeats):
        start = time.perf_counter()
        estimator.predict(X_test)
        times[i] = (time.perf_counter() - start) / len(X_test)
    return times

def throughput(estimator, X_test, duration=5.0):
    if duration <= 0:
        raise ValueError("duration must be positive")
    row = X_test[:1]
    estimator.predict(row)
    gc.collect()
    count = 0
    start = time.perf_counter()
    while time.perf_counter() - start < duration:
        estimator.predict(row)
        count += 1
    elapsed = time.perf_counter() - start
    return count / elapsed

def complexity(estimator):
    if isinstance(estimator, SGDRegressor):
        return f"{np.count_nonzero(estimator.coef_)} non-zero coefficients"
    if isinstance(estimator, RandomForestRegressor):
        return f"{estimator.n_estimators} trees"
    return f"{len(estimator.support_vectors_)} support vectors"

def run_benchmark():
    prototypes = {
        "Linear Model": SGDRegressor(
            penalty="elasticnet", alpha=0.01, l1_ratio=0.25,
            tol=1e-4, random_state=13
        ),
        "RandomForest": RandomForestRegressor(
            random_state=13, n_jobs=1
        ),
        "SVR": SVR(kernel="rbf"),
    }
    X_train, y_train, X_test, _ = generate_dataset()
    atomic, bulk, rates, labels = {}, {}, {}, []

    for name, prototype in prototypes.items():
        model = clone(prototype).fit(X_train, y_train)
        labels.append(f"{name}\n{complexity(model)}")
        atomic[name] = atomic_times(model, X_test)
        bulk[name] = bulk_times(model, X_test)
        rates[name] = throughput(model, X_test)
        print(
            name,
            "atomic p50/p90 us:", np.percentile(atomic[name], [50, 90]) * 1e6,
            "bulk amortized p50/p90 us:",
            np.percentile(bulk[name], [50, 90]) * 1e6,
            "one-row predictions/s:", rates[name]
        )

    for mode, values in (("Atomic", atomic), ("Bulk / 100", bulk)):
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.boxplot([values[name] * 1e6 for name in prototypes])
        ax.set_xticks(range(1, len(labels) + 1), labels)
        ax.set_ylabel("Prediction time per sample (microseconds)")
        ax.set_title(f"{mode}: 100 features")
        ax.grid(axis="y", alpha=0.3)

    feature_counts = [100, 250, 500]
    p90 = []
    for n in feature_counts:
        Xtr, ytr, Xte, _ = generate_dataset(n_features=n)
        model = Ridge().fit(Xtr, ytr)
        p90.append(np.percentile(bulk_times(model, Xte), 90) * 1e6)
    fig, ax = plt.subplots()
    ax.plot(feature_counts, p90, marker="o")
    ax.set_xlabel("Number of features")
    ax.set_ylabel("Bulk amortized p90 (microseconds/sample)")
    ax.set_title("Ridge: feature-count influence")

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.bar(list(rates), list(rates.values()))
    ax.set_ylabel("One-row predictions / second")
    ax.set_title("Warm, in-process throughput; 100 features")
    plt.show()

if __name__ == "__main__":
    run_benchmark()

特征维数的影响，要固定所比较的统计量

原文第二组实验仅使用 Ridge()，依次生成 100、250、500 个特征的数据，每次重新训练，然后进行 30 次批量计时，提取 p90 并画成折线。它回答的是该配置下，特征维数改变时，批量单位样本成本怎样变化。

这条曲线同时受到模型、输入布局、底层实现和计算环境影响，不能把“更多特征必然按固定倍数更慢”当作结论。每个维数重新生成数据，也意味着它不是在同一已训练模型上简单追加无用列。若要判断某个真实产品的维数成本，应设计符合该产品输入和训练流程的实验。

吞吐量：数的是同一进程中的重复调用

原例在默认 0.1 秒窗口内持续对 X_test[[0]] 做单样本预测，再把调用次数除以目标窗口长度。此写法短小，但当最后一次预测越过截止时间时，实际运行时间可能大于目标窗口，用目标窗口作分母会高估吞吐；短窗口也更容易受计时精度和偶发调度影响。

修订版延长窗口，并以实际 elapsed 作分母，仍然反复预测第一条测试样本。它代表预热后的单进程、相同输入调用速率，可能有明显缓存效应。它没有并发请求、队列、网络传输、序列化、业务预处理或数据库访问，不能直接换算成线上 QPS 或 SLA。真实服务应补测不同输入、并发程度、冷启动、批处理策略及端到端延迟。

静态审核与复现限制

原文只给划分和 shuffle 设置了随机种子，make_regression 与随机估计器仍可能变化；本稿补种子是为了减少这一不确定性，不是为了“保证测得相同耗时”。时间测量仍受 CPU 频率、其他进程、线程数、垃圾回收和缓存状态影响。gc.collect() 仅是在阶段之间主动收集，不等于关闭垃圾回收或消除系统噪声。

所示代码没有网络下载、硬编码凭据、shell 调用或文件删除。主要操作风险是大规模数据或过长窗口带来的 CPU、内存占用；应从有界小规模参数开始。静态审查没有发现直接注入入口，但没有执行测试就不能保证运行行为。因此本文没有宣称任何一组基准已经跑过，也没有把示意图当作实测结果。

原作与示例：scikit-learn developers，© 2007–2026 scikit-learn developers；代码 SPDX：BSD-3-Clause。翻译整理、计时修订与技术示意图：未完纪。完整代码许可告知见下方全文。来源：Prediction Latency 官方完整示例。

BSD 3-Clause License

Copyright (c) 2007-2026, The scikit-learn developers.
All rights reserved.

Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

原文性能结果图

以下图为源站历史实验，使用原始计时与配置；不能与本文更改随机种子、预热、线程和5秒窗口后的未来结果直接比较。前两图的SVR标注为660个支持向量，吞吐图为697个，原脚本分别生成数据和拟合模型，因此不是同一个已训练模型的多项指标。前两图均以微秒/样本计量；批量图先除以批大小100，第三图是批量摊销时间的第90百分位。