group_fqcd_jr/app/core/retrieval_fusion.py

"""检索融合：多路召回 → 加权评分 → 统一排序（**纯函数，不碰 Milvus、不碰判定分数**）。

## 为什么需要它

一次检索现在只有"一路向量 + 条件触发的字面通道"，且字面通道被 `VECTOR_CONFIDENT_SCORE = 0.75`
挡住。要让两路**并行**并各司其职，就需要一个地方把它们的候选合起来 —— 那就是本模块。

## 🔴 最重要的约束：`fused_score` **不是**判定分数

判定用的阈值是 `HIGH_SCORE` / `MID_SCORE` / `MIN_GAP` / `E4_MIN_SCORE`（都在客服 Agent 里）。
`app/service/agent/implementations/customer_service.py` 里 `MIN_GAP` 的注释写着实测可选取值区间是
**`(0.0649, 0.0759)`**、两侧裕度**各约 0.005**。那个 `gap` 是用**向量 cosine 分**算的，
任何替换它的实现都会把金标题推出窄带（`M-1` 从 100% 掉下来）。

所以本模块严格区分两个分数：

| 字段 | 含义 | 谁用 |
|---|---|---|
| `best_vector_score` | 该块在**向量路**里的原始 cosine（没有则 `None`） | 🔴 **出口判定** |
| `fused_score` | 各路加权后的融合分 | **只用于候选集合的选择与排序** |

**消费方必须用前者做判定、后者做取舍**，两者不可互换。这条不是风格偏好，是回归门禁的硬要求。

## 融合规则（为什么这么定）

1. **同 `doc_id` 合并**，各路得分**加权求和**：`fused = Σ(w_i · s_i)`。
   ⇒ 两路都命中的条目得分更高 —— 这是"共识条目优先"的期望行为（与 RRF 的精神一致）。
   权重之和约定为 `1.0`，这样 `fused_score` 仍落在 `[0, 1]` 区间、与向量分同量纲，便于人工核读。
2. **不引入新分数尺度**：本模块**不做** `1/(k+rank)` 之类的排名变换 —— 一期两路都是 cosine、
   天然同尺度，加权即可（`W33` 方案 §3.2）。等二期引入**异质通道**（字面锚点给满分 `1.0`）时，
   才需要切换成排名融合；那时本模块加一个 `rrf` 策略即可，**消费方接口不变**。
3. **保序稳定**：同分时按"先出现的路 → 路内原顺序"排，保证**可复算**（两次跑同一输入结果一致）。
4. **`degraded` 按"能否区分故障与没找到"合并**（沿用既有纪律）：
   - 全部路失败 ⇒ `degraded=True, reason="all_routes_failed"`
   - 部分路失败但有结果 ⇒ `degraded=True, reason="partial_route_failure"`，**结果仍然返回**
   - 部分路失败且**没有任何结果** ⇒ 同上（**不得**降级成「没找到」）
   - 全部路成功 ⇒ `degraded=False`，结果为空是**正常的"没命中"**

## 边界

- **档位过滤不归本模块管**：送进来的候选必须**已经过 `visibility` 过滤**（由检索层按 `tiers` 做）。
  本模块不得持有 Milvus 客户端、不得自行检索 —— 否则访客内容可经"融合"这一侧被读出来（`INV-1`）。
- 所有异常（缺字段、类型不对）一律**跳过该候选**而不是抛错：融合是**增益**，
  它坏了不能把本来能答的问题变成答不出来。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class RetrievalRoute:
    """一路召回的结果。`hits` 是**工具返回值形态**的 dict 列表。"""

    name: str
    weight: float
    hits: tuple[Any, ...] = ()
    degraded: bool = False
    reason: str = ""


@dataclass(frozen=True)
class FusedHit:
    """融合后的一条候选。**两个分数各有各的用途，消费方别混用。**"""

    doc_id: str
    payload: dict[str, Any]
    #: 只用于"选哪些块、按什么顺序"。
    fused_score: float
    #: 🔴 用于出口判定的原始向量分（该块没出现在向量路时为 `None`）。
    best_vector_score: float | None = None
    #: 该块来自哪些路（长度 > 1 = 多路共识）。
    routes: tuple[str, ...] = ()
    #: 该块在各路里的原始分，便于核读与排查。
    route_scores: dict[str, float] = field(default_factory=dict)


@dataclass(frozen=True)
class FusionOutcome:
    hits: tuple[FusedHit, ...] = ()
    degraded: bool = False
    reason: str = ""
    routes: tuple[str, ...] = ()


#: 向量路的固定名：只有它产出的分数允许被判定层使用（见模块头「两个分数」）。
VECTOR_ROUTE_NAME = "vector"

#: 参与比对的分数阈值：低于它视作噪声，不进融合（沿用 `E4_MIN_SCORE` 的量级太低，
#: 这里只挡"明显是无意义的 0 分"）。
_MIN_USEFUL_SCORE = 0.0


def _score_of(hit: Any) -> float | None:
    """取一条命中的分数；取不到返回 `None`（该条仍可入库，只是不计分）。"""
    if not isinstance(hit, dict):
        return None
    value = hit.get("score")
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    try:
        return float(str(value))
    except (TypeError, ValueError):
        return None


def _doc_id_of(hit: Any) -> str:
    if not isinstance(hit, dict):
        return ""
    return str(hit.get("doc_id") or "")


def fuse(routes: list[RetrievalRoute] | tuple[RetrievalRoute, ...]) -> FusionOutcome:
    """把多路候选融合成一份排序结果。

    `routes` 顺序即"路内原顺序"的优先级（同分时先出现的路在前）。
    """
    # 传进来的每一路都视为**已尝试**：调用方只构造它真的跑过的路。
    # 不按 `hits` 是否为空过滤 —— "跑成功但零命中"是一个**有意义的信号**
    # （它把"部分失败"与"全部失败"区分开），过滤掉它就会把部分失败误报成全部失败。
    usable = list(routes)
    if not usable:
        return FusionOutcome()

    failed = [r for r in usable if r.degraded]
    degraded = bool(failed)
    if failed and len(failed) == len(usable):
        reason = "all_routes_failed"
    elif failed:
        reason = "partial_route_failure"
    else:
        reason = ""

    # doc_id → 累计；用 dict 保序（Python 3.7+ 保插入序）
    acc: dict[str, dict[str, Any]] = {}
    for route in usable:
        weight = float(route.weight)
        for hit in route.hits:
            doc_id = _doc_id_of(hit)
            if not doc_id:
                continue  # 没有 doc_id 的命中无法去重、也无法引用 ⇒ 丢弃
            score = _score_of(hit)
            if score is not None and score < _MIN_USEFUL_SCORE:
                continue
            entry = acc.get(doc_id)
            if entry is None:
                entry = {"payload": hit, "fused": 0.0, "routes": [], "scores": {},
                         "best_vector": None}
                acc[doc_id] = entry
            # 后出现的路不覆盖 payload：先出现者优先（保序稳定、可复算）
            if score is not None:
                entry["fused"] += weight * score
                entry["scores"][route.name] = score
                if route.name == VECTOR_ROUTE_NAME:
                    prev = entry["best_vector"]
                    entry["best_vector"] = score if prev is None else max(prev, score)
            if route.name not in entry["routes"]:
                entry["routes"].append(route.name)

    fused_hits = [
        FusedHit(
            doc_id=doc_id,
            payload=entry["payload"],
            fused_score=float(entry["fused"]),
            best_vector_score=entry["best_vector"],
            routes=tuple(entry["routes"]),
            route_scores=dict(entry["scores"]),
        )
        for doc_id, entry in acc.items()
    ]
    # 融合分降序；同分保持插入序 ⇒ 可复算
    fused_hits.sort(key=lambda item: item.fused_score, reverse=True)
    return FusionOutcome(
        hits=tuple(fused_hits),
        degraded=degraded,
        reason=reason,
        routes=tuple(r.name for r in usable),
    )


#: RRF（排名融合）的平滑常数。取 60 —— 原始论文与主流实现的通行值。
#: 它的作用是把"第 1 名 vs 第 2 名"的差距压到一个不至于过大的量级，
#: 使多路"共同命中但都不靠前"的条目有机会胜出（这正是融合想要的"共识优先"）。
RRF_K = 60


def fuse_rrf(
    routes: list[RetrievalRoute] | tuple[RetrievalRoute, ...],
    *,
    k: int = RRF_K,
) -> FusionOutcome:
    """排名融合（Reciprocal Rank Fusion）：`score = Σ 1 / (k + rank_i)`。

    ## 什么时候用它（而不是 `fuse`）

    一期两路都是 cosine、天然同尺度，加权求和即可（`fuse`）。
    但引入**异质通道**后（例如字面锚点命中给满分 `1.0`），两路分数**不再是同一把尺子**：
    直接加权会让字面路**压平** `gap`，而 `customer_service` 的 `MIN_GAP` 裕度只有 ±0.005。
    此时改用 RRF —— 它**只吃名次、不吃分数**，异质分数被彻底丢弃 ⇒ 不可能污染判定分。

    ## 与 `fuse` 的共同点（消费方接口不变）

    - 返回仍是 `FusionOutcome`，`FusedHit` 字段语义不变；
    - 🔴 `best_vector_score` **仍然只取向量路的原始 cosine**，绝不由 RRF 分替代 ——
      判定分与排序分的分离是本模块的第一约束（见模块头）；
    - `fused_score` 这里装的是 RRF 原始分（量级约 `1/k`），**只用于排序**；
    - `degraded` 的三态合并口径与 `fuse` 逐字一致。

    ⚠️ 注意量纲差异：`fuse` 的 `fused_score` 落在 `[0, 1]`，RRF 的落在约 `(0, 2/k]`。
    两者都**只用于排序**，因此可互替；但**不得**拿二者跨策略比较大小。
    """
    usable = list(routes)
    if not usable:
        return FusionOutcome()

    failed = [r for r in usable if r.degraded]
    degraded = bool(failed)
    if failed and len(failed) == len(usable):
        reason = "all_routes_failed"
    elif failed:
        reason = "partial_route_failure"
    else:
        reason = ""

    smooth = float(k) if k > 0 else 1.0
    acc: dict[str, dict[str, Any]] = {}
    for route in usable:
        seen: set[str] = set()
        for index, hit in enumerate(route.hits, start=1):
            doc_id = _doc_id_of(hit)
            if not doc_id or doc_id in seen:
                continue  # 同一路里重复出现的同一条，只计它最好的那次名次
            seen.add(doc_id)
            entry = acc.get(doc_id)
            if entry is None:
                entry = {"payload": hit, "rrf": 0.0, "routes": [], "best_vector": None}
                acc[doc_id] = entry
            # 后出现的路不覆盖 payload：先出现者优先（保序稳定、可复算）
            entry["rrf"] += 1.0 / (smooth + index)
            if route.name not in entry["routes"]:
                entry["routes"].append(route.name)
            if route.name == VECTOR_ROUTE_NAME:
                score = _score_of(hit)
                if score is not None:
                    prev = entry["best_vector"]
                    entry["best_vector"] = score if prev is None else max(prev, score)

    fused_hits = [
        FusedHit(
            doc_id=doc_id,
            payload=entry["payload"],
            fused_score=float(entry["rrf"]),
            best_vector_score=entry["best_vector"],
            routes=tuple(entry["routes"]),
            route_scores={},  # RRF 不保留各路原始分：它们已被刻意丢弃（异质尺度）
        )
        for doc_id, entry in acc.items()
    ]
    # 融合分降序；同分保持插入序 ⇒ 可复算
    fused_hits.sort(key=lambda item: item.fused_score, reverse=True)
    return FusionOutcome(
        hits=tuple(fused_hits),
        degraded=degraded,
        reason=reason,
        routes=tuple(r.name for r in usable),
    )


def payloads(outcome: FusionOutcome) -> list[dict[str, Any]]:
    """把融合结果还原成"工具返回值形态"的列表，供既有下游（证据包 / 展示）直接消费。

    🔑 **`score` 字段写回的是 `best_vector_score`（有则写），不是 `fused_score`** ——
    这是本模块唯一一处最容易写错的地方：下游的置信判定读的就是这个 `score`，
    若把它换成融合分，`gap` 就会变，`MIN_GAP` 的 ±0.005 裕度立刻被打破。
    融合分单独放在 `fused_score` 里，**只给排序用**。
    """
    out: list[dict[str, Any]] = []
    for hit in outcome.hits:
        item = dict(hit.payload) if isinstance(hit.payload, dict) else {}
        if hit.best_vector_score is not None:
            item["score"] = round(hit.best_vector_score, 4)
        else:
            # 该块不在向量路里（例如纯字面命中）⇒ 不给一个"看起来像判定分"的数，
            # 显式置 0.0 让下游按"无向量证据"处理，而不是拿融合分冒充。
            item["score"] = 0.0
        item["fused_score"] = round(hit.fused_score, 4)
        item["routes"] = list(hit.routes)
        out.append(item)
    return out