fix(retrieval): RRF 修正同 source 内 chunk 重复加分 + canonical doc 选最佳 rank chunk

  - 原实现 score 累加未按 source 去重，同一 recipe 的多个 chunk 在同一路里
    会被反复加分，违反 RRF "每个 ranker 对每个 doc 贡献一次" 的语义
  - canonical doc 原按"输入顺序首次见到"选取，受 ranked_lists 顺序支配；
    改为按全局最小 rank 选取，rank 相同时按 ranked_lists 顺序优先
  - 同时把每个 source 的 chunk 命中次数另存到 metadata.rrf_chunk_hits，
    便于后续分析
  - _rrf_merge 不再 mutate 输入 Document.metadata，返回新 Document 对象

  更新之后在100 题评测集上 MRR@10 0.898 → 0.939，Faithfulness 0.680 → 0.734。

  Addresses review comment in #106

Author: yzh3434

code/C9/rag_modules/hybrid_retrieval.py

@@ -628,55 +628,83 @@ def _rrf_merge(
         k: int = _RRF_K,
     ) -> List[Document]:
         """
-        Reciprocal Rank Fusion: score(d) = Σ_i 1 / (k + rank_i(d))
+        Reciprocal Rank Fusion: score(d) = Σ_i 1 / (k + best_rank_i(d))
 
         Args:
             ranked_lists: 多路 (source_name, ranked_docs) — docs 按相关度降序
             top_k: 最终返回个数
             k: RRF 平滑常数，默认 60（Cormack et al. 2009）
 
         去重 key：node_id 优先，page_content[:200] hash 兜底。
-        合并后 metadata 写入 rrf_score / rrf_sources / final_score。
-        """
-        rrf_scores: Dict[str, float] = {}
-        doc_index: Dict[str, Document] = {}
-        sources: Dict[str, List[str]] = {}
-        ranks_by_source: Dict[str, Dict[str, int]] = {}
 
-        for source_name, ranked_docs in ranked_lists:
+        同 source 内同 doc_id 多次命中（如一道菜的多个 chunk 共享 recipe.nodeId）：
+            - 算分只取该 source 内最佳 rank（最小 rank）一次，避免重复加分
+            - 命中 chunk 数另存到 rrf_chunk_hits，供后续分析
+
+        canonical doc（最终展示给 LLM 的 page_content）：
+            选全局最小 rank 那个 chunk；rank 相同时按 ranked_lists 顺序优先。
+
+        返回的 Document 是新对象，不会 mutate 输入 list 里的 Document。
+        """
+        # doc_id -> source_name -> 该 source 内最小 rank（用于算分）
+        best_rank_per_source: Dict[str, Dict[str, int]] = {}
+        # doc_id -> source_name -> 该 source 内命中 chunk 次数（信息存档）
+        chunk_hits_per_source: Dict[str, Dict[str, int]] = {}
+        # doc_id -> (global_best_rank, source_priority, doc) — 选 canonical doc
+        best_doc_info: Dict[str, Tuple[int, int, Document]] = {}
+
+        for source_priority, (source_name, ranked_docs) in enumerate(ranked_lists):
             for rank, doc in enumerate(ranked_docs, start=1):
                 node_id = doc.metadata.get("node_id")
                 doc_id = (
                     str(node_id) if node_id is not None
                     else f"hash::{hash(doc.page_content[:200])}"
                 )
 
-                contribution = 1.0 / (k + rank)
-                rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + contribution
+                if doc_id not in best_rank_per_source:
+                    best_rank_per_source[doc_id] = {}
+                    chunk_hits_per_source[doc_id] = {}
+
+                curr_best = best_rank_per_source[doc_id].get(source_name)
+                # 如果是第一次出现或者当前rank比记录的更小，则更新
+                if curr_best is None or rank < curr_best:
+                    best_rank_per_source[doc_id][source_name] = rank
+
+                chunk_hits_per_source[doc_id][source_name] = (
+                    chunk_hits_per_source[doc_id].get(source_name, 0) + 1
+                )
 
-                # 第一次见到这个 doc 时记录为 canonical（通常是 rank 较高的那路）
-                if doc_id not in doc_index:
-                    doc_index[doc_id] = doc
-                    sources[doc_id] = []
-                    ranks_by_source[doc_id] = {}
+                new_key = (rank, source_priority)
+                if (
+                    doc_id not in best_doc_info
+                    or new_key < (best_doc_info[doc_id][0], best_doc_info[doc_id][1])
+                ):
+                    best_doc_info[doc_id] = (rank, source_priority, doc)
 
-                if source_name not in sources[doc_id]:
-                    sources[doc_id].append(source_name)
-                    ranks_by_source[doc_id][source_name] = rank
+        # 每个 source 只用 best rank 算一次贡献
+        rrf_scores: Dict[str, float] = {
+            doc_id: sum(1.0 / (k + r) for r in source_ranks.values())
+            for doc_id, source_ranks in best_rank_per_source.items()
+        }
 
-        # 按 RRF score 降序
         sorted_ids = sorted(
             rrf_scores.keys(), key=lambda d: rrf_scores[d], reverse=True
         )
 
         merged: List[Document] = []
         for doc_id in sorted_ids[:top_k]:
-            doc = doc_index[doc_id]
-            doc.metadata["rrf_score"] = rrf_scores[doc_id]
-            doc.metadata["rrf_sources"] = list(sources[doc_id])
-            doc.metadata["rrf_ranks"] = dict(ranks_by_source[doc_id])
-            doc.metadata["final_score"] = rrf_scores[doc_id]
-            merged.append(doc)
+            _, _, source_doc = best_doc_info[doc_id]
+            # 浅 copy metadata，避免 mutate 上游 Document
+            new_metadata = dict(source_doc.metadata)
+            new_metadata["rrf_score"] = rrf_scores[doc_id]
+            new_metadata["rrf_sources"] = list(best_rank_per_source[doc_id].keys())
+            new_metadata["rrf_ranks"] = dict(best_rank_per_source[doc_id])
+            new_metadata["rrf_chunk_hits"] = dict(chunk_hits_per_source[doc_id])
+            new_metadata["final_score"] = rrf_scores[doc_id]
+            merged.append(Document(
+                page_content=source_doc.page_content,
+                metadata=new_metadata,
+            ))
 
         return merged