批量向量化与限流
把成千上万条切片转成向量，关键是把吞吐与稳定性同时管住：本地模型按显存与内存调 batch_size，托管 API 按配额控并发并对限流错误做指数退避。更要紧的是可恢复性——逐批记录进度，失败时只重试当前批次，重跑时跳过已完成部分。批量任务的失败模式往往是“跑了三小时在最后一批挂掉然后从头再来”，进度落盘是把这类浪费挡在门外的唯一办法。
基本信息
| 本地模型 | 按显存调 batch_size，长文本适当调小 |
|---|---|
| 托管 API | 按配额控并发，限流用指数退避重试 |
| 幂等要求 | 以 chunk_id 为准覆盖写入，重跑不产生重复 |
| 进度记录 | 已处理文档或批次清单落盘，支持断点续跑 |
| 校验 | 入库后比对向量条数与切片条数 |
典型使用场景
- 首次全量入库：几十万条切片需要数小时，必须能断点续跑
- 定期增量：只处理变更文档，避免重复计算已完成部分
- 换模型重建：新旧向量并存期间分批切换，便于回退
- 限流应对：托管服务返回限流错误时自动退避，而不是整批失败
- 成本核算：按批记录调用量与耗时，推算单价与扩容需求
分批编码与进度记录
分批编码与进度记录
def embed_all(texts, batch=128, done=None):
done = done or set() # 已完成 id 集合
out = []
for i in range(0, len(texts), batch):
part = texts[i:i + batch]
if all(t.id in done for t in part):
continue # 断点续跑：跳过已完成
out.extend(embed_batch(part))
save_progress(part) # 每批落盘进度
return out限流退避示意
限流退避示意
import time
def embed_with_retry(part, retries=5):
for attempt in range(retries):
try:
return embed_batch(part)
except RateLimitError:
time.sleep(min(2 ** attempt, 60)) # 指数退避，上限 60 秒
raise RuntimeError('超过最大重试次数')失败处理对照
| 失败类型 | 表现 | 处理 |
|---|---|---|
| 限流 | 返回速率错误 | 指数退避后重试当前批 |
| 单条超长 | 该条报错或截断 | 入库前按长度预校验并标记 |
| 显存不足 | 进程被 kill | 调小 batch_size 并支持续跑 |
| 网络中断 | 连接超时 | 重试 + 进度落盘，避免全量重来 |
入库前先按长度分布抽样，把超长切片单独处理：一条异常长的文本能让整批调用失败，也可能被服务端静默截断，两种结果都难排查。