Coverage for haystack/components/rankers/meta_field.py: 100%
119 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
5from collections import defaultdict
6from collections.abc import Callable
7from dataclasses import replace
8from typing import Any, Literal
10from dateutil.parser import parse as date_parse
12from haystack import Document, component, logging
13from haystack.utils.misc import _deduplicate_documents
15logger = logging.getLogger(__name__)
18@component
19class MetaFieldRanker:
20 """
21 Ranks Documents based on the value of their specific meta field.
23 The ranking can be performed in descending order or ascending order.
25 Usage example:
27 ```python
28 from haystack import Document
29 from haystack.components.rankers import MetaFieldRanker
31 ranker = MetaFieldRanker(meta_field="rating")
32 docs = [
33 Document(content="Paris", meta={"rating": 1.3}),
34 Document(content="Berlin", meta={"rating": 0.7}),
35 Document(content="Barcelona", meta={"rating": 2.1}),
36 ]
38 output = ranker.run(documents=docs)
39 docs = output["documents"]
40 assert docs[0].content == "Barcelona"
41 ```
42 """
44 def __init__(
45 self,
46 meta_field: str,
47 weight: float = 1.0,
48 top_k: int | None = None,
49 ranking_mode: Literal["reciprocal_rank_fusion", "linear_score"] = "reciprocal_rank_fusion",
50 sort_order: Literal["ascending", "descending"] = "descending",
51 missing_meta: Literal["drop", "top", "bottom"] = "bottom",
52 meta_value_type: Literal["float", "int", "date"] | None = None,
53 ) -> None:
54 """
55 Creates an instance of MetaFieldRanker.
57 :param meta_field:
58 The name of the meta field to rank by.
59 :param weight:
60 In range [0,1].
61 0 disables ranking by a meta field.
62 0.5 ranking from previous component and based on meta field have the same weight.
63 1 ranking by a meta field only.
64 :param top_k:
65 The maximum number of Documents to return per query.
66 If not provided, the Ranker returns all documents it receives in the new ranking order.
67 :param ranking_mode:
68 The mode used to combine the Retriever's and Ranker's scores.
69 Possible values are 'reciprocal_rank_fusion' (default) and 'linear_score'.
70 Use the 'linear_score' mode only with Retrievers or Rankers that return a score in range [0,1].
71 :param sort_order:
72 Whether to sort the meta field by ascending or descending order.
73 Possible values are `descending` (default) and `ascending`.
74 :param missing_meta:
75 What to do with documents that are missing the sorting metadata field or whose value for it is `None`.
76 Possible values are:
77 - 'drop' will drop the documents entirely.
78 - 'top' will place the documents at the top of the metadata-sorted list
79 (regardless of 'ascending' or 'descending').
80 - 'bottom' will place the documents at the bottom of metadata-sorted list
81 (regardless of 'ascending' or 'descending').
82 :param meta_value_type:
83 Parse the meta value into the data type specified before sorting.
84 This will only work if all meta values stored under `meta_field` in the provided documents are strings.
85 For example, if we specified `meta_value_type="date"` then for the meta value `"date": "2015-02-01"`
86 we would parse the string into a datetime object and then sort the documents by date.
87 The available options are:
88 - 'float' will parse the meta values into floats.
89 - 'int' will parse the meta values into integers.
90 - 'date' will parse the meta values into datetime objects.
91 - 'None' (default) will do no parsing.
92 """
94 self.meta_field = meta_field
95 self.weight = weight
96 self.top_k = top_k
97 self.ranking_mode = ranking_mode
98 self.sort_order = sort_order
99 self.missing_meta = missing_meta
100 self._validate_params(
101 weight=self.weight,
102 top_k=self.top_k,
103 ranking_mode=self.ranking_mode,
104 sort_order=self.sort_order,
105 missing_meta=self.missing_meta,
106 meta_value_type=meta_value_type,
107 )
108 self.meta_value_type = meta_value_type
110 def _validate_params(
111 self,
112 *,
113 weight: float,
114 top_k: int | None,
115 ranking_mode: Literal["reciprocal_rank_fusion", "linear_score"],
116 sort_order: Literal["ascending", "descending"],
117 missing_meta: Literal["drop", "top", "bottom"],
118 meta_value_type: Literal["float", "int", "date"] | None,
119 ) -> None:
120 if top_k is not None and top_k <= 0:
121 raise ValueError(f"top_k must be > 0, but got {top_k}")
123 if weight < 0 or weight > 1:
124 raise ValueError(
125 f"Parameter <weight> must be in range [0,1] but is currently set to '{weight}'.\n'0' disables sorting "
126 "by a meta field, '0.5' assigns equal weight to the previous relevance scores and the meta field, and "
127 "'1' ranks by the meta field only.\nChange the <weight> parameter to a value in range 0 to 1 when "
128 "initializing the MetaFieldRanker."
129 )
131 if ranking_mode not in ["reciprocal_rank_fusion", "linear_score"]:
132 raise ValueError(
133 "The value of parameter <ranking_mode> must be 'reciprocal_rank_fusion' or 'linear_score', but is "
134 f"currently set to '{ranking_mode}'.\nChange the <ranking_mode> value to 'reciprocal_rank_fusion' or "
135 "'linear_score' when initializing the MetaFieldRanker."
136 )
138 if sort_order not in ["ascending", "descending"]:
139 raise ValueError(
140 "The value of parameter <sort_order> must be 'ascending' or 'descending', "
141 f"but is currently set to '{sort_order}'.\n"
142 "Change the <sort_order> value to 'ascending' or 'descending' when initializing the "
143 "MetaFieldRanker."
144 )
146 if missing_meta not in ["drop", "top", "bottom"]:
147 raise ValueError(
148 "The value of parameter <missing_meta> must be 'drop', 'top', or 'bottom', "
149 f"but is currently set to '{missing_meta}'.\n"
150 "Change the <missing_meta> value to 'drop', 'top', or 'bottom' when initializing the "
151 "MetaFieldRanker."
152 )
154 if meta_value_type not in ["float", "int", "date", None]:
155 raise ValueError(
156 "The value of parameter <meta_value_type> must be 'float', 'int', 'date' or None but is "
157 f"currently set to '{meta_value_type}'.\n"
158 "Change the <meta_value_type> value to 'float', 'int', 'date' or None when initializing the "
159 "MetaFieldRanker."
160 )
162 @component.output_types(documents=list[Document])
163 def run(
164 self,
165 documents: list[Document],
166 top_k: int | None = None,
167 weight: float | None = None,
168 ranking_mode: Literal["reciprocal_rank_fusion", "linear_score"] | None = None,
169 sort_order: Literal["ascending", "descending"] | None = None,
170 missing_meta: Literal["drop", "top", "bottom"] | None = None,
171 meta_value_type: Literal["float", "int", "date"] | None = None,
172 ) -> dict[str, Any]:
173 """
174 Ranks a list of Documents based on the selected meta field by:
176 1. Sorting the Documents by the meta field in descending or ascending order.
177 2. Merging the rankings from the previous component and based on the meta field according to ranking mode and
178 weight.
179 3. Returning the top-k documents.
181 Before ranking, documents are deduplicated by their id, retaining only the document with the highest score
182 if a score is present.
184 :param documents:
185 Documents to be ranked.
186 :param top_k:
187 The maximum number of Documents to return per query.
188 If not provided, the top_k provided at initialization time is used.
189 :param weight:
190 In range [0,1].
191 0 disables ranking by a meta field.
192 0.5 ranking from previous component and based on meta field have the same weight.
193 1 ranking by a meta field only.
194 If not provided, the weight provided at initialization time is used.
195 :param ranking_mode:
196 (optional) The mode used to combine the Retriever's and Ranker's scores.
197 Possible values are 'reciprocal_rank_fusion' (default) and 'linear_score'.
198 Use the 'linear_score' mode only with Retrievers or Rankers that return a score in range [0,1].
199 If not provided, the ranking_mode provided at initialization time is used.
200 :param sort_order:
201 Whether to sort the meta field by ascending or descending order.
202 Possible values are `descending` (default) and `ascending`.
203 If not provided, the sort_order provided at initialization time is used.
204 :param missing_meta:
205 What to do with documents that are missing the sorting metadata field or whose value for it is `None`.
206 Possible values are:
207 - 'drop' will drop the documents entirely.
208 - 'top' will place the documents at the top of the metadata-sorted list
209 (regardless of 'ascending' or 'descending').
210 - 'bottom' will place the documents at the bottom of metadata-sorted list
211 (regardless of 'ascending' or 'descending').
212 If not provided, the missing_meta provided at initialization time is used.
213 :param meta_value_type:
214 Parse the meta value into the data type specified before sorting.
215 This will only work if all meta values stored under `meta_field` in the provided documents are strings.
216 For example, if we specified `meta_value_type="date"` then for the meta value `"date": "2015-02-01"`
217 we would parse the string into a datetime object and then sort the documents by date.
218 The available options are:
219 -'float' will parse the meta values into floats.
220 -'int' will parse the meta values into integers.
221 -'date' will parse the meta values into datetime objects.
222 -'None' (default) will do no parsing.
223 :returns:
224 A dictionary with the following keys:
225 - `documents`: List of Documents sorted by the specified meta field.
227 :raises ValueError:
228 If `top_k` is not > 0.
229 If `weight` is not in range [0,1].
230 If `ranking_mode` is not 'reciprocal_rank_fusion' or 'linear_score'.
231 If `sort_order` is not 'ascending' or 'descending'.
232 If `meta_value_type` is not 'float', 'int', 'date' or `None`.
233 """
234 if not documents:
235 return {"documents": []}
237 top_k = self.top_k if top_k is None else top_k
238 weight = weight if weight is not None else self.weight
239 ranking_mode = ranking_mode or self.ranking_mode
240 sort_order = sort_order or self.sort_order
241 missing_meta = missing_meta or self.missing_meta
242 meta_value_type = meta_value_type or self.meta_value_type
243 self._validate_params(
244 weight=weight,
245 top_k=top_k,
246 ranking_mode=ranking_mode,
247 sort_order=sort_order,
248 missing_meta=missing_meta,
249 meta_value_type=meta_value_type,
250 )
252 deduplicated_documents = _deduplicate_documents(documents)
253 # If the weight is 0 then ranking by meta field is disabled and the original documents should be returned
254 if weight == 0:
255 return {"documents": deduplicated_documents[:top_k]}
257 # A None value can't be sorted against real values, so it is handled like a missing field
258 docs_with_meta_field = [doc for doc in deduplicated_documents if doc.meta.get(self.meta_field) is not None]
259 docs_missing_meta_field = [doc for doc in deduplicated_documents if doc.meta.get(self.meta_field) is None]
261 if len(docs_with_meta_field) == 0:
262 logger.warning(
263 "The parameter <meta_field> is currently set to '{meta_field}', but none of the provided "
264 "Documents with IDs {document_ids} have this meta key.\n"
265 "Set <meta_field> to the name of a field that is present within the provided Documents.\n"
266 "Applying the configured <missing_meta> policy instead of ranking.",
267 meta_field=self.meta_field,
268 document_ids=",".join([doc.id for doc in deduplicated_documents]),
269 )
270 if missing_meta == "drop":
271 return {"documents": []}
272 return {"documents": deduplicated_documents[:top_k]}
274 if len(docs_missing_meta_field) > 0:
275 warning_start = (
276 f"The parameter <meta_field> is currently set to '{self.meta_field}' but the Documents "
277 f"with IDs {','.join([doc.id for doc in docs_missing_meta_field])} don't have this meta key.\n"
278 )
280 if missing_meta == "bottom":
281 logger.warning(
282 "{warning_start}Because the parameter <missing_meta> is set to 'bottom', these Documents will be "
283 "placed at the end of the sorting order.",
284 warning_start=warning_start,
285 )
286 elif missing_meta == "top":
287 logger.warning(
288 "{warning_start}Because the parameter <missing_meta> is set to 'top', these Documents will be "
289 "placed at the top of the sorting order.",
290 warning_start=warning_start,
291 )
292 else:
293 logger.warning(
294 "{warning_start}Because the parameter <missing_meta> is set to 'drop', these Documents will be "
295 "removed from the list of retrieved Documents.",
296 warning_start=warning_start,
297 )
299 # If meta_value_type is provided try to parse the meta values
300 parsed_meta = self._parse_meta(docs_with_meta_field=docs_with_meta_field, meta_value_type=meta_value_type)
301 tuple_parsed_meta_and_docs = list(zip(parsed_meta, docs_with_meta_field, strict=True))
303 # Sort the documents by self.meta_field
304 reverse = sort_order == "descending"
305 try:
306 tuple_sorted_by_meta = sorted(tuple_parsed_meta_and_docs, key=lambda x: x[0], reverse=reverse)
307 except TypeError as error:
308 # Return original documents if mixed types that are not comparable are returned (e.g. int and list)
309 logger.warning(
310 "Tried to sort Documents with IDs {document_ids}, but got TypeError with the message: {error}\n"
311 "Returning the <top_k> of the original Documents since meta field ranking is not possible.",
312 document_ids=",".join([doc.id for doc in docs_with_meta_field]),
313 error=error,
314 )
315 return {"documents": deduplicated_documents[:top_k]}
317 # Merge rankings and handle missing meta fields as specified in the missing_meta parameter
318 sorted_by_meta = [doc for meta, doc in tuple_sorted_by_meta]
319 if missing_meta == "bottom":
320 sorted_documents = sorted_by_meta + docs_missing_meta_field
321 sorted_documents = self._merge_rankings(deduplicated_documents, sorted_documents, weight, ranking_mode)
322 elif missing_meta == "top":
323 sorted_documents = docs_missing_meta_field + sorted_by_meta
324 sorted_documents = self._merge_rankings(deduplicated_documents, sorted_documents, weight, ranking_mode)
325 else:
326 sorted_documents = sorted_by_meta
327 sorted_documents = self._merge_rankings(docs_with_meta_field, sorted_documents, weight, ranking_mode)
329 return {"documents": sorted_documents[:top_k]}
331 def _parse_meta(
332 self, docs_with_meta_field: list[Document], meta_value_type: Literal["float", "int", "date"] | None
333 ) -> list[Any]:
334 """
335 Parse the meta values stored under `self.meta_field` for the Documents provided in `docs_with_meta_field`.
336 """
337 if meta_value_type is None:
338 return [d.meta[self.meta_field] for d in docs_with_meta_field]
340 if not all(isinstance(doc.meta[self.meta_field], str) for doc in docs_with_meta_field):
341 logger.warning(
342 "The parameter <meta_value_type> is currently set to '{meta_field}', but not all of meta values in the "
343 "provided Documents with IDs {document_ids} are strings.\n"
344 "Skipping parsing of the meta values.\n"
345 "Set all meta values found under the <meta_field> parameter to strings to use <meta_value_type>.",
346 meta_field=meta_value_type,
347 document_ids=",".join([doc.id for doc in docs_with_meta_field]),
348 )
349 return [d.meta[self.meta_field] for d in docs_with_meta_field]
351 parse_fn: Callable
352 if meta_value_type == "float":
353 parse_fn = float
354 elif meta_value_type == "int":
355 parse_fn = int
356 else:
357 parse_fn = date_parse
359 try:
360 meta_values = [parse_fn(d.meta[self.meta_field]) for d in docs_with_meta_field]
361 except ValueError as error:
362 logger.warning(
363 "Tried to parse the meta values of Documents with IDs {document_ids}, but got ValueError with the "
364 "message: {error}\n"
365 "Skipping parsing of the meta values.",
366 document_ids=",".join([doc.id for doc in docs_with_meta_field]),
367 error=error,
368 )
369 meta_values = [d.meta[self.meta_field] for d in docs_with_meta_field]
371 return meta_values
373 def _merge_rankings(
374 self,
375 documents: list[Document],
376 sorted_documents: list[Document],
377 weight: float,
378 ranking_mode: Literal["reciprocal_rank_fusion", "linear_score"],
379 ) -> list[Document]:
380 """
381 Merge the two different rankings for Documents sorted both by their content and by their meta field.
382 """
383 scores_map: dict = defaultdict(int)
385 if ranking_mode == "reciprocal_rank_fusion":
386 for i, (document, sorted_doc) in enumerate(zip(documents, sorted_documents, strict=True)):
387 scores_map[document.id] += self._calculate_rrf(rank=i) * (1 - weight)
388 scores_map[sorted_doc.id] += self._calculate_rrf(rank=i) * weight
389 elif ranking_mode == "linear_score":
390 for i, (document, sorted_doc) in enumerate(zip(documents, sorted_documents, strict=True)):
391 score = float(0)
392 if document.score is None:
393 logger.warning("The score wasn't provided; defaulting to 0.")
394 elif document.score < 0 or document.score > 1:
395 logger.warning(
396 "The score {score} for Document {document_id} is outside the [0,1] range; defaulting to 0",
397 score=document.score,
398 document_id=document.id,
399 )
400 else:
401 score = document.score
403 scores_map[document.id] += score * (1 - weight)
404 scores_map[sorted_doc.id] += self._calc_linear_score(rank=i, amount=len(sorted_documents)) * weight
406 scored_docs = [replace(doc, score=scores_map[doc.id]) for doc in documents]
408 return sorted(scored_docs, key=lambda doc: doc.score if doc.score else -1, reverse=True)
410 @staticmethod
411 def _calculate_rrf(rank: int, k: int = 61) -> float:
412 """
413 Calculates the reciprocal rank fusion.
415 The constant K is set to 61 (60 was suggested by the original paper, plus 1 as python lists are 0-based and
416 the [paper](https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf) used 1-based ranking).
417 """
418 return 1 / (k + rank)
420 @staticmethod
421 def _calc_linear_score(rank: int, amount: int) -> float:
422 """
423 Calculate the meta field score as a linear score between the greatest and the lowest score in the list.
425 This linear scaling is useful for:
426 - Reducing the effect of outliers
427 - Creating scores that are meaningfully distributed in the range [0,1],
428 similar to scores coming from a Retriever or Ranker.
429 """
430 return (amount - rank) / amount