先对齐向量空间再谈召回——EmbeddingGemma 2 做图文跨模态检索

图文检索召回不准时，先别急着调 top-k、相似度阈值或加 rerank。先用 20~50 对图文把「编码 → 对齐 → 召回」三步跑通一遍：确认文本向量和图片向量落在同一个向量空间、维度和 dtype 一致且都做了归一化，再分别用文本查图、用图查文本，看两个方向的召回是否对称。效果差时最常见的误判，就是把对齐问题当成检索参数问题去反复调参。

判断路径：先建 20~50 对图文小数据集，确认文本与图片编码后的维度、dtype、归一化是否一致，再跑双向 top-k 对照。若文本查图明显差于图片查文本，优先查文本侧截断与语言覆盖；若图片查文本差，优先查截图类、极端长宽比和多主体图片；两个方向都差，则回到向量空间对齐本身，而不是先调检索参数。小数据集只能验证流程与对齐是否成立，不能替代大规模效果评估。

搭一个 20~50 对的图文小数据集并写清目录结构

这个小数据集的作用不是评估最终效果，而是让后续每一步都有可复现的输入。建议 20~50 对起步，覆盖两类语言、照片与截图两类图片、长短两种描述长度；如果全部样本都是同一种风格，后面很难判断偏差来自模型还是数据。

目录约定如下，data_dir 只放数据，脚本和缓存放在别处：

data/mm_mini/
├── images/
│   ├── 0001.jpg
│   └── 0002.png
├── captions.jsonl
└── README.md

图片命名用固定宽度数字，与 JSONL 里的 id 一一对应，避免按文件名排序时出现 10 排在 2 前面的情况。captions.jsonl 每行一条，字段保持稳定，示例条目如下：

{"id": "0001", "image": "images/0001.jpg", "text": "一只橘猫趴在窗台上", "lang": "zh", "type": "photo"}

字段含义：id 是唯一键，用于召回结果回查原图与原描述；image 是相对 data_dir 的路径，不要写绝对路径，方便整套数据搬目录；text 是这张图的主要描述；lang 标注语言，中英混排可以标成 zh_en；type 标注 photo、screenshot、chart 等图片类型。lang 和 type 在第四步归因失败样本时可以直接分组，所以从第一行数据开始就要填，不要事后补。

把文本与图片分别编码到同一向量集合

编码这一步只做三件事：文本批量编码、图片批量编码、检查两者是否落在同一向量集合。EmbeddingGemma 2 这类模型的输出维度、最大输入长度和是否自带归一化，都以你本地加载的权重与 tokenizer 配置为准，不要凭记忆写死。批量编码骨架如下，MODEL 和 load_image 替换成你本地可用的实现，其余逻辑不变：

import json, pathlib
import numpy as np

def l2(x, eps=1e-12):
    n = np.linalg.norm(x, axis=1, keepdims=True)
    return x / np.maximum(n, eps)

def encode_texts(texts, batch_size=32):
    vecs = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        # 占位：替换为本地模型提供的文本编码调用
        vecs.append(MODEL.encode_text(batch))
    return l2(np.concatenate(vecs, axis=0).astype(np.float32))

def encode_images(paths, batch_size=16):
    vecs = []
    for i in range(0, len(paths), batch_size):
        batch = [load_image(p) for p in paths[i:i + batch_size]]
        vecs.append(MODEL.encode_image(batch))
    return l2(np.concatenate(vecs, axis=0).astype(np.float32))

编码完先断言，再算相似度。断言失败说明两条分支没有对齐，这时候算出来的 top-k 没有参考价值：

def check_alignment(text_vec, image_vec):
    assert text_vec.ndim == 2 and image_vec.ndim == 2, "向量应为 [N, D] 二维"
    assert text_vec.shape[1] == image_vec.shape[1], (
        f"文本维度 {text_vec.shape[1]} 与图片维度 {image_vec.shape[1]} 不一致，"
        "两条编码分支不在同一空间")
    assert text_vec.dtype == image_vec.dtype, (
        f"dtype 不一致：{text_vec.dtype} vs {image_vec.dtype}")
    for name, v in (("text", text_vec), ("image", image_vec)):
        norms = np.linalg.norm(v, axis=1)
        assert np.allclose(norms, 1.0, atol=1e-3), (
            f"{name} 向量未归一化，范数示例 {norms[:3]}")
    print("alignment_ok", text_vec.shape)

如果模型本身已经输出归一化向量，再套一次 l2 通常不会破坏结果，但两侧必须用同一种处理方式。一边归一化一边不归一化，内积相似度会被向量模长带偏，表现就是某几张图的相似度总是异常高。

用文本查图库、用图片查文本库各跑一遍 top-k

两个方向要分开跑。文本查图库，是把文本向量和 image_emb 做内积取前 k；图片查文本库，是把图片向量和 text_emb 做内积取前 k。两个方向共用同一个 topk 函数即可：

def topk(query_vec, bank_vec, bank_ids, k=3):
    sims = bank_vec @ query_vec
    order = np.argsort(-sims)[:k]
    return [(bank_ids[i], round(float(sims[i]), 4)) for i in order]

q_text = encode_texts(["雨天的街道"])[0]
print(topk(q_text, image_emb, image_ids, k=3))

q_img = encode_images(["images/0007.jpg"])[0]
print(topk(q_img, text_emb, text_ids, k=3))

输出按下面格式记录，每条查询都要人工标一次相关程度，标注结果才是判断对齐好坏的依据。下面的条目只是记录格式示例，实际 id 和顺序以你本地跑出的输出为准：

路径 A：text -> image
q03 "雨天的街道" -> img_0012(相关) / img_0007(部分) / img_0031(无关)

路径 B：image -> text
img_0012 -> q03(相关) / q11(无关) / q07(部分)

对照时看三件事：同一对图文在两个方向是否都能进 top-3，也就是互为近邻；文本查图里排在第一位的是不是明显无关图；图片查文本时召回的描述是否只命中同一条文本、其余全是无关。只有文本查图差、图片查文本正常，问题多半在文本侧；反过来则先查图片侧。两个方向都差，先回到 check_alignment 的维度、dtype、归一化结果，而不是加大 k 或加阈值过滤。

挑出对齐失败的样本，记录它们的共性

把两个方向里人工标为无关、或者本该命中却没进 top-k 的条目汇总成失败样本列表。每条至少记四列：样本 id、查询方向、查询内容摘要、判为失败的理由。然后按下面几个维度归类，看偏差集中在哪一类：

长度：文本 token 数是否超过 tokenizer 的最大长度而被截断。可以打印 tokenizer 编码后的长度分布，超过上限的样本单独标出来。
语言：中文、英文、中英混排分别统计失败条数。模型语料覆盖弱的语言，容易整体偏移。
图片类型：截图、图表、带大量文字的图片、极端长宽比、低信息量纯色图、多主体图。这几类在跨模态对齐里通常更容易偏。
其他可查项：图片是否有旋转或 EXIF 方向问题、文本描述是否与图片主体不一致（数据本身写错）、同一条文本是否被多张图共用导致歧义。

归类之后得到一个可行动的判断：如果失败集中在长文本，就先统一截断策略并按描述主体写短文本；如果集中在截图类图片，就要判断当前模型是否适合这类画面，必要时把截图类样本从主检索库里拆出来单测；如果失败分散在各语言各类型之间，那更可能是向量空间本身没有对齐好，需要换编码权重或检查两条编码分支。

把编码与对齐步骤收成一个可复用函数

把上面的编码、断言、建库、查询收成一个函数，换数据集时只改 data_dir，不改流程：

def build_retriever(data_dir, model, *, batch_size=32,
                    max_length=None, k=3):
    """构建双向图文检索所需的向量与元数据。

    data_dir   : 含 images/ 与 captions.jsonl 的数据目录
    model      : 同时提供 encode_text / encode_image 的编码器
    batch_size : 编码批大小，内存或显存不足时调小
    max_length : 文本截断长度，None 表示沿用 tokenizer 默认值
    k          : 每条查询返回的候选数量
    返回 dict: text_ids / text_emb / image_ids / image_emb / meta
    """
    rows = load_manifest(data_dir)
    texts = apply_max_length([r["text"] for r in rows], max_length)
    paths = [str(pathlib.Path(data_dir) / r["image"]) for r in rows]
    image_emb = encode_images(paths, batch_size)
    text_emb = encode_texts(texts, batch_size)
    check_alignment(text_emb, image_emb)
    return {
        "text_ids": [r["id"] for r in rows],
        "text_emb": text_emb,
        "image_ids": [r["id"] for r in rows],
        "image_emb": image_emb,
        "meta": rows,
    }

一次完整调用如下，换数据集只改第一个参数：

R = build_retriever("data/mm_mini", model=MODEL, batch_size=32, k=3)

print(topk(encode_texts(["雨天的街道"])[0],
           R["image_emb"], R["image_ids"], k=3))

print(topk(encode_images(["images/0007.jpg"])[0],
           R["text_emb"], R["text_ids"], k=3))

参数里最容易被忽略的是 max_length 和 batch_size：前者决定长描述会不会被悄悄截断，后者只影响内存占用，不影响向量结果。换新数据集后，先跑一次 check_alignment，再看双向 top-k 是否对称，确认对齐成立之后，才谈召回策略和检索参数。