CPU 跑 AI 推理有多快?用固定模型测延迟和吞吐量

固定逻辑回归模型、输入与线程设置,预热后用 perf_counter_ns 统计 CPU 推理 P50/P95 和样本吞吐;保存模型参数指纹和环境,说明与在线服务及 LLM 测速的区别。

CPU 跑 AI 推理有多快,要固定模型、输入尺寸、数据类型、线程设置和计时范围再测。本文用一个 16 特征的 scikit-learn 逻辑回归分类模型,在同一进程里测单次调用延迟和批量吞吐。结果只适用于这个小模型,不代表大语言模型的生成速度,也不能换算为 GPU 峰值算力。

先明确测的是什么时间

本例只计时 model.predict_proba(inputs):包含标准化、分类器计算和库的输入检查,不包含样本生成、模型训练、文件读取、网络请求和服务排队。批量大小为 1、64、512,输入固定为 float64,每次返回 [batch, 2] 概率矩阵。

CPU 跑 AI 推理有多快?用固定模型测延迟和吞吐量

P50 是 1000 次调用耗时的中位数;P95 是按NumPy percentile计算的 95 分位数,用来观察较慢调用。吞吐量使用“测量轮次 × 每批样本数 ÷ 计时总秒数”,单位为样本/秒;它不是每秒 HTTP 请求数,batch=512 的一次调用处理的是 512 个样本。

准备可重复的模型与环境

本机测试环境为 Windows 11、Python 3.13.14、scikit-learn 1.9.1、NumPy 2.5.3、threadpoolctl 3.7.0,CPU 标识为 Intel64 Family 6 Model 198 Stepping 2。脚本从固定随机种子的合成数据训练模型,训练发生在计时之前;这些数据不是生产流量。

python -m pip install scikit-learn numpy threadpoolctl
python -c "import numpy, sklearn; print(numpy.__version__, sklearn.__version__)"

LogisticRegression 官方文档描述了该分类器及 predict_proba。这里选择它是为了让读者用很小的本地样例检查计时方法;判断实际服务是否满足延迟要求,应替换为你自己的固定模型和有代表性的输入。

预热、计时、验证输出并保存条件

保存完整代码为 cpu_benchmark.py,执行 python cpu_benchmark.py。每批先预热 30 次,再记录 1000 次调用。Python 的 perf_counter_ns 文档说明该计时器适合测量短时段,使用两次读数的差值,而不是把其绝对值当成时间戳。

from pathlib import Path
from time import perf_counter_ns
import hashlib, json, platform, sys
import numpy as np
import sklearn
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from threadpoolctl import threadpool_info, threadpool_limits

X, y = make_classification(n_samples=4000, n_features=16, n_informative=8,
                           n_redundant=2, random_state=42)
X = np.ascontiguousarray(X, dtype=np.float64)
model = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000, random_state=42))
rounds, warmup = 1000, 30
results = []
with threadpool_limits(limits=1, user_api='blas'):
    model.fit(X, y)  # 不计入推理时间
    linear = model.named_steps['logisticregression']
    scaler = model.named_steps['standardscaler']
    state = b''.join(a.tobytes() for a in [scaler.mean_, scaler.scale_, linear.coef_,
                                         linear.intercept_, linear.classes_])
    fingerprint = hashlib.sha256(state).hexdigest()
    for batch in [1, 64, 512]:
        inputs = X[:batch].copy()  # 不计入时间
        expected = model.predict_proba(inputs)
        assert expected.shape == (batch, 2)
        assert np.allclose(expected.sum(axis=1), 1)
        for _ in range(warmup):
            model.predict_proba(inputs)
        times = np.empty(rounds, dtype=np.int64)
        for i in range(rounds):
            start = perf_counter_ns()
            output = model.predict_proba(inputs)
            times[i] = perf_counter_ns() - start
        assert np.allclose(output, expected)
        results.append({'batch': batch, 'p50_ms': float(np.percentile(times, 50) / 1e6),
                        'p95_ms': float(np.percentile(times, 95) / 1e6),
                        'samples_per_second': float(batch * rounds * 1e9 / times.sum()),
                        'measured_seconds': float(times.sum() / 1e9)})
    pools = [{k: p.get(k) for k in ['internal_api', 'version', 'num_threads', 'architecture']}
             for p in threadpool_info()]
report = {'python': sys.version.split()[0], 'numpy': np.__version__, 'sklearn': sklearn.__version__,
          'os': platform.platform(), 'cpu': platform.processor(), 'dtype': str(X.dtype),
          'features': 16, 'rounds': rounds, 'warmup': warmup, 'model_sha256': fingerprint,
          'supported_blas_pools': pools, 'results': results}
Path('cpu_benchmark.json').write_text(json.dumps(report, indent=2), encoding='utf-8')
print(json.dumps(report, indent=2))

脚本把 scaler 和分类器的参数拼成字节后计算 model_sha256,用来核对本例计时用的模型参数是否相同;这不替代整个运行环境的版本记录。模型生成和输入复制在计时外,循环内不打印日志,最后检查输出与计时前的一致,防止只测到错误或不同计算。

threadpoolctl 项目说明支持在代码块中限制已支持的 BLAS 线程。这里 user_api="blas" 只限制相应 BLAS 线程池,不保证整台机器、所有 OpenMP 库或所有运行时都变成单线程。输出会列出线程池实际设置,比较两次运行前应查看这些记录。

一次真实运行的结果怎样读

下面是 2026 年 10 月 1 日在上述机器实际执行后的记录,耗时按一次整批调用计算。测试轮次的计时总量每组约 0.24–0.26 秒,样本很短且运行时有其他进程,适合作为流程校验,不能作为该 CPU 型号的稳定性能排名。

每批样本数 P50,毫秒/批 P95,毫秒/批 样本/秒
1 0.2069 0.4179 4210
64 0.2083 0.3950 270453
512 0.2336 0.4255 1986982

本次批量吞吐较高,说明这个固定输入的小模型可以把调用开销分摊到更多样本;不能据此保证真实在线请求也有相同提升。服务等到 512 条再成批处理还会引入等待时间,业务端到端延迟应另测。

先核对 cpu_benchmark.json 的模型指纹、版本、线程记录、数据类型和每组 measured_seconds,再比较指标。正式比较时延长到多秒级窗口,重复运行几次,交替不同 batch 的测试顺序,记录系统负载和功耗模式;差异与波动幅度相近时,不宣称优化有效。

常见的错误测法与替换边界

  • 把训练、读文件和第一次调用混在一起:这会得到另一种计时范围,应另记“冷启动”或“端到端”指标。
  • 只记录最快一次:不能代表普通用户体验。至少保留中位数、尾部耗时和完整条件。
  • 两个模型精度或计算不同:速度变化可能是少算了内容,应先核对输出和任务质量。
  • 从这里推导 LLM 的 token/s:本模型输出二分类概率,不生成 token;语言模型需另分首 token 延迟、后续生成速度、上下文长度和输出长度。

如果你需要检查训练任务是否真正用到 NVIDIA GPU,可参考站内nvidia-smi 利用率与进程核对教程;设备利用率与本文的推理延迟不是同一个指标。替换推理框架时还要检查它是否异步执行,必须在结果计算完成的边界计时,不能照搬一个计时调用就宣布性能结论。

Ai菜鸟网。发布者:AI小管家,转载请注明出处:https://www.alyyhw.com/30315.html

赞 (0)
AI小管家的头像AI小管家
PyTorch 微调怎么冻结和解冻参数?检查梯度与优化器是否生效
上一篇 1天前
PyTorch 梯度累积怎么写?处理小批次与最后不足一组的数据
下一篇 1天前

相关推荐

联系我们

联系我们

1

在线咨询: QQ交谈

邮件:admin@example.com

工作时间:周一至周五,9:30-18:30,节假日休息

关注微信
关注微信
分享本页
返回顶部