传统文本向量模型把整段文本压缩成一个固定长度的向量，检索时只需计算一次向量相似度。多向量模型则保留多个词元或片段级向量，把查询与文档的细粒度匹配推迟到检索阶段完成，这种方法通常称为晚交互（Late Interaction）。
来源摘要未提供具体 API 和模型信息，因此下面不假定某个专用模型接口，而是基于 Sentence Transformers 的词元向量演示晚交互的核心机制。示例可以直接运行，但若用于生产检索，应换成经过晚交互目标训练的模型。
单向量为什么会丢失细节
双编码器通常分别计算查询和文档的单个向量：
q = Encoder(query)
d = Encoder(document)
score = cosine(q, d)
这种设计适合大规模近似最近邻检索：每篇文档只保存一个向量，索引紧凑，打分也很快。但压缩会产生信息瓶颈。例如查询包含产品名、错误码和操作动作时，一个全局向量未必能同时保留这些信号。
多向量模型不急于完成压缩。它为查询生成向量序列 Q = [q1, q2, ...]，也为文档生成向量序列 D = [d1, d2, ...]。典型的 MaxSim 打分方式是：每个查询向量在文档中找到最相似的向量，然后汇总这些最佳匹配。
score(Q, D) = sum_i max_j cosine(q_i, d_j)
这意味着查询中的“连接超时”可以匹配文档中的相关词元，而错误码可以独立匹配另一个位置。查询与文档不必先被压成两个点，交互发生在最终打分阶段。
用 Sentence Transformers 跑通 MaxSim
下面的示例使用通用 Sentence Transformers 模型提取词元向量，再实现一个最小 MaxSim 检索器。
先安装依赖：
python -m pip install -U sentence-transformers torch
创建 late_interaction_demo.py：
from sentence_transformers import SentenceTransformer
import torch
import torch.nn.functional as F
MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
model = SentenceTransformer(MODEL_NAME)
def encode_tokens(texts: list[str]) -> list[torch.Tensor]:
"""返回每段文本的词元向量，形状分别为 [token_count, dimension]。"""
embeddings = model.encode(
texts,
output_value="token_embeddings",
convert_to_tensor=True,
show_progress_bar=False,
)
return [F.normalize(item, p=2, dim=-1) for item in embeddings]
def maxsim(query: torch.Tensor, document: torch.Tensor) -> float:
# 归一化后，矩阵乘法得到所有查询词元与文档词元的余弦相似度。
similarities = query @ document.T
# 每个查询词元选择一个最佳文档词元，再对查询长度取平均。
return similarities.max(dim=1).values.mean().item()
query = "Python HTTP connection timeout"
documents = [
"Configure request timeouts and retry failed HTTP connections in Python.",
"Use CSS grid to build a responsive page layout.",
"A database connection pool can reduce connection setup overhead.",
]
query_vectors = encode_tokens([query])[0]
document_vectors = encode_tokens(documents)
ranking = sorted(
(
(maxsim(query_vectors, vectors), text)
for text, vectors in zip(documents, document_vectors)
),
reverse=True,
)
for score, text in ranking:
print(f"{score:.4f}\t{text}")
运行：
python late_interaction_demo.py
这个程序展示的是打分结构，不是一个经过验证的生产方案。all-MiniLM-L6-v2 主要面向句子级表示，并非专门按 MaxSim 目标训练。生产环境应选用明确支持词元级晚交互的模型，并遵循它要求的查询前缀、文档前缀、池化方式和归一化规则。
从演示代码走向检索系统
直接比较所有查询词元和所有文档词元，计算量大致与 查询词元数 × 文档词元数 成正比。更棘手的是存储：单向量检索每篇文档保存一个向量，而多向量检索可能要保存几十甚至几百个向量。
更实用的架构通常是两阶段检索：
- 用单向量或稀疏检索，从全库召回几十到几百篇候选文档。
- 只对候选集执行多向量 MaxSim，获得更精细的排序。
下面是可以改造进服务端的批量重排函数：
from sentence_transformers import SentenceTransformer
import torch
import torch.nn.functional as F
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
def rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
texts = [query] + [item["text"] for item in candidates]
vectors = model.encode(
texts,
output_value="token_embeddings",
convert_to_tensor=True,
show_progress_bar=False,
)
vectors = [F.normalize(value, p=2, dim=-1) for value in vectors]
query_vectors = vectors[0]
results = []
for candidate, document_vectors in zip(candidates, vectors[1:]):
matrix = query_vectors @ document_vectors.T
score = matrix.max(dim=1).values.mean().item()
results.append({**candidate, "late_interaction_score": score})
return sorted(
results,
key=lambda item: item["late_interaction_score"],
reverse=True,
)[:top_k]
真实系统还需要处理批量张量、GPU 显存、最大文本长度和文档切块。逐篇执行 Python 循环适合验证算法，但吞吐量较高时，应按长度分桶并批量计算，避免短文本被大量填充。
评分规则不只有一种
MaxSim 求和会让查询长度影响分数范围。示例采用平均值，让不同长度的查询更容易比较，但具体模型可能是在求和目标上训练的，擅自修改聚合方式会改变排序分布。
还需要明确处理以下问题：
- 特殊词元是否参与匹配，例如
[CLS]、[SEP]和填充词元。 - 查询词元是否需要权重，停用词是否会产生过多贡献。
- 长文档是否按窗口切块，以及同一文档的多个块如何合并。
- 文档向量是否量化为 FP16、INT8 或更低精度。
- 检索分数是否需要与 BM25、单向量分数做归一化融合。
这些不是外围优化，而是检索质量和成本的组成部分。尤其是填充词元，如果没有通过 attention mask 排除，可能被错误地选为最佳匹配。
采用前的检查清单
多向量晚交互适合关键词、实体、属性和局部语义都很重要的搜索场景，例如技术文档、商品检索和专业知识库。它通常能换来更细粒度的相关性判断，但代价是更大的索引、更复杂的执行路径和更高的重排延迟。
上线前应至少确认：模型确实针对晚交互训练；离线评测使用业务查询和人工相关性标签；索引体积在文档增长后仍可接受；候选集大小与延迟预算匹配；切块、截断和特殊词元规则与训练阶段一致。
如果单向量召回已经稳定，合理的落地顺序是先把晚交互作为候选集重排器，而不是立即替换整个向量索引。这样既能测量真实的排序收益，也能把存储和计算风险控制在较小范围内。