开始前,会这些就够了
- 知道点积和单位向量
- 知道文本可以被编码为向量
- 能读二维列表与 max、sum
- 需要补基础时:先补向量、点积与范数
遇到不熟悉的词,先看这 2 个解释
打开完整术语小抄 ↗查询里有两个关注点:一个沿横轴,一个沿纵轴。文档 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 函数计算的是点积,主例有效向量已经单位化;如果传入非单位向量,结果不再自动具有余弦的意义。
展开收起代码 · 小实验
长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。
"""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 的位置都删掉,混淆补齐与查询扩展。
自己试一小步
只改变一条有效文档向量,怎样找出哪些查询项的最佳匹配被切换?
- 构造三个查询向量与四个有效文档向量,保证向量均已归一化。
- 逐项记录最大值和取得最大值的文档位置。
- 缓慢旋转其中一个文档向量,找出匹配位置切换的区间。
- 加入零补齐及全负相似度案例,比较正确掩码和错误处理。
做到这些,就算完成
- 同时报告总分、每项贡献和对应位置。
- 区分唯一最大值与并列最大值的处理。
- 结论只针对固定向量打分,不外推为真实检索质量。
查原始资料
想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。
- ColBERT: Efficient and Effective Passage Search via Contextualized Late Interaction over BERT (新标签页)↗
后期交互、逐词
MaxSim、单位化向量和查询扩展的语义。 - ColBERT 官方 colbert.py 源码 (新标签页)↗
文档掩码在最大值归约前应用,随后按查询项求和。
- ColBERTv2: Effective and Efficient Retrieval via Lightweight Late Interaction (新标签页)↗
多向量表示与评分之外的训练、压缩和索引边界。