From cb6df0283f25cf3aee7b4e2dc870eaa29f2b1d19 Mon Sep 17 00:00:00 2001 From: hylin Date: Wed, 16 Sep 2026 17:42:00 +0800 Subject: [PATCH] fix: keep the highest-scoring duplicate in merge and RRF joins `_merge` and `_reciprocal_rank_fusion` both accumulate a fused score per document id while writing `documents_map[doc.id] = doc` unconditionally, so the copy that survives is whichever list came last. `_concatenate` and `_distribution_based_rank_fusion` instead keep the highest-scoring copy, and `_deduplicate_documents` -- three lines above the RRF helper in the same module -- already implements exactly that rule. Duplicates share an id but not necessarily a payload: two retrievers can return the same document with different content or meta. The fused score in both modes is order-independent, so the mismatch means the score stays stable while the returned content silently depends on connection order: list_a = [Document(id="d1", content="full text", score=0.9)] list_b = [Document(id="d1", content="stub", score=0.1)] merge([list_a, list_b]) -> score 0.5, content "stub" merge([list_b, list_a]) -> score 0.5, content "full text" Keep the higher-scoring copy in both, matching the other two modes. Fixing the shared `_reciprocal_rank_fusion` helper covers every caller, not just `DocumentJoiner`. The score arithmetic is untouched. The `-inf` sort key that `_concatenate` already used is lifted into a small `_score_or_neg_inf` helper so both call sites in the joiner share one definition. --- .../components/joiners/document_joiner.py | 16 ++++++++-- haystack/utils/misc.py | 10 ++++++- ...st-scoring-duplicate-3f8c21ab6d4e7a19.yaml | 11 +++++++ .../joiners/test_document_joiner.py | 29 +++++++++++++++++++ 4 files changed, 63 insertions(+), 3 deletions(-) create mode 100644 releasenotes/notes/document-joiner-keep-highest-scoring-duplicate-3f8c21ab6d4e7a19.yaml diff --git a/haystack/components/joiners/document_joiner.py b/haystack/components/joiners/document_joiner.py index c976e09c81a..1d708147993 100644 --- a/haystack/components/joiners/document_joiner.py +++ b/haystack/components/joiners/document_joiner.py @@ -16,6 +16,11 @@ logger = logging.getLogger(__name__) +def _score_or_neg_inf(document: Document) -> float: + """Sort key that ranks a Document without a score below every scored one.""" + return document.score if document.score is not None else -inf + + class JoinMode(Enum): """ Enum for join mode. @@ -188,7 +193,7 @@ def _concatenate(document_lists: list[list[Document]]) -> list[Document]: for doc in itertools.chain.from_iterable(document_lists): docs_per_id[doc.id].append(doc) for docs in docs_per_id.values(): - doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score is not None else -inf) + doc_with_best_score = max(docs, key=_score_or_neg_inf) output.append(doc_with_best_score) return output @@ -207,7 +212,14 @@ def _merge(self, document_lists: list[list[Document]]) -> list[Document]: for documents, weight in zip(document_lists, weights, strict=True): for doc in documents: scores_map[doc.id] += (doc.score if doc.score is not None else 0) * weight - documents_map[doc.id] = doc + # Duplicates share an id but not necessarily a payload: two retrievers can return + # the same document with different content or meta. Keep the highest-scoring copy, + # as `_concatenate` and `_distribution_based_rank_fusion` do -- overwriting + # unconditionally would keep whichever list happened to come last, making the + # returned payload depend on connection order even though the fused score does not. + incumbent = documents_map.get(doc.id) + if incumbent is None or _score_or_neg_inf(doc) > _score_or_neg_inf(incumbent): + documents_map[doc.id] = doc return [replace(doc, score=scores_map[doc.id]) for doc in documents_map.values()] diff --git a/haystack/utils/misc.py b/haystack/utils/misc.py index c349ea199d8..4bfe01d4d2d 100644 --- a/haystack/utils/misc.py +++ b/haystack/utils/misc.py @@ -180,7 +180,15 @@ def _reciprocal_rank_fusion( for documents, weight in zip(document_lists, resolved_weights, strict=True): for rank, doc in enumerate(documents): scores_map[doc.id] += (weight * len(document_lists)) / (k + rank) - documents_map[doc.id] = doc + # Duplicates share an id but not necessarily a payload, so keep the highest-scoring + # copy the way `_deduplicate_documents` does. Overwriting unconditionally would keep + # whichever list came last, making the returned payload depend on list order even + # though the fused score does not. + incumbent = documents_map.get(doc.id) + if incumbent is None or (doc.score if doc.score is not None else -inf) > ( + incumbent.score if incumbent.score is not None else -inf + ): + documents_map[doc.id] = doc for _id in scores_map: scores_map[_id] /= len(document_lists) / k diff --git a/releasenotes/notes/document-joiner-keep-highest-scoring-duplicate-3f8c21ab6d4e7a19.yaml b/releasenotes/notes/document-joiner-keep-highest-scoring-duplicate-3f8c21ab6d4e7a19.yaml new file mode 100644 index 00000000000..fa92a71e557 --- /dev/null +++ b/releasenotes/notes/document-joiner-keep-highest-scoring-duplicate-3f8c21ab6d4e7a19.yaml @@ -0,0 +1,11 @@ +--- +fixes: + - | + `DocumentJoiner` now keeps the highest-scoring copy of a duplicate document in `merge` and + `reciprocal_rank_fusion` mode, matching what `concatenate` and + `distribution_based_rank_fusion` already do. Both modes previously overwrote the stored + document unconditionally, so when two retrievers returned the same id with different content + or meta, the copy that survived was whichever list happened to come last. The fused score is + order-independent in both modes, so the returned payload silently depended on connection + order while the score did not. The shared `_reciprocal_rank_fusion` helper is fixed too, so + every caller benefits. diff --git a/test/components/joiners/test_document_joiner.py b/test/components/joiners/test_document_joiner.py index c643b8453df..1663366e316 100644 --- a/test/components/joiners/test_document_joiner.py +++ b/test/components/joiners/test_document_joiner.py @@ -188,6 +188,35 @@ def test_run_with_merge_join_mode_handles_zero_score(self): output = joiner.run([documents_1, documents_2]) assert output["documents"][0].score == 0.0 + def test_run_with_merge_join_mode_keeps_highest_scoring_duplicate(self): + # Duplicates share an id but not necessarily a payload. The fused score is a weighted sum + # and so is order-independent; the kept copy must be too, and must match what + # `concatenate` and `distribution_based_rank_fusion` keep -- the highest-scoring one. + high = Document(id="d1", content="full text", score=0.9) + low = Document(id="d1", content="stub", score=0.1) + + for lists in ([high], [low]), ([low], [high]): + output = DocumentJoiner(join_mode="merge", weights=[0.5, 0.5]).run(list(lists)) + assert len(output["documents"]) == 1 + assert output["documents"][0].content == "full text" + + def test_run_with_rrf_join_mode_keeps_highest_scoring_duplicate(self): + high = Document(id="d1", content="full text", score=0.9) + low = Document(id="d1", content="stub", score=0.1) + + for lists in ([high], [low]), ([low], [high]): + output = DocumentJoiner(join_mode="reciprocal_rank_fusion").run(list(lists)) + assert len(output["documents"]) == 1 + assert output["documents"][0].content == "full text" + + def test_run_with_merge_join_mode_keeps_scored_duplicate_over_unscored(self): + scored = Document(id="d1", content="scored", score=0.2) + unscored = Document(id="d1", content="unscored", score=None) + + for lists in ([scored], [unscored]), ([unscored], [scored]): + output = DocumentJoiner(join_mode="merge", weights=[0.5, 0.5]).run(list(lists)) + assert output["documents"][0].content == "scored" + def test_run_with_merge_join_mode(self): joiner = DocumentJoiner(join_mode="merge", weights=[1.5, 0.5]) documents_1 = [Document(content="a", score=1.0), Document(content="b", score=2.0)]