LFM2.5 编码器:在 CPU 上压低长上下文推理成本

2026-07-28 19 预计阅读时间: 1 分钟
来源: huggingface.co AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:7 分钟

LFM2.5-Encoders 把目标放在一个很现实的部署场景上:使用 CPU 处理长上下文编码任务。对检索、语义匹配、重排前的特征提取和文档聚类系统来说,这意味着团队可以重新评估是否每条推理链路都必须占用 GPU,同时也需要更严谨地测量延迟、内存和截断行为。

长上下文改变了 CPU 推理的瓶颈

短句编码通常容易批处理,性能也相对稳定;输入变成长文档后,问题会明显复杂。分词时间、模型计算量、注意力相关开销和中间张量内存都会增长。此时只报告“每秒处理多少条文本”并不够,因为一条文本可能只有 50 个 token,也可能接近模型的上下文上限。

评估 LFM2.5-Encoders 时,至少应同时记录这些指标:

  • p50p95 单批延迟,而不是只看平均值;
  • 实际输入 token 数和是否发生截断;
  • 每秒处理的 token 数;
  • 进程峰值常驻内存;
  • 不同线程数、批大小和输入长度下的吞吐变化;
  • 输出向量在目标检索或分类任务上的质量。

“CPU 上更快”必须绑定具体条件,例如处理器型号、物理核心数、推理精度、运行时和上下文长度。缺少这些条件的单个速度数字很难用于容量规划。

不要让截断制造虚假的性能优势

长上下文基准最常见的问题,是 tokenizer 在不显眼的位置把输入截短。程序看起来处理了完整文档,模型实际上可能只看到了固定长度的前缀。另一个问题是用字符数代替 token 数;不同语言、标点密度和代码片段会产生完全不同的分词结果。

因此,基准程序应该显式设置 max_length,打印每个样本的 token 数,并把截断后的长度写入结果。生产系统还应决定超长输入的处理策略:

  • 直接截断,适合信息集中在文档开头的场景;
  • 滑动窗口分块,再聚合多个向量;
  • 按标题、段落或代码结构进行语义分块;
  • 先使用廉价规则筛选片段,再交给编码器。

这些策略会改变速度和检索质量,不能只在模型层面做比较。

可以这样实践:建立可复现的 CPU 基准

由于来源摘要没有给出具体仓库名、模型标识和调用 API,下面假设 LFM2.5-Encoders 提供与 Hugging Face Transformers 兼容的本地目录或模型 ID。运行前把 MODEL_ID 替换成实际标识。脚本会执行均值池化,并输出预热后的延迟、token 吞吐和向量维度。

先安装依赖:

python -m venv .venv
. .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install torch transformers

保存并运行下面的 bench_encoder.py

import os
import statistics
import time

import torch
from transformers import AutoModel, AutoTokenizer

MODEL_ID = os.environ.get("MODEL_ID", "replace-with-lfm2.5-encoder-id")
MAX_LENGTH = int(os.environ.get("MAX_LENGTH", "4096"))
THREADS = int(os.environ.get("OMP_NUM_THREADS", "4"))
RUNS = int(os.environ.get("RUNS", "10"))

# Replace these samples with representative production documents.
texts = [
    "CPU inference benchmark. " * 800,
    "Long-context encoders should be measured by tokens, latency, and memory. " * 600,
]

torch.set_num_threads(THREADS)
torch.set_grad_enabled(False)

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")

batch = tokenizer(
    texts,
    padding=True,
    truncation=True,
    max_length=MAX_LENGTH,
    return_tensors="pt",
)
actual_tokens = int(batch["attention_mask"].sum().item())
lengths = batch["attention_mask"].sum(dim=1).tolist()


def encode():
    outputs = model(**batch).last_hidden_state
    mask = batch["attention_mask"].unsqueeze(-1)
    summed = (outputs * mask).sum(dim=1)
    vectors = summed / mask.sum(dim=1).clamp(min=1)
    return torch.nn.functional.normalize(vectors, p=2, dim=1)

# Warm up kernels and memory allocation before timing.
for _ in range(2):
    encode()

latencies = []
for _ in range(RUNS):
    started = time.perf_counter()
    vectors = encode()
    latencies.append(time.perf_counter() - started)

latencies_ms = sorted(value * 1000 for value in latencies)
p95_index = min(len(latencies_ms) - 1, int(len(latencies_ms) * 0.95))
mean_seconds = statistics.mean(latencies)

print(f"model={MODEL_ID}")
print(f"threads={THREADS} max_length={MAX_LENGTH}")
print(f"tokens_per_sample={lengths} total_tokens={actual_tokens}")
print(f"mean_ms={mean_seconds * 1000:.2f}")
print(f"p95_ms={latencies_ms[p95_index]:.2f}")
print(f"tokens_per_second={actual_tokens / mean_seconds:.2f}")
print(f"embedding_shape={tuple(vectors.shape)}")

使用固定线程数运行,避免不同机器上的默认线程策略干扰结果:

MODEL_ID=/path/to/lfm2.5-encoder \
OMP_NUM_THREADS=8 \
MAX_LENGTH=4096 \
RUNS=20 \
python bench_encoder.py

这段代码中的均值池化只是兼容性示例。实际部署必须采用模型说明指定的池化方式、前缀格式和归一化规则,否则即使程序能运行,向量质量也可能不正确。

上线前的决策清单

CPU 编码器适合流量可预测、延迟预算允许、数据本地性重要或 GPU 成本敏感的服务,但并不自动适合所有长上下文任务。批量过大可能提高吞吐,却会抬高尾延迟和内存峰值;线程开得过多也可能因争用而变慢。

上线前应固定一组真实文档,分别测试目标机器上的线程数、批大小和上下文长度;同时运行检索召回率或下游任务评估。确认没有意外截断,记录模型版本、运行时版本和 CPU 型号,并用并发压测验证 p95p99。只有速度、内存与任务质量同时达到要求,LFM2.5-Encoders 的 CPU 长上下文能力才真正转化为可用的工程收益。


相关推荐