Coverage for haystack/document_stores/types/filter_policy.py: 83%
65 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +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, they must have the same operator.
69 If `init_logical_filter["operator"]` and `runtime_logical_filter["operator"]` are the same, the conditions
70 of both filters are combined. Otherwise, the `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(
91 init_logical_filter, runtime_logical_filter, "AND"
92 )
93 # Output:
94 {
95 "operator": "AND",
96 "conditions": [
97 {"field": "meta.type", "operator": "==", "value": "article"},
98 {"field": "meta.rating", "operator": ">=", "value": 3},
99 {"field": "meta.genre", "operator": "IN", "value": ["economy", "politics"]},
100 {"field": "meta.publisher", "operator": "==", "value": "nytimes"},
101 ]
102 }
103 ```
104 """
105 if init_logical_filter["operator"] == runtime_logical_filter["operator"]:
106 return {
107 "operator": str(init_logical_filter["operator"]),
108 "conditions": init_logical_filter["conditions"] + runtime_logical_filter["conditions"],
109 }
111 logger.warning(
112 "The provided logical operators, {parsed_operator} and {operator}, do not match so the parsed logical "
113 "filter, {init_logical_filter}, will be ignored and only the provided logical filter,{runtime_logical_filter}, "
114 "will be used. Update the logical operators to match to include the parsed filter.",
115 parsed_operator=init_logical_filter["operator"],
116 operator=runtime_logical_filter["operator"],
117 init_logical_filter=init_logical_filter,
118 runtime_logical_filter=runtime_logical_filter,
119 )
120 runtime_logical_filter["operator"] = str(runtime_logical_filter["operator"])
121 return runtime_logical_filter
124def combine_init_comparison_and_runtime_logical_filters(
125 init_comparison_filter: dict[str, Any],
126 runtime_logical_filter: dict[str, Any],
127 logical_operator: Literal["AND", "OR", "NOT"],
128) -> dict[str, Any]:
129 """
130 Combine a runtime logical filter with the init comparison filter using the provided logical_operator.
132 We only add the init_comparison_filter if logical_operator matches the existing
133 runtime_logical_filter["operator"]. Otherwise, we return the runtime_logical_filter unchanged.
135 __Example__:
137 ```python
138 runtime_logical_filter = {
139 "operator": "AND",
140 "conditions": [
141 {"field": "meta.type", "operator": "==", "value": "article"},
142 {"field": "meta.rating", "operator": ">=", "value": 3},
143 ]
144 }
145 init_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"}
146 new_filters = combine_init_comparison_and_runtime_logical_filters(
147 init_comparison_filter, runtime_logical_filter, "AND"
148 )
149 # Output:
150 {
151 "operator": "AND",
152 "conditions": [
153 {"field": "meta.type", "operator": "==", "value": "article"},
154 {"field": "meta.rating", "operator": ">=", "value": 3},
155 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
156 ]
157 }
158 ```
159 """
160 if runtime_logical_filter["operator"] == logical_operator:
161 conditions = runtime_logical_filter["conditions"]
162 fields = {c.get("field") for c in conditions}
163 if init_comparison_filter["field"] not in fields:
164 conditions.append(init_comparison_filter)
165 else:
166 logger.warning(
167 "The init filter, {init_filter}, is ignored as the field is already present in the existing "
168 "filters, {filters}.",
169 init_filter=init_comparison_filter,
170 filters=runtime_logical_filter,
171 )
172 return {"operator": str(runtime_logical_filter["operator"]), "conditions": conditions}
174 logger.warning(
175 "The provided logical_operator, {logical_operator}, does not match the logical operator found in "
176 "the runtime filters, {filters_logical_operator}, so the init filter will be ignored.",
177 logical_operator=logical_operator,
178 filters_logical_operator=runtime_logical_filter["operator"],
179 )
180 runtime_logical_filter["operator"] = str(runtime_logical_filter["operator"])
181 return runtime_logical_filter
184def combine_runtime_comparison_and_init_logical_filters(
185 runtime_comparison_filter: dict[str, Any],
186 init_logical_filter: dict[str, Any],
187 logical_operator: Literal["AND", "OR", "NOT"],
188) -> dict[str, Any]:
189 """
190 Combine an init logical filter with the runtime comparison filter using the provided logical_operator.
192 We only add the runtime_comparison_filter if logical_operator matches the existing
193 init_logical_filter["operator"]. Otherwise, we return the runtime_comparison_filter unchanged.
195 __Example__:
197 ```python
198 init_logical_filter = {
199 "operator": "AND",
200 "conditions": [
201 {"field": "meta.type", "operator": "==", "value": "article"},
202 {"field": "meta.rating", "operator": ">=", "value": 3},
203 ]
204 }
205 runtime_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"}
206 new_filters = combine_runtime_comparison_and_init_logical_filters(
207 runtime_comparison_filter, init_logical_filter, "AND"
208 )
209 # Output:
210 {
211 "operator": "AND",
212 "conditions": [
213 {"field": "meta.type", "operator": "==", "value": "article"},
214 {"field": "meta.rating", "operator": ">=", "value": 3},
215 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
216 ]
217 }
218 ```
219 """
220 if init_logical_filter["operator"] == logical_operator:
221 conditions = init_logical_filter["conditions"]
222 fields = {c.get("field") for c in conditions}
223 if runtime_comparison_filter["field"] in fields:
224 logger.warning(
225 "The runtime filter, {runtime_filter}, will overwrite the existing filter with the same "
226 "field in the init logical filter.",
227 runtime_filter=runtime_comparison_filter,
228 )
229 conditions = [c for c in conditions if c.get("field") != runtime_comparison_filter["field"]]
230 conditions.append(runtime_comparison_filter)
231 return {"operator": str(init_logical_filter["operator"]), "conditions": conditions}
233 logger.warning(
234 "The provided logical_operator, {logical_operator}, does not match the logical operator found in "
235 "the init logical filter, {filters_logical_operator}, so the init logical filter will be ignored.",
236 logical_operator=logical_operator,
237 filters_logical_operator=init_logical_filter["operator"],
238 )
239 return runtime_comparison_filter
242def combine_two_comparison_filters(
243 init_comparison_filter: dict[str, Any],
244 runtime_comparison_filter: dict[str, Any],
245 logical_operator: Literal["AND", "OR", "NOT"],
246) -> dict[str, Any]:
247 """
248 Combine a comparison filter with the `init_comparison_filter` using the provided `logical_operator`.
250 If `runtime_comparison_filter` and `init_comparison_filter` target the same field, `init_comparison_filter`
251 is ignored and `runtime_comparison_filter` is returned unchanged.
253 __Example__:
255 ```python
256 runtime_comparison_filter = {"field": "meta.type", "operator": "==", "value": "article"},
257 init_comparison_filter = {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
258 new_filters = combine_two_comparison_filters(
259 init_comparison_filter, runtime_comparison_filter, "AND"
260 )
261 # Output:
262 {
263 "operator": "AND",
264 "conditions": [
265 {"field": "meta.type", "operator": "==", "value": "article"},
266 {"field": "meta.date", "operator": ">=", "value": "2015-01-01"},
267 ]
268 }
269 ```
270 """
271 if runtime_comparison_filter["field"] == init_comparison_filter["field"]:
272 logger.warning(
273 "The parsed filter, {parsed_filter}, is ignored as the field is already present in the existing "
274 "filters, {filters}.",
275 parsed_filter=init_comparison_filter,
276 filters=runtime_comparison_filter,
277 )
278 return runtime_comparison_filter
280 return {"operator": str(logical_operator), "conditions": [init_comparison_filter, runtime_comparison_filter]}
283def apply_filter_policy(
284 filter_policy: FilterPolicy,
285 init_filters: dict[str, Any] | None = None,
286 runtime_filters: dict[str, Any] | None = None,
287 default_logical_operator: Literal["AND", "OR", "NOT"] = "AND",
288) -> dict[str, Any] | None:
289 """
290 Apply the filter policy to the given initial and runtime filters to determine the final set of filters used.
292 The function combines or replaces the initial and runtime filters based on the specified filter policy.
294 :param filter_policy: The policy to apply when handling the filters. It can be one of the following:
295 - `FilterPolicy.REPLACE`: Runtime filters will replace the initial filters.
296 - `FilterPolicy.MERGE`: Runtime filters will be merged with the initial filters. If there are overlapping keys,
297 values from the runtime filters will overwrite those from the initial filters.
298 :param init_filters: The initial filters set during the initialization of the relevant retriever.
299 :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
300 can change for each query/retriever run invocation.
301 :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
302 :returns: A dictionary containing the resulting filters based on the provided policy.
303 """
304 if filter_policy == FilterPolicy.MERGE and runtime_filters and init_filters:
305 # now we merge filters
306 if is_comparison_filter(init_filters) and is_comparison_filter(runtime_filters):
307 return combine_two_comparison_filters(init_filters, runtime_filters, default_logical_operator)
308 if is_comparison_filter(init_filters) and is_logical_filter(runtime_filters):
309 return combine_init_comparison_and_runtime_logical_filters(
310 init_filters, runtime_filters, default_logical_operator
311 )
312 if is_logical_filter(init_filters) and is_comparison_filter(runtime_filters):
313 return combine_runtime_comparison_and_init_logical_filters(
314 runtime_filters, init_filters, default_logical_operator
315 )
316 if is_logical_filter(init_filters) and is_logical_filter(runtime_filters):
317 return combine_two_logical_filters(init_filters, runtime_filters)
319 return runtime_filters or init_filters