xsl
aa1287d83d
DeterministicEmbedder（可换 BGE-M3 的接口位）+ kb_chunk 双索引写入 + 硬过滤→向量→关键词→RRF 重排管线 + kb_gap 缺口埋点（T-P2-03/04/06，TD-08）。 实现： - embedding.py：EmbeddingProvider Protocol + DeterministicEmbedder （sha256 种子→归一化 1024 维假向量，同文本同向量/异文本近似正交） + build_embedding_provider 工厂（未知模型名抛 CONFIG_ERROR）。 - repo.py：upsert_chunks（整实体先删后插，逐行带 embedding_model/version， 可重跑判据 T-P2-03）；hybrid_search（§8.3 CTE：硬过滤 published/ entity_types/occasion/avoid_body_shapes → <=> 余弦 topN → ts_rank topN → RRF）；record_gap（咨询锁串行化同 query+filter_ctx 的 hit_count 递增）。 - rerank.py：RRF/权重/新鲜度纯数学（独立于 SQL 以便单测逐项断言）。 - service.py：RetrievalService.search 编排 embed→检索→（空结果）gap 埋点， SearchOutcome.degraded 标记降级。 测试：单测 +30（test_embedding 确定性/维度/归一化/工厂拒绝、test_rerank 公式逐项断言、test_retrieval_service 编排打桩）；集成 +5 （test_hybrid_retrieval：禁忌实体同文最近向量仍零出现、draft 零出现、 关键词路径、换模型整组替换、kb_gap hit_count=2、partial 索引 EXPLAIN 验证）。 全绿：ruff / mypy(retrieval+shared+domain) / lint-imports / 268 单测 / 21 集成。 偏差说明： 1) 新增 rerank.py（任务文档只点名 embedding/repo/service，但重排数学必须 不连库可单测，独立模块是最低成本实现，仍属 packages/retrieval/**）。 2) upsert_chunks 的 upsert 粒度是实体（先删后插）：kb_chunk 无业务自然键， 行级 ON CONFLICT 不可行；换模型名=新版本整组替换，不留旧版本残向量。 3) 禁忌过滤用 COALESCE(attrs->avoid_body_shapes, '[]')兜底：jsonb NULL 的 ? 运算结果是 NULL，直接 NOT 会把未声明禁忌的好实体一并排除。 4) 关键词路过滤用 tsv @@ tsquery 而非 ts_rank>0：ts_rank 对不匹配文本返回 ~1e-20 正 epsilon，阈值过滤会把无关 chunk 全放进关键词路。 5) RRF 分量按 1/(k+1) 归一到 (0,1]，使 0.45/0.2/0.25/0.1 权重量纲可比 （原始倒数 0.45×1/61 与规则项 0.25×1 完全不可比）。
146 lines
5.4 KiB
Python
146 lines
5.4 KiB
Python
"""RRF 融合重排的纯数学部分（架构 §8.3 唯一口径）.
|
||
|
||
刻意与 SQL 分离：重排公式必须能被单测逐项断言（T-P2-04 验收 3），
|
||
混进 repo 就会把数学正确性测试拖进数据库。repo 只负责召回与名次，
|
||
这里把名次变成最终分。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
|
||
#: RRF 常数。60 是业界默认值：名次差异对分数的影响被压成平滑倒数，
|
||
#: 单路召回头部的偏差不至于淹没另一路。
|
||
RRF_K = 60
|
||
|
||
#: 新鲜度半衰期（天）：30 天前的内容得 0.5，60 天前 0.25……
|
||
#: 知识库是慢更新的，半衰期必须显著大于内容评审周期，否则新版本刚
|
||
#: 发布就会压制尚未重审的旧版本。
|
||
FRESHNESS_HALF_LIFE_DAYS = 30.0
|
||
|
||
#: 架构 §8.3 唯一口径：0.45 向量 + 0.2 关键词 + 0.25 规则 + 0.1 新鲜度。
|
||
#: 与 Settings.retrieval_rerank_weights 的默认值保持一致（可被 env 覆盖）。
|
||
DEFAULT_RERANK_WEIGHTS: dict[str, float] = {
|
||
"vector": 0.45,
|
||
"keyword": 0.2,
|
||
"rule": 0.25,
|
||
"freshness": 0.1,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class RecallRow:
|
||
"""混合检索的一行召回结果：两路召回各自的名次，未被召回的一路为 ``None``."""
|
||
|
||
chunk_id: int
|
||
entity_id: int
|
||
chunk_type: str
|
||
text: str
|
||
updated_at: datetime
|
||
vec_rank: int | None = None # 向量路名次（1 起）
|
||
vec_distance: float | None = None # 向量余弦距离，观测用
|
||
kw_rank: int | None = None # 关键词路名次（1 起）
|
||
kw_tsrank: float | None = None # ts_rank 原始分，观测用
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ScoredChunk:
|
||
"""重排后的最终结果。各分量都归一到 [0, 1]，final_score 是加权和。"""
|
||
|
||
chunk_id: int
|
||
entity_id: int
|
||
chunk_type: str
|
||
text: str
|
||
vec_score: float
|
||
kw_score: float
|
||
rule_score: float
|
||
freshness_score: float
|
||
final_score: float
|
||
|
||
|
||
def rrf_component(rank: int | None, k: int = RRF_K) -> float:
|
||
"""单路 RRF 分量 ``1/(k+rank)``，再除以 ``1/(k+1)`` 归一到 (0, 1].
|
||
|
||
rank=1 → 1.0，rank=k+1 → 0.5。若直接用原始倒数，0.45×(1/61) ≈ 0.007
|
||
与规则项 0.25×1.0 完全不可比，架构 §8.3 的权重就失去"相对重要性"的
|
||
含义 —— 归一化只是线性缩放，不改变单路内的排序。
|
||
"""
|
||
if rank is None or rank <= 0:
|
||
return 0.0
|
||
return (k + 1) / (k + rank)
|
||
|
||
|
||
def rrf_score(vec_rank: int | None, kw_rank: int | None, k: int = RRF_K) -> float:
|
||
"""经典 RRF 原始分：``1/(k+rank_vec) + 1/(k+rank_kw)``（观测/调试用）.
|
||
|
||
未被某路召回的 chunk 只累计被召回那一路的倒数。
|
||
"""
|
||
total = 0.0
|
||
for rank in (vec_rank, kw_rank):
|
||
if rank is not None and rank > 0:
|
||
total += 1.0 / (k + rank)
|
||
return total
|
||
|
||
|
||
def freshness_score(updated_at: datetime, now: datetime | None = None) -> float:
|
||
"""新鲜度：``0.5 ** (age_days / 半衰期)``，越新越接近 1.
|
||
|
||
``now`` 缺省取当前 UTC 时刻；显式传入是为了让单测能构造已知年龄。
|
||
未来时间戳（时钟漂移）按 0 天处理，得 1.0。
|
||
"""
|
||
current = now if now is not None else datetime.now(UTC)
|
||
age_seconds = (current - updated_at).total_seconds()
|
||
age_days = max(age_seconds, 0.0) / 86400.0
|
||
# float ** float 在 typeshed 里是 Any（负底数分数幂会出 complex），显式收窄。
|
||
return float(0.5 ** (age_days / FRESHNESS_HALF_LIFE_DAYS))
|
||
|
||
|
||
def rerank(
|
||
rows: Sequence[RecallRow],
|
||
*,
|
||
weights: Mapping[str, float] | None = None,
|
||
rule_boost: Mapping[int, float] | None = None,
|
||
now: datetime | None = None,
|
||
top_k: int | None = None,
|
||
) -> list[ScoredChunk]:
|
||
"""融合重排：``final = w_v×vec + w_k×kw + w_r×rule + w_f×fresh``（架构 §8.3）.
|
||
|
||
- ``vec``/``kw`` 是归一化的 RRF 分量（见 :func:`rrf_component`）；
|
||
- ``rule`` 直接取 ``rule_boost[chunk_id]``（调用方传入的规则权重，
|
||
规则引擎联动属后续任务，管线只认接口）；
|
||
- ``fresh`` 是半衰期衰减分。
|
||
|
||
排序稳定：final_score 降序，同分按 chunk_id 升序 —— 同一数据集重跑
|
||
结果可复现。``top_k`` 为 ``None`` 时不截断。
|
||
"""
|
||
resolved_weights = weights if weights is not None else DEFAULT_RERANK_WEIGHTS
|
||
w_vec = float(resolved_weights.get("vector", 0.0))
|
||
w_kw = float(resolved_weights.get("keyword", 0.0))
|
||
w_rule = float(resolved_weights.get("rule", 0.0))
|
||
w_fresh = float(resolved_weights.get("freshness", 0.0))
|
||
boosts = rule_boost or {}
|
||
|
||
out = []
|
||
for row in rows:
|
||
vec = rrf_component(row.vec_rank)
|
||
kw = rrf_component(row.kw_rank)
|
||
rule = float(boosts.get(row.chunk_id, 0.0))
|
||
fresh = freshness_score(row.updated_at, now)
|
||
out.append(
|
||
ScoredChunk(
|
||
chunk_id=row.chunk_id,
|
||
entity_id=row.entity_id,
|
||
chunk_type=row.chunk_type,
|
||
text=row.text,
|
||
vec_score=vec,
|
||
kw_score=kw,
|
||
rule_score=rule,
|
||
freshness_score=fresh,
|
||
final_score=w_vec * vec + w_kw * kw + w_rule * rule + w_fresh * fresh,
|
||
)
|
||
)
|
||
out.sort(key=lambda c: (-c.final_score, c.chunk_id))
|
||
return out[:top_k] if top_k is not None else out
|