Pathway LLM xpack 重排序实战：LLMReranker、CrossEncoderReranker 与 EncoderReranker 的原理与用法

在 RAG 流水线中，初检召回（稀疏检索 / 向量检索）只解决"找到候选文档"的问题，不保证"排在前面的文档都相关"。Pathway Live Data Framework 的 LLM xpack（rerankers 模块）提供了三种即插即用的重排序器（Reranker）：基于 LLM 打分的 LLMReranker、基于 CrossEncoder 的 CrossEncoderReranker，以及基于 SentenceTransformer 编码器的 EncoderReranker。读完本篇，你将掌握每种重排序器的适用场景、完整可运行的 Pipeline 示例、关键参数（如激活函数、缓存策略）的取值方式，以及底层打分逻辑的源码级实现，能在 Pathway 流式管道中直接落地"召回 → 重排 → Top-K 过滤"的完整 RAG 检索链路。

为什么 RAG 需要 Reranker

在 RAG 系统中，初始的稀疏检索通常基于余弦相似度。这类检索往往把整段文档的语义压缩进单个向量，容易忽略细粒度的语义差别，导致召回结果中混入与查询不相关的文档。它快，但不准。

常见的改进方式是引入重排序：用更精细的模型对初始候选集重新评估、排序，只保留最相关的文档。Reranker 的核心工作方式是把 (query, document) 作为配对输入给模型，由模型判断给定的 document 对回答 query 有多大贡献，并输出一个相关度分数。

Pathway LLM xpack 提供的重排序器包括（来源：官方用户指南）：

重排序器	类型	底层实现	分数含义
LLMReranker	Pointwise（逐条打分）	调用外部 LLM API	1~5 整数相关度（转 float）
CrossEncoderReranker	Pointwise	sentence_transformers 的 CrossEncoder	0~1（带激活函数）或原始 logits
EncoderReranker	Pointwise	sentence_transformers 的 SentenceTransformer	归一化向量的内积（余弦相似度）

三者都以 (doc, query) 两列表达式的形式调用，返回一列 float 分数，可无缝嵌入 Pathway 表管道（Table Pipeline）。

LLMReranker：让 LLM 充当相关性裁判

LLMReranker 让指定的 LLM 对"给定文档对给定查询的相关度"按 1~5 分打分。典型用法如下（示例继承自用户指南，其中 API_KEY 需替换为实际密钥）：

from pathway.xpacks.llm import rerankers
from pathway.xpacks.llm import llms
import pandas as pd
import pathway as pw

docs = [
    {"text": "John drinks coffee"},
    {"text": "Someone drinks tea"},
    {"text": "Nobody drinks coca-cola"},
]

query = "What does John drink?"

df = pd.DataFrame({"docs": docs, "prompt": query})

chat = llms.OpenAIChat(
    model="gpt-4o-mini",
    api_key=API_KEY,
    response_format="{'type': 'json_object'}",  # 强制 JSON 输出，保证解析稳定
)
reranker = rerankers.LLMReranker(llm=chat)

input = pw.debug.table_from_pandas(df)
res = input.select(rank=reranker(pw.this.docs["text"], pw.this.prompt))

几个值得注意的细节：

• 列表达式顺序是 (doc, query)：reranker 是可调用的对象，第一个参数是文档（这里通过 pw.this.docs["text"] 从 JSON 列中提取 text 字段），第二个参数是查询。
• response_format 的作用：示例中对 OpenAIChat 显式传入 JSON 对象输出格式。这不是巧合——默认的响应解析器依赖严格的 JSON 输出（见下文），强制 JSON 模式能显著降低解析失败概率。
• temperature=0：从源码看，LLMReranker.__call__ 在调用 LLM 时固定使用 temperature=0（rerankers.py），以换取打分的确定性。

构造函数参数

LLMReranker(
    llm,                    # 必填：llms.BaseChat 实例，重排期间被调用的 Chat
    *,
    prompt_template = prompts.prompt_rerank,   # str | Callable[[str, str], str] | pw.UDF
    response_parser = prompts.parse_score_json,  # pw.UDF | Callable[[str], float]
)

（参数定义见 rerankers.py）

prompt_template 可以是字符串模板、普通函数或 pw.UDF，接收 (doc, query) 两个参数。默认模板 prompts.prompt_rerank 生成的提示词要求模型扮演 RAG 助手，按 1~5 分给文档打分（5 分表示问题可以基于该文档回答且文档非常有帮助，1 分表示文档与问题完全无关），并且只输出形如 {"score": <int>} 的 jsonl 响应：

You are a helper agent for RAG applications.
Given a question, and a context document,
rate the document's relevance for the question on a scale between 1 and 5.
...
Reply in jsonl format according to the following format:
{"score": <int>}
Do not output any other text apart from the jsonl response.
-- Context documents --
{context}
...

response_parser 用于解析 LLM 返回的文本并转成 float。默认的 prompts.parse_score_json 用 json.loads 读取 score 字段；如果 LLM 返回的不是预期 JSON，会抛出 ValueError（提示中包含原始返回文本，便于排查）。

异步路径：从源码结构看，如果传入的 LLM 执行器是 FullyAsyncExecutor，LLMReranker 会用 _coerce_fully_async 包装解析器，使打分解析与 LLM 异步调用保持一致的并发模型（rerankers.py）；否则走同步解析。

CrossEncoderReranker：本地 CrossEncoder 打分

CrossEncoderReranker 工作在文本对上，输出 0~1 的分数（如果未传激活函数，则输出原始 logits）。分数越高表示文档对查询越相关。它是 pw.UDF 的子类，模型来自 sentence_transformers 库的 CrossEncoder（模型加载发生在 optional_imports("xpack-llm-local") 作用域内，即依赖 pathway-xpack-llm-local 扩展包提供）。

from pathway.xpacks.llm import rerankers
import pandas as pd
import torch
import pathway as pw

docs = [
    {"text": "John drinks coffee"},
    {"text": "Someone drinks tea"},
    {"text": "Nobody drinks coca-cola"},
]

query = "What does John drink?"

df = pd.DataFrame({"docs": docs, "prompt": query})

reranker = rerankers.CrossEncoderReranker(
    model_name="cross-encoder/ms-marco-MiniLM-L-6-v2",
    default_activation_function=torch.nn.Sigmoid(),  # 使输出落在 0..1
)

input = pw.debug.table_from_pandas(df)
res = input.select(
    rank=reranker(pw.this.docs["text"], pw.this.prompt), text=pw.this.docs["text"]
)
pw.debug.compute_and_print(res)

关键点：

• default_activation_function=torch.nn.Sigmoid()：CrossEncoder 的原始输出是 logits，未经归一化。传入 Sigmoid 后输出被压到 0~1 区间，可以直接与"相关度"对应；不传该参数时得到的是原始 logits，只能用于同一模型内的相对排序，不能跨模型比较绝对值。
• model_name 指定预训练 CrossEncoder 模型名，官方文档建议参考 cross-encoder/ms-marco-TinyBERT-L-2-v2 等更轻量的模型（类文档见 rerankers.py），用户指南示例则使用 cross-encoder/ms-marco-MiniLM-L-6-v2。
• cache_strategy：构造函数支持 cache_strategy 参数（默认 None），用于配置 UDF 级缓存策略，对重复的 (doc, query) 输入可避免重复推理。
• 其余 **init_kwargs 会透传给 CrossEncoder 构造函数，因此上例中的 default_activation_function 正是通过这条通道生效的（rerankers.py）。

打分逻辑非常直接：model.predict([[query, doc]]) 对单个 (query, doc) 对做前向推理，返回第一个分数（rerankers.py）。

EncoderReranker：双塔编码 + 余弦相似度

EncoderReranker 使用 SentenceTransformer 编码器计算查询与文档的相关度。与 CrossEncoder 把 (query, doc) 拼在一起联合编码不同，Encoder 方案对 query 和 doc 分别编码，再计算相似度——这正是初检阶段常用的"单向量"范式，用它做重排时精度介于 CrossEncoder 与初检之间，但单条编码结果可以缓存复用。

from pathway.xpacks.llm import rerankers
import pandas as pd
import pathway as pw

docs = [
    {"text": "John drinks coffee"},
    {"text": "Someone drinks tea"},
    {"text": "Nobody drinks coca-cola"},
]

query = "What does John drink?"

df = pd.DataFrame({"docs": docs, "prompt": query})

reranker = rerankers.EncoderReranker(
    model_name="all-mpnet-base-v2",
)

input = pw.debug.table_from_pandas(df)
res = input.select(
    rank=reranker(pw.this.docs["text"], pw.this.prompt), text=pw.this.docs["text"]
)

从源码看，其打分实现是：对 [query, doc] 两个句子调用 model.encode(..., normalize_embeddings=True) 得到归一化向量，再返回两者的内积 embeddings[0] @ embeddings[1].T——由于向量已归一化，该内积就是余弦相似度（rerankers.py）。

参数与 CrossEncoderReranker 完全同构：model_name（必填）、cache_strategy（默认 None）、其余 kwargs 透传给 SentenceTransformer。类文档中建议的中英文模型示例是 BAAI/bge-large-zh-v1.5，用户指南英文示例用的是 all-mpnet-base-v2。

补充：源码中还包含一个 FlashRankReranker（rerankers.py），基于第三方 flashrank 库实现，默认模型 ms-marco-TinyBERT-L-2-v2、max_length=512；当前用户指南尚未收录，使用时请以实际安装的版本源码为准。

重排之后：用 rerank_topk_filter 做 Top-K 过滤

重排序的落地价值不仅在于"打分"，还在于"筛选"。rerankers 模块自带一个 UDF rerank_topk_filter，按分数对文档列做降序排序并只保留前 k 个，返回 (docs, scores) 元组：

from pathway.xpacks.llm import rerankers
import pandas as pd
import pathway as pw

retrieved_docs = [
    {"text": "Something"},
    {"text": "Something else"},
    {"text": "Pathway"},
]
df = pd.DataFrame({"docs": retrieved_docs, "reranker_scores": [1.0, 3.0, 2.0]})
table = pw.debug.table_from_pandas(df)

# 按 (docs, scores) 逐行聚合后过滤 top-2
docs_table = table.reduce(
    doc_list=pw.reducers.tuple(pw.this.docs),
    score_list=pw.reducers.tuple(pw.this.reranker_scores),
).select(
    docs_scores_tuple=rerankers.rerank_topk_filter(pw.this.doc_list, pw.this.score_list, 2)
).select(
    doc_list=pw.this.docs_scores_tuple[0],
    score_list=pw.this.docs_scores_tuple[1],
)

参数说明（来自源码 docstring）：

• docs：文档（或 chunk）列表列，每行独立过滤；
• scores：与 docs 一一对应的重排分数列表列；
• k：过滤后保留的文档数（默认 5）。

实现上就是 sorted(zip(docs, scores), key=lambda tup: tup[1], reverse=True) 后截断，并把结果以 tuple 列形式写回（rerankers.py）。

测试用例中的行为验证

仓库中的测试 test_rerankers.py 用模拟 LLM 锁定了 LLMReranker 的契约行为，可以直接作为预期行为依据：

• LLM 返回 '{"score": 1}' / '{"score": 5}' 时，await_futures() 后的表中 rank 列分别为 1.0 / 5.0——即"字符串分数"被解析为 float 列；
• LLM 返回非 JSON 文本（如 "text"）时，计算阶段抛出 ValueError，与 parse_score_json 的失败路径一致；
• 使用 pw.udfs.fully_async_executor() 的 Chat 走 fully-async 路径，打分结果同样正确（test_rerankers.py）；
• rerank_topk_filter 测试对 10 篇文档、10 个分数取 top-3，断言保留的是分数为 [9.5, 9.5, 5.555] 的三篇文档（test_rerankers.py），验证了降序排序与截断语义。

选型建议与工程注意事项

结合源码与测试，给出以下落地建议：

1. 质量优先、成本不敏感：用 LLMReranker + 强模型，并像指南示例那样给 Chat 设置 JSON 输出格式，避免 parse_score_json 抛错。1~5 分制天然是离散档位，便于按分数阈值截断。
2. 本地模型、精度优先：用 CrossEncoderReranker，传入 torch.nn.Sigmoid() 激活以获得可比的 0~1 分数；候选集较小时，逐对前向推理的成本可接受。
3. 本地模型、速度优先：用 EncoderReranker，其分数就是余弦相似度，可直接与初检分数比较；query 侧向量可缓存。
4. 统一管道形态：三种重排序器调用签名一致（reranker(doc_expr, query_expr) -> float 列），可以在同一张表上并存多列分数做 A/B 比较；最终用 rerank_topk_filter 完成 Top-K 截断。
5. 依赖前提：CrossEncoderReranker 与 EncoderReranker 在构造时通过 optional_imports("xpack-llm-local") 导入 sentence_transformers，需要安装对应的本地模型扩展包；LLMReranker 只依赖能调通的 llms.BaseChat 实例。
6. 缓存：两个本地重排序器都支持 cache_strategy 参数，对重复输入的流式场景（例如同一 query 在多张表上重放）可以显著降低推理次数。

以上全部重排序器均位于 python/pathway/xpacks/llm/rerankers.py，由 xpacks/llm 包 以 pathway.xpacks.llm.rerankers 形式导出，配合同包的 llms、embedders、vector_store 等模块即可搭建完整的流式 RAG 检索链路。