函数签名
from typing import Callable, List, Literal, Optional, Sequence, TypedDict
class SimilarityResult(TypedDict):
a: int
b: int
score: float
method: str
normalized: bool
def pairwise_text_similarity(
texts: Sequence[str],
*,
method: Literal["tfidf", "bert", "sbert"] = "tfidf",
top_k: int = 5,
normalize: bool = True,
threshold: Optional[float] = None,
tokenizer: Optional[Callable[[str], List[str]]] = None,
batch_size: int = 32,
device: Optional[str] = None,
) -> List[SimilarityResult]:
"""
Compute pairwise similarity within a collection of texts for search recall and
content deduplication. Supports TF-IDF or (S)BERT embeddings, optional
preprocessing via tokenizer, threshold filtering, and global Top-K output.
Results are sorted by score in descending order.
This API targets pairwise similarity among all unique text indices in `texts`.
For single-query scenarios, a common pattern is to pass `[query] + corpus`
and post-filter results where `a == 0` to obtain Top-K neighbors of the query.
"""
...
参数说明
- texts: Sequence[str] - 文本集合，长度需≥2；将对其内部所有不重复索引对进行相似度计算
- method: Literal['tfidf','bert','sbert'] - 相似度表示方法，默认 'tfidf'
- top_k: int - 返回的全局Top-K相似对数量（按score降序），范围[1, 50]，默认5
- normalize: bool - 是否对向量进行归一化（如L2归一化以使用余弦相似度），默认True
- threshold: Optional[float] - 分数阈值，范围[0, 1]；若提供，仅返回score≥threshold的结果
- tokenizer: Optional[Callable[[str], List[str]]] - 预处理/分词函数，接收字符串返回token列表；常用于TF-IDF
- batch_size: int - 批处理大小，默认32；主要用于BERT/SBERT嵌入生成的批量推理
- device: Optional[str] - 设备标识，如'cpu'或'cuda'（或实现支持的其他形式）；主要用于BERT/SBERT
返回值说明
- List[SimilarityResult] - 相似对列表，按score降序排序；每个元素包含:
- a: int - 文本A的索引（基于输入texts）
- b: int - 文本B的索引（a < b）
- score: float - 相似度分数
- method: str - 实际使用的方法（'tfidf'|'bert'|'sbert'）
- normalized: bool - 是否使用了向量归一化
若指定threshold，仅返回score≥threshold的结果。
文档字符串框架
"""
计算文本集合内的两两相似度，用于搜索召回与内容去重。
支持TF-IDF或BERT/SBERT嵌入，提供可选的预处理、阈值过滤与全局Top-K输出，
结果按分数降序返回。
参数
----------
texts : Sequence[str]
文本集合，长度需≥2。将对其内部所有不重复索引对(i, j), i < j进行相似度计算。
method : Literal['tfidf','bert','sbert'], optional
相似度表示方法，默认 'tfidf'。'bert' 与 'sbert' 通常依赖深度模型生成句向量。
top_k : int, optional
全局Top-K相似对数量（按score降序），范围[1, 50]，默认5。若候选少于K，则返回所有候选。
normalize : bool, optional
是否对向量进行归一化（如L2归一化以使用余弦相似度），默认True。
threshold : Optional[float], optional
分数阈值，范围[0, 1]。若提供，仅返回score≥threshold的结果；若未提供，则不进行阈值裁剪。
tokenizer : Optional[Callable[[str], List[str]]], optional
文本预处理/分词函数；当使用TF-IDF时常用以自定义分词与清洗逻辑。
batch_size : int, optional
批处理大小，默认32；主要用于'bert'/'sbert'嵌入生成阶段的批量推理。
device : Optional[str], optional
计算设备，如'cpu'或'cuda'（实现可支持诸如'cuda:0'等更具体标识）；主要用于'bert'/'sbert'。
返回
-------
List[SimilarityResult]
相似对结果列表，按score降序排序。每个元素为TypedDict:
- a (int): 文本A的索引（基于输入texts）
- b (int): 文本B的索引（a < b）
- score (float): 相似度分数
- method (str): 实际使用的方法（'tfidf'|'bert'|'sbert'）
- normalized (bool): 是否进行了向量归一化
若指定threshold，仅返回score≥threshold的结果。
异常
-------
ValueError
- texts长度<2
- top_k不在[1, 50]范围内
- threshold不为None且不在[0, 1]范围内
- batch_size < 1
TypeError
- tokenizer的签名不符合 Callable[[str], List[str]] 要求
RuntimeError
- 所选方法所需的运行时依赖/设备不可用（例如深度模型或设备不可用）
NotImplementedError
- 方法或特性在当前实现中未提供
说明
-------
- 本函数对输入集合进行全局两两相似度计算。Top-K指全局最相似的K对，而非按每个文本分别的Top-K。
- 单次查询场景可通过传入 [query] + corpus 的方式，事后筛选 a == 0 的结果以获取查询的Top-K相似文本。
- 复杂度：向量生成通常为 O(N)，两两相似度计算通常为 O(N^2)；大规模数据需注意内存与时间开销。
- normalize为True时通常等价于使用余弦相似度；为False时可能使用内积或实现定义的度量。
示例
-------
# 用法示意（实现细节根据具体工程而定）
# results = pairwise_text_similarity(
# texts=["foo", "bar", "baz", "..."],
# method="tfidf",
# top_k=10,
# normalize=True,
# threshold=0.3,
# tokenizer=None,
# batch_size=64,
# device="cpu",
# )
"""
