Coverage for haystack/document_stores/types/filter_policy.py: 85%
65 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 enum import Enum
6from typing import Any, Literal
8from haystack import logging
10logger = logging.getLogger(__name__)
13class FilterPolicy(Enum):
14 """
15 Policy to determine how filters are applied in retrievers interacting with document stores.
16 """
18 # Runtime filters replace init filters during retriever run invocation.
19 REPLACE = "replace"
21 # Runtime filters are merged with init filters, with runtime filters overwriting init values.
22 MERGE = "merge"
24 def __str__(self) -> str:
25 return self.value
27 @staticmethod
28 def from_str(filter_policy: str) -> "FilterPolicy":
29 """
30 Convert a string to a FilterPolicy enum.
32 :param filter_policy: The string to convert.
33 :return: The corresponding FilterPolicy enum.
34 """
35 enum_map = {e.value.lower(): e for e in FilterPolicy}
36 policy = enum_map.get(filter_policy.lower() if filter_policy else "")
37 if policy is None:
38 msg = f"Unknown FilterPolicy type '{filter_policy}'. Supported types are: {list(enum_map.keys())}"
39 raise ValueError(msg)
40 return policy
43def is_comparison_filter(filter_item: dict[str, Any]) -> bool:
44 """
45 Check if the given filter is a comparison filter.
47 :param filter_item: The filter to check.
48 :returns: True if the filter is a comparison filter, False otherwise.
49 """
50 return all(key in filter_item for key in ["field", "operator", "value"])
53def is_logical_filter(filter_item: dict[str, Any]) -> bool:
54 """
55 Check if the given filter is a logical filter.
57 :param filter_item: The filter to check.
58 :returns: True if the filter is a logical filter, False otherwise.
59 """
60 return "operator" in filter_item and "conditions" in filter_item
63def combine_two_logical_filters(
64 init_logical_filter: dict[str, Any], runtime_logical_filter: dict[str, Any]
65) -> dict[str, Any]:
66 """
67 Combine two logical filters.
69 If both operators are `"AND"`, a single `"AND"` holding the conditions of both is returned. If both use the
70 same other operator, both filters are nested under an `"AND"`. Otherwise, `init_logical_filter` is ignored and
71 `runtime_logical_filter` is returned.
73 __Example__:
75 ```python
76 init_logical_filter = {
77 "operator": "AND",
78 "conditions": [
79 {"field": "meta.type", "operator": "==", "value": "article"},
80 {"field": "meta.rating", "operator": ">=", "value": 3},
81 ]
82 }
83 runtime_logical_filter = {
84 "operator": "AND",
85 "conditions": [
86 {"field": "meta.genre", "operator": "IN", "value": ["economy", "politics"]},
87 {"field": "meta.publisher", "operator": "==", "value": "nytimes"},
88 ]
89 }
90 new_filters = combine_two_logical_filters(init_logical_filter, runtime_logical_filter)
91 # Output:
92 {
93 "operator": "AND",
94 "conditions": [
95 {"field": "meta.type", "operator": "==", "value": "article"},
96 {"field": "meta.rating", "operator": ">=", "value": 3},
97 {"field": "meta.genre", "operator": "IN", "value": ["economy", "politics"]},
98 {"field": "meta.publisher", "operator": "==", "value": "nytimes"},
99 ]
100 }
101 ```
103 With `"OR"` on both sides the filters are nested instead of concatenated:
105 ```python
106 init_logical_filter = {
107 "operator": "OR",
108 "conditions": [{"field": "meta.type", "operator": "==", "value": "article"}],
109 }
110 runtime_logical_filter = {
111 "operator": "OR",
112 "conditions": [{"field": "meta.genre", "operator": "==", "value": "economy"}],
113 }
114 new_filters = combine_two_logical_filters(init_logical_filter, runtime_logical_filter)
115 # Output:
116 {"operator": "AND", "conditions": [init_logical_filter, runtime_logical_filter]}
117 ```
118 """
119 if init_logical_filter["operator"] == runtime_logical_filter["operator"]:
120 if init_logical_filter["operator"] == "AND":
121 return {
122 "operator": "AND",
123 "conditions": init_logical_filter["conditions"] + runtime_logical_filter["conditions"],
124 }
125 return {
126 "operator": "AND",
127 "conditions": [
128 {**init_logical_filter, "conditions": list(init_logical_filter["conditions"])},
129 {**runtime_logical_filter, "conditions": list(runtime_logical_filter["conditions"])},
130 ],
131 }
133 logger.warning(
134 "The provided logical operators, {parsed_operator} and {operator}, do not match so the parsed logical "
135 "filter, {init_logical_filter}, will be ignored and only the provided logical filter,{runtime_logical_filter}, "
136 "will be used. Update the logical operators to match to include the parsed filter.",
137 parsed_operator=init_logical_filter["operator"],
138 operator=runtime_logical_filter["operator"],
139 init_logical_filter=init_logical_filter,
140 runtime_logical_filter=runtime_logical_filter,
141 )
142 return {
143 "operator": str(runtime_logical_filter["operator"]),
144 "conditions": list(runtime_logical_filter["conditions"]),
145 }
148def combine_init_comparison_and_runtime_logical_filters(
149 init_comparison_filter: dict[str, Any],
150 runtime_logical_filter: dict[str, Any],
151 logical_operator: Literal["AND", "OR", "NOT"],
152) -> dict[str, Any]:
153 """
154 Combine a runtime logical filter with the init comparison filter using the provided logical_operator.
156 We only add the init_comparison_filter if logical_operator matches the existing
157 runtime_logical_filter["operator"]. Otherwise, we return the runtime_logical_filter unchanged.
159 __Example__:
161 ```python
162 runtime_logical_filter = {
163 "operator": "AND",
164 "conditions": [
165 {"field": "meta.type", "operator": "==", "value": "article"},
166 {"field": "meta.rating", "operator": ">=", "value": 3},
167 ]
168 }
169 init_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"}
170 new_filters = combine_init_comparison_and_runtime_logical_filters(
171 init_comparison_filter, runtime_logical_filter, "AND"
172 )
173 # Output:
174 {
175 "operator": "AND",
176 "conditions": [
177 {"field": "meta.type", "operator": "==", "value": "article"},
178 {"field": "meta.rating", "operator": ">=", "value": 3},
179 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
180 ]
181 }
182 ```
183 """
184 if runtime_logical_filter["operator"] == logical_operator:
185 conditions = list(runtime_logical_filter["conditions"])
186 fields = {c.get("field") for c in conditions}
187 if init_comparison_filter["field"] not in fields:
188 conditions.append(init_comparison_filter)
189 else:
190 logger.warning(
191 "The init filter, {init_filter}, is ignored as the field is already present in the existing "
192 "filters, {filters}.",
193 init_filter=init_comparison_filter,
194 filters=runtime_logical_filter,
195 )
196 return {"operator": str(runtime_logical_filter["operator"]), "conditions": conditions}
198 logger.warning(
199 "The provided logical_operator, {logical_operator}, does not match the logical operator found in "
200 "the runtime filters, {filters_logical_operator}, so the init filter will be ignored.",
201 logical_operator=logical_operator,
202 filters_logical_operator=runtime_logical_filter["operator"],
203 )
204 return {
205 "operator": str(runtime_logical_filter["operator"]),
206 "conditions": list(runtime_logical_filter["conditions"]),
207 }
210def combine_runtime_comparison_and_init_logical_filters(
211 runtime_comparison_filter: dict[str, Any],
212 init_logical_filter: dict[str, Any],
213 logical_operator: Literal["AND", "OR", "NOT"],
214) -> dict[str, Any]:
215 """
216 Combine an init logical filter with the runtime comparison filter using the provided logical_operator.
218 We only add the runtime_comparison_filter if logical_operator matches the existing
219 init_logical_filter["operator"]. Otherwise, we return the runtime_comparison_filter unchanged.
221 __Example__:
223 ```python
224 init_logical_filter = {
225 "operator": "AND",
226 "conditions": [
227 {"field": "meta.type", "operator": "==", "value": "article"},
228 {"field": "meta.rating", "operator": ">=", "value": 3},
229 ]
230 }
231 runtime_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"}
232 new_filters = combine_runtime_comparison_and_init_logical_filters(
233 runtime_comparison_filter, init_logical_filter, "AND"
234 )
235 # Output:
236 {
237 "operator": "AND",
238 "conditions": [
239 {"field": "meta.type", "operator": "==", "value": "article"},
240 {"field": "meta.rating", "operator": ">=", "value": 3},
241 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
242 ]
243 }
244 ```
245 """
246 if init_logical_filter["operator"] == logical_operator:
247 conditions = list(init_logical_filter["conditions"])
248 fields = {c.get("field") for c in conditions}
249 if runtime_comparison_filter["field"] in fields:
250 logger.warning(
251 "The runtime filter, {runtime_filter}, will overwrite the existing filter with the same "
252 "field in the init logical filter.",
253 runtime_filter=runtime_comparison_filter,
254 )
255 conditions = [c for c in conditions if c.get("field") != runtime_comparison_filter["field"]]
256 conditions.append(runtime_comparison_filter)
257 return {"operator": str(init_logical_filter["operator"]), "conditions": conditions}
259 logger.warning(
260 "The provided logical_operator, {logical_operator}, does not match the logical operator found in "
261 "the init logical filter, {filters_logical_operator}, so the init logical filter will be ignored.",
262 logical_operator=logical_operator,
263 filters_logical_operator=init_logical_filter["operator"],
264 )
265 return runtime_comparison_filter
268def combine_two_comparison_filters(
269 init_comparison_filter: dict[str, Any],
270 runtime_comparison_filter: dict[str, Any],
271 logical_operator: Literal["AND", "OR", "NOT"],
272) -> dict[str, Any]:
273 """
274 Combine a comparison filter with the `init_comparison_filter` using the provided `logical_operator`.
276 If `runtime_comparison_filter` and `init_comparison_filter` target the same field, `init_comparison_filter`
277 is ignored and `runtime_comparison_filter` is returned unchanged.
279 __Example__:
281 ```python
282 runtime_comparison_filter = {"field": "meta.type", "operator": "==", "value": "article"},
283 init_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
284 new_filters = combine_two_comparison_filters(
285 init_comparison_filter, runtime_comparison_filter, "AND"
286 )
287 # Output:
288 {
289 "operator": "AND",
290 "conditions": [
291 {"field": "meta.type", "operator": "==", "value": "article"},
292 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
293 ]
294 }
295 ```
296 """
297 if runtime_comparison_filter["field"] == init_comparison_filter["field"]:
298 logger.warning(
299 "The parsed filter, {parsed_filter}, is ignored as the field is already present in the existing "
300 "filters, {filters}.",
301 parsed_filter=init_comparison_filter,
302 filters=runtime_comparison_filter,
303 )
304 return runtime_comparison_filter
306 return {"operator": str(logical_operator), "conditions": [init_comparison_filter, runtime_comparison_filter]}
309def apply_filter_policy(
310 filter_policy: FilterPolicy,
311 init_filters: dict[str, Any] | None = None,
312 runtime_filters: dict[str, Any] | None = None,
313 default_logical_operator: Literal["AND", "OR", "NOT"] = "AND",
314) -> dict[str, Any] | None:
315 """
316 Apply the filter policy to the given initial and runtime filters to determine the final set of filters used.
318 The function combines or replaces the initial and runtime filters based on the specified filter policy.
320 :param filter_policy: The policy to apply when handling the filters. It can be one of the following:
321 - `FilterPolicy.REPLACE`: Runtime filters will replace the initial filters.
322 - `FilterPolicy.MERGE`: Runtime filters will be merged with the initial filters. If there are overlapping keys,
323 values from the runtime filters will overwrite those from the initial filters.
324 :param init_filters: The initial filters set during the initialization of the relevant retriever.
325 :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
326 can change for each query/retriever run invocation.
327 :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
328 :returns: A dictionary containing the resulting filters based on the provided policy.
329 """
330 if filter_policy == FilterPolicy.MERGE and runtime_filters and init_filters:
331 # now we merge filters
332 if is_comparison_filter(init_filters) and is_comparison_filter(runtime_filters):
333 return combine_two_comparison_filters(init_filters, runtime_filters, default_logical_operator)
334 if is_comparison_filter(init_filters) and is_logical_filter(runtime_filters):
335 return combine_init_comparison_and_runtime_logical_filters(
336 init_filters, runtime_filters, default_logical_operator
337 )
338 if is_logical_filter(init_filters) and is_comparison_filter(runtime_filters):
339 return combine_runtime_comparison_and_init_logical_filters(
340 runtime_filters, init_filters, default_logical_operator
341 )
342 if is_logical_filter(init_filters) and is_logical_filter(runtime_filters):
343 return combine_two_logical_filters(init_filters, runtime_filters)
345 return runtime_filters or init_filters