qinghuan
← 回到知识探索

AI 直觉 / 可下载小项目

整段向量一样像,逐词匹配为什么能分出差别?

用二维向量算出 ColBERT 的 MaxSim,检查方向、重复项与补齐掩码,理解多向量检索到底保留了什么。

展开本文目录
  1. 下载完整练习包
  2. 先写出四个点,再算两种分数
  3. 先独立编码,最后才做逐词交互
  4. 交换两边与复制词,效果并不对称
  5. 补齐用的零,可能意外赢过真实词
  6. 打印每个查询项贡献,定位错误更直接
  7. 多保留一些信息,也增加了存储与评测工作
  8. 记住这些要点
  9. 别把例子用过头
  10. 自己试一小步
  11. 查原始资料
开始前,会这些就够了
遇到不熟悉的词,先看这 2 个解释
MaxSim
每个查询词先在文档里找最像的一项,再把各查询词得到的分数相加。
评分掩码
明确告诉打分程序哪些只是凑齐长度的占位位置,不能参与挑选最大值或求和。
打开完整术语小抄 ↗

查询里有两个关注点:一个沿横轴,一个沿纵轴。文档 A 各有一个词对准它们;文档 B 的两个词都落在对角线上。若先把整段词向量平均再归一化,两份文档方向完全一样,余弦分数都是 1。

如果保留每个词的向量,让查询的每一项各自找最像的文档词,A 可以得到两次满分,B 却只能得到两次约 0.707。这就是本文要拆开的 MaxSim 计算。它只证明两种打分规则保留的信息不同,不证明这些人工坐标等同于真实语义。

把例子带到自己的电脑上

下载完整练习包

需要 Python 3.10 或更新版本,无第三方依赖;制作时实测 Windows / Python 3.14.0。使用标准库,不需要安装第三方包。下载 ZIP 后选择“全部提取”,进入能看到 README.md 的文件夹。

下载 ZIP · 5 个文件
包里有什么
  • colbert_maxsim.py:完整可运行示例
  • README.md:运行说明与独立练习
  • observed.stdout.txt:本次实际输出
  • verification.json:源码和产物验证记录
  • SHA256SUMS.txt:内容摘要

在这个文件夹打开 PowerShell,逐行运行

python --version
python -B colbert_maxsim.py

解压到独立练习文件夹,在含 colbert_maxsim.py 的目录打开终端运行。人工二维向量的 MaxSim、非对称性、置换重复与补齐掩码;未调用编码器、建立检索索引、训练模型或评测真实语料。

找不到命令或文件时,先按 README 检查环境和所在目录。文中标出的阅读时间只估算文章篇幅,安装环境和动手练习可按自己的节奏进行。

先写出四个点,再算两种分数

查询向量为 q₁=[1,0]、q₂=[0,1]。文档 A 为 d₁=[1,0]、d₂=[0,1];文档 B 的两个向量都是 [1/√2,1/√2]。它们都是单位向量,点积因此等于余弦相似度。

三组向量分别平均后再归一化,都沿 [1/√2,1/√2],所以整段余弦无法区分 A 与 B。逐项计算时,查询对 A 的相似度矩阵为 [[1,0],[0,1]];对 B 则四个位置都是约 0.707107。

沿每一行取最大值,再把两行相加:A 为 1+1=2;B 为 0.707107+0.707107≈1.414214。分数可以超过 1,因为它是多个词的贡献之和,并不是概率,也不是整段的一个余弦值。

先独立编码,最后才做逐词交互

令 Q 是查询向量列表、D 是文档向量列表,打分式为 S(Q,D)=Σᵢ maxⱼ(qᵢ·dⱼ)。i 遍历查询位置,j 遍历有效文档位置。max 是每个查询项寻找自己的最佳匹配,sum 则把所有查询项的匹配强度加起来。

ColBERT 先分别编码查询与文档,保留词级向量,在打分阶段才让它们交互,所以称为后期交互。真实向量会受各自文本上下文影响;本文直接给定二维向量,完全省略编码器,便于逐项核对算术。

多个查询项可以选择同一个文档向量,公式没有“一词只能用一次”的约束。它不是一对一配对算法,也不会仅凭这条打分式证明文档同时满足多个独立事实。

交换两边与复制词,效果并不对称

取 Q=[e₁,e₁]、D=[e₁],其中 e₁=[1,0]。正向打分有两个查询位置,各得 1,总分为 2;交换两边,只有一个查询位置,总分为 1。因此 S(Q,D) 通常不等于 S(D,Q),不能把它当作对称距离。

固定向量时,重排文档位置不会改变每行最大值;增加一个完全相同的文档向量也不会改变最大值。程序用 17 个置换或重复案例检查这些性质。复制查询向量却会多加一份贡献:在主例 A 中再加一个 e₁,分数从 2 变成 3。

这些性质只针对已经编码好的向量列表。把原始文本中的词调序或重复后再重新编码,其他词的上下文向量也可能改变,所以不能据此声称真实 ColBERT 忽略词序或重复文本永远不影响结果。

补齐用的零,可能意外赢过真实词

查询 q=[1,0],唯一有效文档向量为 [−0.6,0.8],相似度是 −0.6。为了凑齐批次长度,后面补一个 [0,0]。如果不排除补齐位置,最大值会选中 0,分数就从 −0.6 被抬到 0。

因此文档掩码要在 max 之前生效:本文直接跳过无效位置;张量实现也可以先把无效相似度置为足够小的值。已核对的官方 colbert_score_reduce 在取最大值前使用 D_mask 排除补齐分数。把向量本身置零并不能独自完成这一步。

查询侧若有本来不应参与评分的占位位置,也需要在求和时排除。本文的 query_mask 只表示这种明确无效的位置。原版 ColBERT 还有使用 [MASK] 产生查询扩展向量的设计,那些参与建模的位置不能因为名字叫 mask 就一律删掉。

若文档全是无效位置,本例拒绝评分;若查询没有有效位置也拒绝。这样避免把空集合最大值或默认 0 随意伪装成正常相关度。生产系统可以采用另一个明确约定,但需要让排序与空输入处理共同遵守它。

打印每个查询项贡献,定位错误更直接

保存为 colbert_maxsim.py,运行 python -B colbert_maxsim.py。它打印整段余弦、两份文档的 MaxSim 和逐项贡献,核对非对称性、固定向量置换与重复、文档掩码和查询掩码,并写入 report.json。

程序也拒绝全无效掩码、维度不一致和掩码长度不对齐。maxsim 函数计算的是点积,主例有效向量已经单位化;如果传入非单位向量,结果不再自动具有余弦的意义。

展开收起代码 · 小实验python · 103 行

长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。

小实验 · python
"""Fixed-vector late interaction, with explicit document and query masks."""
import itertools
import json
import math
from pathlib import Path


def dot(a,b):
    if len(a) != len(b):
        raise ValueError('vector dimensions differ')
    return sum(x*y for x,y in zip(a,b))


def unit(vector):
    length = math.sqrt(dot(vector,vector))
    if not math.isfinite(length) or length <= 0:
        raise ValueError('nonzero finite vector required')
    return [x/length for x in vector]


def pooled_unit(vectors):
    return unit([sum(row[d] for row in vectors)/len(vectors) for d in range(len(vectors[0]))])


def maxsim(query, document, query_mask=None, document_mask=None):
    if not query or not document:
        raise ValueError('nonempty query and document required')
    dim = len(query[0])
    if dim == 0 or any(len(row)!=dim or not all(math.isfinite(x) for x in row)
                       for row in query+document):
        raise ValueError('finite equal-width vectors required')
    qm = [True]*len(query) if query_mask is None else query_mask
    dm = [True]*len(document) if document_mask is None else document_mask
    if len(qm)!=len(query) or len(dm)!=len(document) or not all(type(v) is bool for v in qm+dm):
        raise ValueError('one boolean mask per position required')
    if not any(qm) or not any(dm):
        raise ValueError('at least one active position on both sides required')
    matrix = [[dot(q,d) for d in document] for q in query]
    contributions = [max(value for value,active in zip(row,dm) if active)
                     for row,active_query in zip(matrix,qm) if active_query]
    return {'score':sum(contributions),'contributions':contributions,'matrix':matrix}


def main():
    e1, e2 = [1.0,0.0], [0.0,1.0]
    diagonal = unit([1.0,1.0])
    query, doc_a, doc_b = [e1,e2], [e1,e2], [diagonal,diagonal]
    a, b = maxsim(query,doc_a), maxsim(query,doc_b)
    pooled_a = dot(pooled_unit(query),pooled_unit(doc_a))
    pooled_b = dot(pooled_unit(query),pooled_unit(doc_b))
    assert math.isclose(pooled_a,1) and math.isclose(pooled_b,1)
    assert math.isclose(a['score'],2) and math.isclose(b['score'],math.sqrt(2))
    # Fixed-vector permutation and exact duplicate invariances, not raw-text claims.
    invariant_cases = 0
    for doc in [doc_a,doc_b,[[0.6,0.8],[-1.0,0.0],e2]]:
        reference = maxsim(query,doc)['score']
        for permutation in itertools.permutations(doc):
            assert math.isclose(maxsim(query,list(permutation))['score'],reference)
            invariant_cases += 1
        for token in doc:
            assert math.isclose(maxsim(query,doc+[token])['score'],reference)
            invariant_cases += 1
    duplicated_query = maxsim(query+[e1],doc_a)['score']
    forward = maxsim([e1,e1],[e1])['score']
    reverse = maxsim([e1],[e1,e1])['score']
    assert forward == 2 and reverse == 1 and duplicated_query == 3
    valid_negative = [-0.6,0.8]
    masked = maxsim([e1],[valid_negative,[0.0,0.0]],document_mask=[True,False])['score']
    bad_padding = maxsim([e1],[valid_negative,[0.0,0.0]])['score']
    assert masked == -0.6 and bad_padding == 0
    masked_query = maxsim(query+[e1],doc_a,query_mask=[True,True,False])['score']
    assert masked_query == a['score']
    assert maxsim(query+[e1],doc_a)['score'] == masked_query + 1
    rejected = 0
    for call in [lambda:maxsim(query,doc_a,document_mask=[False,False]),
                 lambda:maxsim(query,doc_a,query_mask=[False,False]),
                 lambda:maxsim(query,[[1.0]]),
                 lambda:maxsim(query,doc_a,document_mask=[True])]:
        try:
            call()
        except ValueError:
            rejected += 1
    assert rejected == 4
    report={'scope':'fixed vectors only; no text encoder, index, retrieval dataset or training',
        'query':query,'doc_a':doc_a,'doc_b':doc_b,'a':a,'b':b,
        'pooled_cosines':[pooled_a,pooled_b],'invariance_cases':invariant_cases,
        'query_duplicate_score':duplicated_query,'asymmetry_forward':forward,'asymmetry_reverse':reverse,
        'negative_valid_masked_score':masked,'incorrect_zero_padding_score':bad_padding,
        'masked_query_score':masked_query,'rejected_inputs':rejected,'checks_passed':True}
    Path('report.json').write_text(json.dumps(report,indent=2)+'\n',encoding='utf-8')
    print('pooled cosine: A={:.6f} B={:.6f}'.format(pooled_a,pooled_b))
    print('MaxSim: A={:.6f} B={:.6f}'.format(a['score'],b['score']))
    print('A token contributions =',a['contributions'])
    print('B token contributions =',[round(x,6) for x in b['contributions']])
    print('duplicate query token: score={:.6f}'.format(duplicated_query))
    print('asymmetry: S(Q,D)={:.6f} S(D,Q)={:.6f}'.format(forward,reverse))
    print('negative valid match: masked={:.6f} unmasked-zero-padding={:.6f}'.format(masked,bad_padding))
    print('invariance_cases={} rejected_inputs={}'.format(invariant_cases,rejected))
    print('checks=PASS; report.json written')


if __name__ == '__main__':
    main()

怎样运行:在独立练习文件夹中执行 python -B colbert_maxsim.py。2026-10-17 已在 Windows、Python 3.14.0 运行此份源码并通过自带检查。人工二维向量的 MaxSim、非对称性、置换重复与补齐掩码;未调用编码器、建立检索索引、训练模型或评测真实语料。

运行后对照结果
pooled cosine: A=1.000000 B=1.000000
MaxSim: A=2.000000 B=1.414214
A token contributions = [1.0, 1.0]
B token contributions = [0.707107, 0.707107]
duplicate query token: score=3.000000
asymmetry: S(Q,D)=2.000000 S(D,Q)=1.000000
negative valid match: masked=-0.600000 unmasked-zero-padding=0.000000
invariance_cases=17 rejected_inputs=4
checks=PASS; report.json written

多保留一些信息,也增加了存储与评测工作

保留每个词的向量会比每段只保留一个向量需要更多记录。实际 ColBERT 系统还要处理编码、索引、候选召回与最终打分;ColBERTv2 另有训练监督和压缩方案。本文没有实现这些部分,也不从二维算式推断某个真实索引的速度或体积。

想用于 Agent 的资料检索,应另外用真实查询和相关文档标签比较召回率、排序指标、查询耗时与索引成本。这里没有读取私人资料、调用嵌入模型、训练检索器或跑语料评测;人工例子提供的是一份能检查打分语义与掩码错误的最小参照。

记住这些要点

  • MaxSim 为每个查询位置取最佳文档匹配,再相加。
  • 整段均值方向相同,不代表逐词匹配信息相同。
  • 它通常不对称,也没有一对一配对约束。
  • 文档补齐必须在 max 前排除,置零本身不够。

别把例子用过头

  • 把分数超过 1 当成概率溢出错误。
  • 把文档重复不变推广成原始文本重复不变。
  • 认为一个文档词只能被一个查询词匹配。
  • 把所有叫 mask 的位置都删掉,混淆补齐与查询扩展。

自己试一小步

只改变一条有效文档向量,怎样找出哪些查询项的最佳匹配被切换?

  1. 构造三个查询向量与四个有效文档向量,保证向量均已归一化。
  2. 逐项记录最大值和取得最大值的文档位置。
  3. 缓慢旋转其中一个文档向量,找出匹配位置切换的区间。
  4. 加入零补齐及全负相似度案例,比较正确掩码和错误处理。

做到这些,就算完成

  • 同时报告总分、每项贡献和对应位置。
  • 区分唯一最大值与并列最大值的处理。
  • 结论只针对固定向量打分,不外推为真实检索质量。

查原始资料

想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。