把一整篇长文档直接喂给 EmbeddingGemma 2，通常会在显存上先撞墙：batch 越大峰值越高，而块切得太碎又会把跨段语义拆散，检索时召回变差。可行的处理顺序是先把单条输入长度压到可控范围，再用批大小阶梯测试找出当前设备的可用边界，最后用同一组查询比较两种块长的 top-k 结果，决定粒度取舍。

建议按「定块长与重叠 → 批大小阶梯测试 → 两种块长召回对照」的顺序推进。批大小是设备相关的量，换显卡、换精度、换块长都要重新测；块长是否牺牲召回，只有拿同一组查询的 top-k 结果对照才能判断，不能凭感觉定。

确定单条输入的最大长度与切块规则

先让每条输入长度可控，再谈批大小。如果单条输入本身就可能上千 token，那么无论批大小设成 1 还是 8，显存都不可预测。切块函数要把「块长」和「重叠窗口」都做成参数，重叠是为了避免答案正好落在切口上。

def chunk_text(text, chunk_size=512, overlap=64, unit='char'):
    # chunk_size / overlap 的单位由 unit 决定：char 或 token
    if overlap >= chunk_size:
        raise ValueError('overlap 必须小于 chunk_size')
    step = chunk_size - overlap
    units = list(text) if unit == 'char' else tokenizer.encode(text)
    chunks = []
    for start in range(0, len(units), step):
        piece = units[start:start + chunk_size]
        if not piece:
            break
        chunks.append(''.join(piece) if unit == 'char' else tokenizer.decode(piece))
    return chunks

按字符切实现简单、不依赖分词器，按 token 切更贴近模型真实输入长度，代价是多一次编解码。两种方式都要在落地前验证：逐条打印每个块的字符数和 tokenizer 编码长度，确认最大值不超过设定上限。尾部不足一块的残片是否保留，要看下游是问答检索还是全文召回，做问答时残片往往噪声更大，可以直接丢弃或与上一块合并。

用批大小阶梯测试记录显存峰值与耗时

阶梯测试的目的是找到当前设备上的可用批大小边界，而不是追求某个最优值。测试时块长、重叠、精度、设备都固定，只改批大小，按 1 / 8 / 16 / 32 依次跑同一批语料，出现 OOM 就把该档记为不可用，并回到上一档确认稳定性。

显存读取可以用两条途径：进程内用 torch 的峰值统计，设备侧用 nvidia-smi 采样。前者更贴近实际分配量，后者能看到整卡占用。

import time, torch

def run_batch(model, texts):
    torch.cuda.reset_peak_memory_stats()
    torch.cuda.synchronize()
    t0 = time.perf_counter()
    with torch.inference_mode():
        emb = model.encode(texts, batch_size=len(texts), convert_to_numpy=True)
    torch.cuda.synchronize()
    dt_ms = (time.perf_counter() - t0) * 1000
    peak_mib = torch.cuda.max_memory_allocated() / 1024 ** 2
    return emb, dt_ms, peak_mib

设备侧采样命令（另开一个终端运行，记录测试窗口内的最高值）：

nvidia-smi `--query-gpu`=memory.used,memory.total `--format`=csv -lms 200

记录表建议按下面字段固定下来，每次测试只填数值，避免换环境后无法对比：

注意峰值统计要在每档开始前 reset，否则上一档的残留会污染记录；耗时测量前先 synchronize，否则拿到的是异步提交时间。选中档位时建议留一档余量，因为实际语料里总会有比测试样本更长的块。

比较不同切块粒度下的召回结果

块长变小，定位更准但答案容易被切口截断；块长变大，单块上下文更多，但向量是把更多内容平均后的结果，细节查询的相似度分数可能下降。判断方法不是看分数高低，而是看检索回来的块拼回原文后，是否覆盖了回答问题所需的最小上下文。

做法：固定同一个 EmbeddingGemma 2 权重、同一套归一化方式、同一个 top-k（例如 k=5），只改块长，对同一组查询各跑一次，记录返回的块 ID、分数和块内文本片段，然后人工核对。

判断标准可以简化成两条：答案是否完整落在某一条返回块内；返回块里是否混入了大量无关内容。两条都过，块长就可以接受；如果只有小块能命中而大块总是漏，说明重叠窗口不够或块长超过语义跨度。查询集要覆盖长答案、跨段答案、以及只出现在文档某一段的细节问题，只测一两个问题容易得出偏乐观的结论。

把批处理写成流式生成器避免一次性载入

把全部语料先切块、再全部读进列表，峰值内存会被语料规模放大。改成生成器后，内存占用只和当前批次有关，语料总量超出内存也能跑完。

def stream_chunks(paths, chunk_size=512, overlap=64):
    for path in paths:
        with open(path, 'r', encoding='utf-8') as f:
            text = f.read()
        for chunk in chunk_text(text, chunk_size=chunk_size, overlap=overlap):
            yield {'path': path, 'text': chunk}

def iter_batches(records, batch_size):
    batch = []
    for r in records:
        batch.append(r)
        if len(batch) == batch_size:
            yield batch
            batch = []
    if batch:
        yield batch

# 调用示例
out = []
for batch in iter_batches(stream_chunks(paths), batch_size=16):
    texts = [item['text'] for item in batch]
    emb, dt_ms, peak_mib = run_batch(model, texts)
    out.append(emb)
    print(f'batch={len(texts)} peak={peak_mib:.1f}MiB time={dt_ms:.1f}ms')

这里要清楚一点：流式只降低峰值内存，不会让单批的显存需求变小，批大小仍然由阶梯测试的结果决定。批次写出方式也建议流式，边算边追加到 .npy 或写入索引，不要在最后统一落盘。

输出一份可复制的编码配置表

把影响结果的所有参数集中到一个配置文件里，便于版本化和回滚。配置变更后旧向量与新向量不能混用同一份索引，至少要在索引元数据里写入配置版本或哈希，需要回滚时恢复上一份配置重新编码。

# embed_config.yaml
model:
  name: embeddinggemma-2
  revision: （固定到某个 commit 或版本标识）
  device: cuda:0
  dtype: float16        # 可选 float32 / bfloat16
  normalize: true
chunking:
  unit: token           # token 或 char
  chunk_length: 512
  overlap: 64
  drop_tail: false
encoding:
  batch_size: 16        # 由阶梯测试确定，换卡需重测
  max_seq_len: 512
  show_progress: false
retrieval:
  top_k: 5
  metric: cosine
output:
  format: npy
  index_version: v1

验证方式很直接：加载配置后打印生效值，确认 chunk_length 与 max_seq_len 的关系、overlap 小于 chunk_length、batch_size 与当前设备实测边界一致。回滚时改回上一版配置并整体重编码，不要只对新文档追加，否则同一索引里会混入两种块长的向量。