基于多模态特征融合的图像文本检索：ALS 共享空间·余弦检索·一致性清洗（优秀范文一）

摘要：本文面向「以文搜图」的跨模态检索问题，在合成图文对数据集（300 对、16 维双模态向量、8 个语义主题、12.7% 描述错配噪声）上完成从特征诊断到检索评估的全流程。诊断发现两模态编码坐标系异构且尺度失衡（文本平均范数为图像的 3.6 倍），直接余弦检索退化为随机（Recall@1 仅 0.003）。为此提出交替最小二乘（ALS）线性对齐：以配对约束 $\min\sum\|P_i x-P_t z\|^2$ 交替求解两个投影，把异构坐标系接入 $d=8$ 维共享空间，损失 20 轮内从 104.8 收敛到 1.6，检索性能跃升至 Recall@1 0.757、R@5 0.950、mAP 0.845。针对错配噪声，利用对齐残差的分布特性证明「高残差富集噪声对」（最大残差十分位中错配占比过半），据此剔除重训使 R@1 再升至 0.800、mAP 升至 0.860，同时观察到长尾召回（R@10）小幅回落的诚实代价。错误案例分析显示残余错误集中于语义相邻主题。全文纯标准库实现、固定种子可复现，正文、配图、附录、真源四路数字一致。

关键词：跨模态检索；共享嵌入空间；交替最小二乘；余弦相似度；Recall@K；数据清洗

一、问题重述

给定图像集合与对应文本描述，需构建跨模态检索系统：

分别给出图像与文本的向量表示，并诊断其可用性；
设计融合/对齐方法，把两种模态映射到可比对的共享语义空间；
实现「query 文本 → 候选排序 → Top-K 图像」的检索管线；
以 Recall@K 与 mAP 评估系统，定位主要错误模式并给出数据质量改进手段。

二、模型假设

同一对的图像与文本共享一个不可观测的实例因子（场景内容本身），两模态向量均由「主题原型 + 实例因子 + 模态噪声」线性叠加生成；
两模态使用彼此独立的编码器，坐标系异构、尺度不同；
少量描述与图像不符（错配对）随机产生，比例约一成；
检索正确性以「是否检回配对的原图」为唯一判据，主题匹配但不配对不算命中。

三、符号说明

符号	含义
$x_i\in\mathbb{R}^{16},\ z_i\in\mathbb{R}^{16}$	第 $i$ 对的图像/文本向量
$P_i,P_t$	图像/文本到共享空间的投影矩阵（ $8\times16$ ）
$\tau$	错配对剔除比例
$\mathrm{rank}_i$	第 $i$ 对正确图像在检索序列中的名次

四、双模态特征与可用性诊断（问题 1、2）

300 对样本覆盖海滩、城市、森林、雪山、沙漠、湖泊、花园、夜空 8 个主题。诊断发现两个必须处理的障碍：

尺度失衡：文本向量平均 L2 范数 47.74，是图像（13.38）的 3.6 倍——若用未归一化的点积或欧氏距离，数值大的模态将主导排序；
坐标系异构：两模态编码器独立训练，即使语义相同，向量的方向分布也毫无对应关系。

图1 多模态图文检索四问建模框架

图2 两模态向量范数诊断

图2 直观呈现了尺度差。这两个诊断决定了后文技术路线：先归一化消除尺度，再做线性对齐消除坐标系差异——顺序不能颠倒。值得强调的是，诊断本身不需要标签之外的额外信息：范数比只需一次全库扫描，坐标系异构则可通过「同对向量夹角分布接近随机」直接证实。把问题定性为「表示层失配」而非「数据不足」，避免了盲目堆模型深度的弯路，也为后续每个建模决策提供了事实基础。

五、ALS 线性对齐与检索管线（问题 3）

设共享空间维数 $d=8$ ，目标：

$\min_{P_i,P_t}\ \sum_{i=1}^{n}\big\|P_i x_i - P_t z_i\big\|^2$

固定一侧时另一侧是标准最小二乘问题：把目标按 $P_i$ 展开为 $\mathrm{tr}(P_i A P_i^{\mathsf T})-2\,\mathrm{tr}(P_i B^{\mathsf T})+\mathrm{const}$ ，其中 $A=\sum_i x_i x_i^{\mathsf T}$ 为图像 Gram 矩阵、 $B=\sum_i (P_t z_i)x_i^{\mathsf T}$ 为交叉协方差，令梯度为零得闭式解 $P_i=B A^{-1}$ （图像侧 Gram 施加 $10^{-4}$ 岭正则防奇异；文本侧因尺度更大、条件数更差，取更强的 $10^{-2}$ ）；对称地固定 $P_i$ 可解 $P_t$ ，如此交替迭代直至损失不再下降。相比依赖负采样与温度超参的对比学习，ALS 只消费正配对约束、无离散超参，20 轮即把损失从 104.8 收敛到 1.6（图3），在 300 对小样本上稳定可靠。

共享维数取 $d=8$ 并非任意：它与主题数相等，恰好容纳「主题原型 + 实例因子」的全部可控变异，同时把 16 维中其余 8 维当作模态私有噪声截断。若 $d$ 取满 16 维，对齐会把噪声坐标一并强行拟合，错配对的干扰随维数放大； $d$ 过小又会挤掉实例因子，使同主题样本在共享空间中过度坍缩。这种「瓶颈式降维」与主成分分析截断次要方差的思路一脉相承。

图3 ALS 对齐损失收敛曲线

图4 三方案 Recall@K 对比

表 1 的对比给出了决定性结论：不做对齐，一切归一化技巧都是徒劳——无对齐方案的 R@1 仅 0.003（接近 $1/300$ 的随机水平）；接入共享空间后 R@1 跃至 0.757、R@10 达 0.987、mAP 0.845。

评估口径需作明确界定。Recall@K 衡量「正确图像进入前 $K$ 名」的查询比例，反映检索系统的头部体验；mAP 在本任务的单相关项设定下退化为平均倒数名次 $\frac{1}{n}\sum_i 1/\mathrm{rank}_i$ ，对名次靠前的命中赋予更高权重，能区分「第 2 名命中」与「第 50 名勉强进圈」这两种在 R@10 口径下不可分的情形。两者联用方能同时刻画头部精度与整体排序质量。

方案	R@1	R@5	R@10	mAP
无对齐直接余弦	0.003	0.007	0.020	0.015
ALS 对齐 + 归一化	0.757	0.950	0.987	0.845
对齐 + 一致性清洗	0.800	0.927	0.963	0.860

六、错误案例分析与一致性清洗（问题 4）

6.1 主题混淆结构

Top-1 结果的主题混淆矩阵（图5）对角命中 239/300。残余错误的主体是语义相邻主题互换：夜空→沙漠（4 次）、海滩→雪山（3 次）——夜空与沙漠共享「大面积暗色背景」的低层视觉统计，海滩与雪山共享「亮色地平线」结构。分主题看（图6），夜空的 Top-1 命中率最低（70.3%），花园、湖泊、沙漠并列最高（78.4%）。这说明错误并非随机噪声，而是「主题间视觉语义重叠」的系统性混淆，可通过引入更细粒度的判别性特征缓解。

图5 主题混淆矩阵

图6 各主题 Top-1 命中率

混淆的空间根源可在共享嵌入中得到直观印证。图7 给出对齐后的共享空间散点：两模态投影交叠成 8 个主题簇，同主题的图文向量彼此毗邻，说明对齐确实建立了跨模态对应关系；与此同时，夜空、沙漠等语义相邻簇之间的边界带明显偏窄，少量文本投影越界落入邻簇领地——这正是混淆矩阵中系统性互换现象的几何形态，也与「细粒度判别特征」的改进方向相互呼应。

图7 对齐后的共享嵌入空间

6.2 残差探针与清洗增益

错配对的文本描述的是别的场景，其对齐残差天然偏大。把全部样本按残差降序切十分位（图8）：最大残差分位 D1 中错配对占比过半，而低残差分位几乎为零——残差是无需标注的噪声探针。剔除残差最高的 10% 样本重训后，R@1 从 0.757 升至 0.800（+4.3pp）、mAP 升至 0.860。

图8 残差十分位的错配富集

值得诚实指出的代价：清洗剔除了部分「难但正确」的高残差样本，长尾召回 R@10 由 0.987 回落至 0.963。因此工程上推荐两段式策略：线上主库用清洗后模型保证头部精度；对被剔除样本建立人工复核队列，确认无误后回流再训练。

剔除比例 $\tau$ 的选取应与噪声率联动：本库实测错配占 12.7%，取 $\tau=10\%$ 意味着以轻微漏杀换取极高查准——D1 分位内错配浓度过半，剔除它相当于定向拆除污染最重的地层。若冒进地取 $\tau=30\%$ ，大量「难但正确」的高方差样本（罕见光照、非常规构图）会被连带误删，模型退化为只认识典型样本； $\tau$ 过小则污染清除不彻底、增益有限。工程上宜把 $\tau$ 当作受监控量而非固定常量：按估计噪声率设定初值，再依据每轮回流数据的残差富集曲线动态校正。

七、模型检验与优缺点

优点：①ALS 闭式交替求解零超参、小样本稳定，避免了对比学习的调参负担——InfoNCE 类方法依赖大规模负采样，在 300 对规模下负例池稀薄、表征易向平凡解塌缩，ALS 仅以配对平方误差为目标反而更为稳健；②归一化与对齐的先后次序由诊断数据驱动而非经验拍板；③残差探针把「数据质量」纳入建模闭环并量化了增益与代价。

缺点：①线性投影无法捕捉非线性跨模态映射，真实图文对宜用核化 ALS 或深度对齐；②实例因子假设简化了真实场景的多对象组合语义；③评估仅含单一相关项，未处理多相关文档场景；④清洗依赖「错配残差必然偏大」这一前提，若噪声来自高质量但语义漂移的描述，残差探针的富集能力会衰减，届时需引入置信学习等方法交叉验证。

八、结论

本文完成了跨模态检索从「特征诊断 → 空间对齐 → 检索评估 → 数据治理」的完整闭环。核心结论：异构坐标系的线性对齐是把检索从随机水平（R@1=0.003）带入实用区间（R@1=0.757）的决定性一步；基于对齐残差的一致性清洗以少量长尾召回为代价换取头部精度提升（R@1=0.800、mAP=0.860），适合与人工复核配合使用。方法论上，「先诊断尺度与坐标系、再选对齐方法」以及「把数据质量做成可度量的闭环」，对所有双塔式多模态系统均有借鉴价值。

附录：核心 Python 实现

# -*- coding: utf-8 -*-
"""跨模态检索核心：ALS 线性对齐 + 余弦检索 + Recall@K/mAP。
在 assets/problems/papers 目录下独立运行，输出与正文一致的权威数字。"""
import math, random

SEED, NTOPIC, DIM, D = 20260825, 8, 16, 8
NPAIR, NOISE = 300, 0.12
TOPICS = ["海滩", "城市", "森林", "雪山", "沙漠", "湖泊", "花园", "夜空"]

def protos(seed):
    rng = random.Random(seed)
    base = [[rng.gauss(0, 1) for _ in range(DIM)] for _ in range(NTOPIC)]
    out = []
    for k in range(NTOPIC):
        v = list(base[k])
        if k > 0:
            v = [0.55*base[k-1][j] + 0.83*v[j] for j in range(DIM)]
        n = math.sqrt(sum(x*x for x in v)) + 1e-9
        out.append([x/n*6.0 for x in v])
    return out

def gen():
    rng = random.Random(SEED)
    ip, tp = protos(SEED+11), protos(SEED+22)
    pr = [[rng.gauss(0, 1)/math.sqrt(DIM) for _ in range(DIM)]
          for _ in range(DIM)]
    ps = []
    for i in range(NPAIR):
        topic = i % NTOPIC
        noisy = rng.random() < NOISE
        tt = rng.randrange(NTOPIC) if noisy else topic
        u = [rng.gauss(0, 3.0) for _ in range(DIM)]
        ru = [sum(pr[a][b]*u[b] for b in range(DIM)) for a in range(DIM)]
        xv = [ip[topic][j] + u[j] + rng.gauss(0, 0.7) for j in range(DIM)]
        zv = [tp[tt][j]*5.0 + 3.2*ru[j] + rng.gauss(0, 1.2) for j in range(DIM)]
        ps.append(dict(id=i, topic=topic, clean=not noisy, x=xv, z=zv))
    return ps

def inv(A):
    n = len(A)
    M = [list(A[i]) + [float(i == j) for j in range(n)] for i in range(n)]
    for c in range(n):
        p = max(range(c, n), key=lambda r: abs(M[r][c]))
        M[c], M[p] = M[p], M[c]
        d = M[c][c] + 1e-10
        M[c] = [v/d for v in M[c]]
        for r in range(n):
            if r != c and M[r][c]:
                f = M[r][c]
                M[r] = [M[r][j]-f*M[c][j] for j in range(2*n)]
    return [row[n:] for row in M]

def dot(M, v):
    return [sum(M[r][c]*v[c] for c in range(len(v)))
            for r in range(len(M))]

def gram(X):
    A = [[0.0]*DIM for _ in range(DIM)]
    for x in X:
        for i in range(DIM):
            for j in range(DIM):
                A[i][j] += x[i]*x[j]
    return A

def sq(a, b):
    return sum((u-v)**2 for u, v in zip(a, b))

def als(ps, iters=20):
    X = [p["x"] for p in ps]; Z = [p["z"] for p in ps]
    Ai = inv([[v+(1e-4 if i == j else 0) for j, v in enumerate(r)]
              for i, r in enumerate(gram(X))])
    Ati = inv([[v+(1e-2 if i == j else 0) for j, v in enumerate(r)]
               for i, r in enumerate(gram(Z))])
    Pt = [[float(i == j and j < D) for j in range(DIM)] for i in range(D)]
    curve = []
    for _ in range(iters):
        B = [[sum(dot(Pt, z)[r]*x[c] for z, x in zip(Z, X))
              for c in range(DIM)] for r in range(D)]
        Pi = [[sum(B[r][k]*Ai[k][c] for k in range(DIM))
               for c in range(DIM)] for r in range(D)]
        Bt = [[sum(dot(Pi, x)[r]*z[c] for x, z in zip(X, Z))
               for c in range(DIM)] for r in range(D)]
        Pt = [[sum(Bt[r][k]*Ati[k][c] for k in range(DIM))
               for c in range(DIM)] for r in range(D)]
        curve.append(sum(sq(dot(Pi, x), dot(Pt, z))
                         for x, z in zip(X, Z))/len(X))
    return Pi, Pt, curve

def nrm(v):
    n = math.sqrt(sum(u*u for u in v)) + 1e-12
    return [u/n for u in v]

def ranks(ps, fx, fz):
    lib = [(p["id"], nrm(fx(p))) for p in ps]
    out = {}
    for q in ps:
        qv = nrm(fz(q))
        s = sorted(lib, key=lambda t: -sum(a*b for a, b in zip(qv, t[1])))
        out[q["id"]] = next(k+1 for k, t in enumerate(s) if t[0] == q["id"])
    return out

def metrics(ps, rk):
    n = len(ps)
    r1 = sum(1 for p in ps if rk[p["id"]] <= 1)/n
    r5 = sum(1 for p in ps if rk[p["id"]] <= 5)/n
    r10 = sum(1 for p in ps if rk[p["id"]] <= 10)/n
    mp = sum(1.0/rk[p["id"]] for p in ps)/n
    return r1, r5, r10, mp

ps = gen()
mx = sum(math.sqrt(sum(v*v for v in p["x"])) for p in ps)/len(ps)
mz = sum(math.sqrt(sum(v*v for v in p["z"])) for p in ps)/len(ps)
print("文本/图像平均范数比 %.1f" % (mz/mx))
Pi, Pt, cv = als(ps)
print("ALS 损失 %.1f -> %.3f" % (cv[0], cv[-1]))
m_no = metrics(ps, ranks(ps, lambda p: p["x"], lambda p: p["z"]))
m_al = metrics(ps, ranks(ps, lambda p: dot(Pi, p["x"]),
                         lambda p: dot(Pt, p["z"])))
res = sorted(((sq(dot(Pi, p["x"]), dot(Pt, p["z"])), p["id"]) for p in ps),
             reverse=True)
keep = set(i for _, i in res[int(len(res)*0.10):])
Pi2, Pt2, _ = als([p for p in ps if p["id"] in keep])
m_cl = metrics(ps, ranks(ps, lambda p: dot(Pi2, p["x"]),
                         lambda p: dot(Pt2, p["z"])))
for name, m in (("无对齐", m_no), ("ALS对齐", m_al), ("清洗重训", m_cl)):
    print("%s R@1 %.3f R@5 %.3f R@10 %.3f mAP %.3f" % ((name,) + m))