Coverage for haystack/components/joiners/answer_joiner.py: 98%
57 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import itertools
6from collections.abc import Callable
7from enum import Enum
8from math import inf
9from typing import Any
11from haystack import component, default_from_dict, default_to_dict, logging
12from haystack.core.component.types import Variadic
13from haystack.dataclasses.answer import ExtractedAnswer, GeneratedAnswer
15logger = logging.getLogger(__name__)
17AnswerType = GeneratedAnswer | ExtractedAnswer
20class JoinMode(Enum):
21 """
22 Enum for AnswerJoiner join modes.
23 """
25 CONCATENATE = "concatenate"
27 def __str__(self) -> str:
28 return self.value
30 @staticmethod
31 def from_str(string: str) -> "JoinMode":
32 """
33 Convert a string to a JoinMode enum.
34 """
35 enum_map = {e.value: e for e in JoinMode}
36 mode = enum_map.get(string)
37 if mode is None:
38 msg = f"Unknown join mode '{string}'. Supported modes in AnswerJoiner are: {list(enum_map.keys())}"
39 raise ValueError(msg)
40 return mode
43@component
44class AnswerJoiner:
45 """
46 Merges multiple lists of `Answer` objects into a single list.
48 Use this component to combine answers from different Generators into a single list.
49 Currently, the component supports only one join mode: `CONCATENATE`.
50 This mode concatenates multiple lists of answers into a single list.
52 ### Usage example
54 In this example, AnswerJoiner merges answers from two different Generators:
56 ```python
57 from haystack.components.builders import AnswerBuilder
58 from haystack.components.joiners import AnswerJoiner
60 from haystack.core.pipeline import Pipeline
62 from haystack.components.generators.chat import OpenAIResponsesChatGenerator
63 from haystack.dataclasses import ChatMessage
66 query = "What's Natural Language Processing?"
67 messages = [ChatMessage.from_system("You are a helpful, respectful and honest assistant. Be super concise."),
68 ChatMessage.from_user(query)]
70 pipe = Pipeline()
71 pipe.add_component("llm_1", OpenAIResponsesChatGenerator())
72 pipe.add_component("llm_2", OpenAIResponsesChatGenerator())
73 pipe.add_component("aba", AnswerBuilder())
74 pipe.add_component("abb", AnswerBuilder())
75 pipe.add_component("joiner", AnswerJoiner())
77 pipe.connect("llm_1.replies", "aba")
78 pipe.connect("llm_2.replies", "abb")
79 pipe.connect("aba.answers", "joiner")
80 pipe.connect("abb.answers", "joiner")
82 results = pipe.run(data={"llm_1": {"messages": messages},
83 "llm_2": {"messages": messages},
84 "aba": {"query": query},
85 "abb": {"query": query}})
86 ```
87 """
89 def __init__(
90 self, join_mode: str | JoinMode = JoinMode.CONCATENATE, top_k: int | None = None, sort_by_score: bool = False
91 ) -> None:
92 """
93 Creates an AnswerJoiner component.
95 :param join_mode:
96 Specifies the join mode to use. Available modes:
97 - `concatenate`: Concatenates multiple lists of Answers into a single list.
98 :param top_k:
99 The maximum number of Answers to return. Must be `None` or greater than 0.
100 :param sort_by_score:
101 If `True`, sorts the answers by score in descending order.
102 If an answer has no score, it is handled as if its score is -infinity.
104 :raises ValueError:
105 If `top_k` is not `None` and is less than or equal to 0.
106 """
107 if top_k is not None and top_k <= 0:
108 raise ValueError("top_k must be greater than 0.")
109 if isinstance(join_mode, str):
110 join_mode = JoinMode.from_str(join_mode)
111 join_mode_functions: dict[JoinMode, Callable[[list[list[AnswerType]]], list[AnswerType]]] = {
112 JoinMode.CONCATENATE: self._concatenate
113 }
114 self.join_mode_function: Callable[[list[list[AnswerType]]], list[AnswerType]] = join_mode_functions[join_mode]
115 self.join_mode = join_mode
116 self.top_k = top_k
117 self.sort_by_score = sort_by_score
119 @component.output_types(answers=list[AnswerType])
120 def run(self, answers: Variadic[list[AnswerType]], top_k: int | None = None) -> dict[str, Any]:
121 """
122 Joins multiple lists of Answers into a single list depending on the `join_mode` parameter.
124 If the instance was created with `sort_by_score=True`, the merged Answers are sorted by
125 score in descending order before `top_k` is applied; Answers without a score are handled
126 as if their score were -infinity. Otherwise, the input order is preserved.
128 :param answers:
129 Nested list of Answers to be merged.
131 :param top_k:
132 The maximum number of Answers to return. Overrides the instance's `top_k` if provided.
133 A value of 0 returns no answers. Must not be negative.
135 :returns:
136 A dictionary with the following keys:
137 - `answers`: Merged list of Answers, sorted by score if `sort_by_score` was set to
138 `True` on the instance, otherwise in input order
140 :raises ValueError:
141 If `top_k` is negative.
142 """
143 answers_list = list(answers)
144 join_function = self.join_mode_function
145 output_answers: list[AnswerType] = join_function(answers_list)
147 if self.sort_by_score:
148 output_answers = sorted(
149 output_answers,
150 key=lambda answer: score if (score := getattr(answer, "score", None)) is not None else -inf,
151 reverse=True,
152 )
153 if any(getattr(answer, "score", None) is None for answer in output_answers):
154 logger.info(
155 "Some of the Answers AnswerJoiner got have score=None. It was configured to sort Answers by "
156 "score, so those with score=None were sorted as if they had a score of -infinity."
157 )
159 if top_k is not None:
160 if top_k < 0:
161 raise ValueError("top_k must not be negative.")
162 output_answers = output_answers[:top_k]
163 elif self.top_k is not None:
164 output_answers = output_answers[: self.top_k]
165 return {"answers": output_answers}
167 def _concatenate(self, answer_lists: list[list[AnswerType]]) -> list[AnswerType]:
168 """
169 Concatenate multiple lists of Answers, flattening them into a single list.
171 :param answer_lists: List of lists of Answers to be flattened.
172 """
173 return list(itertools.chain.from_iterable(answer_lists))
175 def to_dict(self) -> dict[str, Any]:
176 """
177 Serializes the component to a dictionary.
179 :returns:
180 Dictionary with serialized data.
181 """
182 return default_to_dict(self, join_mode=str(self.join_mode), top_k=self.top_k, sort_by_score=self.sort_by_score)
184 @classmethod
185 def from_dict(cls, data: dict[str, Any]) -> "AnswerJoiner":
186 """
187 Deserializes the component from a dictionary.
189 :param data:
190 The dictionary to deserialize from.
191 :returns:
192 The deserialized component.
193 """
194 return default_from_dict(cls, data)