diff --git a/langgraph/store/redis/__init__.py b/langgraph/store/redis/__init__.py index a9172e0..9a57c47 100644 --- a/langgraph/store/redis/__init__.py +++ b/langgraph/store/redis/__init__.py @@ -33,6 +33,7 @@ from langgraph.checkpoint.redis.util import unwrap_connection_errors from langgraph.store.redis.aio import AsyncRedisStore from langgraph.store.redis.base import ( + FILTER_PAGE_SIZE, REDIS_KEY_SEPARATOR, STORE_PREFIX, STORE_VECTOR_PREFIX, @@ -40,6 +41,7 @@ RedisDocument, _decode_ns, _group_ops, + _matches_filter, _namespace_to_text, _row_to_item, _row_to_search_item, @@ -455,6 +457,29 @@ def _batch_put_ops( for main_key, (related_keys, ttl_minutes) in ttl_tracking.items(): self._apply_ttl_to_keys(main_key, related_keys, ttl_minutes) + def _search_filtered( + self, query_str: str, filter: dict[str, Any], offset: int, limit: int + ) -> list[Any]: + """Page through the documents matching query_str and return those that pass filter. + + Value fields aren't part of the search index, so the filter runs client-side and + offset and limit have to apply to the filtered documents, not to a page of Redis + results. + """ + matched: list[Any] = [] + start = 0 + while len(matched) < offset + limit: + res = self.store_index.search( + Query(query_str).paging(start, FILTER_PAGE_SIZE) + ) + for doc in res.docs: + if _matches_filter(json.loads(doc.json).get("value", {}), filter): + matched.append(doc) + start += len(res.docs) + if not res.docs or start >= res.total: + break + return matched[offset : offset + limit] + def _batch_search_ops( self, search_ops: list[tuple[int, SearchOp]], @@ -565,20 +590,10 @@ def _batch_search_ops( continue # Apply value filters if needed - if op.filter: - matches = True - value = store_doc.get("value", {}) - for key, expected in op.filter.items(): - actual = value.get(key) - if isinstance(expected, list): - if actual not in expected: - matches = False - break - elif actual != expected: - matches = False - break - if not matches: - continue + if op.filter and not _matches_filter( + store_doc.get("value", {}), op.filter + ): + continue # If refresh_ttl is true, add to list for refreshing if op.refresh_ttl: @@ -626,31 +641,19 @@ def _batch_search_ops( results[idx] = items else: # Regular search - # Create a query with LIMIT and OFFSET parameters - query = Query(query_str).paging(offset, limit) - - # Execute search with limit and offset applied by Redis - res = self.store_index.search(query) + if op.filter: + docs = self._search_filtered(query_str, op.filter, offset, limit) + else: + # Without a filter, Redis applies offset and limit directly + res = self.store_index.search( + Query(query_str).paging(offset, limit) + ) + docs = res.docs items = [] refresh_keys = [] # Track keys that need TTL refreshed - for doc in res.docs: + for doc in docs: data = json.loads(doc.json) - # Apply value filters - if op.filter: - matches = True - value = data.get("value", {}) - for key, expected in op.filter.items(): - actual = value.get(key) - if isinstance(expected, list): - if actual not in expected: - matches = False - break - elif actual != expected: - matches = False - break - if not matches: - continue # If refresh_ttl is true, add the key to refresh list if op.refresh_ttl: diff --git a/langgraph/store/redis/aio.py b/langgraph/store/redis/aio.py index 75adcdc..51160ec 100644 --- a/langgraph/store/redis/aio.py +++ b/langgraph/store/redis/aio.py @@ -31,6 +31,7 @@ from langgraph.checkpoint.redis.util import unwrap_connection_errors from langgraph.store.redis.base import ( + FILTER_PAGE_SIZE, REDIS_KEY_SEPARATOR, STORE_PREFIX, STORE_VECTOR_PREFIX, @@ -39,6 +40,7 @@ _decode_ns, _ensure_string_or_literal, _group_ops, + _matches_filter, _namespace_to_text, _row_to_item, _row_to_search_item, @@ -675,6 +677,29 @@ async def _batch_put_ops( for main_key, (related_keys, ttl_minutes) in ttl_tracking.items(): await self._apply_ttl_to_keys(main_key, related_keys, ttl_minutes) + async def _asearch_filtered( + self, query_str: str, filter: dict[str, Any], offset: int, limit: int + ) -> list[Any]: + """Page through the documents matching query_str and return those that pass filter. + + Value fields aren't part of the search index, so the filter runs client-side and + offset and limit have to apply to the filtered documents, not to a page of Redis + results. + """ + matched: list[Any] = [] + start = 0 + while len(matched) < offset + limit: + res = await self.store_index.search( + Query(query_str).paging(start, FILTER_PAGE_SIZE) + ) + for doc in res.docs: + if _matches_filter(json.loads(doc.json).get("value", {}), filter): + matched.append(doc) + start += len(res.docs) + if not res.docs or start >= res.total: + break + return matched[offset : offset + limit] + async def _batch_search_ops( self, search_ops: Sequence[tuple[int, SearchOp]], @@ -787,20 +812,10 @@ async def _batch_search_ops( continue # Apply value filters if needed - if op.filter: - matches = True - value = store_doc.get("value", {}) - for key, expected in op.filter.items(): - actual = value.get(key) - if isinstance(expected, list): - if actual not in expected: - matches = False - break - elif actual != expected: - matches = False - break - if not matches: - continue + if op.filter and not _matches_filter( + store_doc.get("value", {}), op.filter + ): + continue # If refresh_ttl is true, add to list for refreshing if op.refresh_ttl: @@ -849,31 +864,21 @@ async def _batch_search_ops( else: # Regular search - # Create a query with LIMIT and OFFSET parameters - query = Query(query_str).paging(offset, limit) - - # Execute search with limit and offset applied by Redis - res = await self.store_index.search(query) + if op.filter: + docs = await self._asearch_filtered( + query_str, op.filter, offset, limit + ) + else: + # Without a filter, Redis applies offset and limit directly + res = await self.store_index.search( + Query(query_str).paging(offset, limit) + ) + docs = res.docs items = [] refresh_keys = [] # Track keys that need TTL refreshed - for doc in res.docs: + for doc in docs: data = json.loads(doc.json) - # Apply value filters - if op.filter: - matches = True - value = data.get("value", {}) - for key, expected in op.filter.items(): - actual = value.get(key) - if isinstance(expected, list): - if actual not in expected: - matches = False - break - elif actual != expected: - matches = False - break - if not matches: - continue # If refresh_ttl is true, add the key to refresh list if op.refresh_ttl: diff --git a/langgraph/store/redis/base.py b/langgraph/store/redis/base.py index 76c4f83..93b53e3 100644 --- a/langgraph/store/redis/base.py +++ b/langgraph/store/redis/base.py @@ -55,6 +55,59 @@ REDIS_KEY_SEPARATOR = ":" STORE_PREFIX = "store" STORE_VECTOR_PREFIX = "store_vectors" +# Number of documents fetched per page when a search filter is applied client-side +FILTER_PAGE_SIZE = 500 + + +def _matches_filter(value: dict[str, Any], filter: dict[str, Any]) -> bool: + """Return whether a stored value matches a search filter. + + Follows the semantics of LangGraph's InMemoryStore: comparison operators + ($eq, $ne, $gt, $gte, $lt, $lte) and nested dicts matched key by key. A + list matches any of its elements, as this store has always done. + """ + return all( + _filter_value_matches(value.get(key), expected) + for key, expected in filter.items() + ) + + +def _filter_value_matches(actual: Any, expected: Any) -> bool: + if isinstance(expected, dict): + if any(key.startswith("$") for key in expected): + return all( + _apply_filter_operator(actual, operator, operand) + for operator, operand in expected.items() + ) + return isinstance(actual, dict) and all( + _filter_value_matches(actual.get(key), value) + for key, value in expected.items() + ) + if isinstance(expected, list): + return actual in expected + return bool(actual == expected) + + +def _apply_filter_operator(actual: Any, operator: str, operand: Any) -> bool: + if operator == "$eq": + return bool(actual == operand) + if operator == "$ne": + return bool(actual != operand) + if operator in ("$gt", "$gte", "$lt", "$lte"): + try: + left, right = float(actual), float(operand) + except (TypeError, ValueError): + # A missing or non-numeric value never satisfies a range comparison + return False + if operator == "$gt": + return left > right + if operator == "$gte": + return left >= right + if operator == "$lt": + return left < right + return left <= right + raise ValueError(f"Unsupported operator: {operator}") + # Schemas for Redis Search indices SCHEMAS = [ diff --git a/tests/test_async_store.py b/tests/test_async_store.py index 26e52cf..19481cd 100644 --- a/tests/test_async_store.py +++ b/tests/test_async_store.py @@ -107,6 +107,47 @@ async def test_basic_ops(store: AsyncRedisStore) -> None: assert deleted_item is None +@pytest.mark.asyncio +async def test_search_filter_with_comparison_operators(store: AsyncRedisStore) -> None: + """Operator filters match like InMemoryStore instead of silently returning nothing.""" + namespace = ("ops", str(uuid4())) + for i in range(6): + await store.aput( + namespace, f"item{i}", {"score": i, "status": "done" if i % 2 else "open"} + ) + + async def keys(filter: dict) -> list[str]: + return sorted( + item.key for item in await store.asearch(namespace, filter=filter) + ) + + assert await keys({"score": {"$gt": 3}}) == ["item4", "item5"] + assert await keys({"score": {"$gte": 4}}) == ["item4", "item5"] + assert await keys({"score": {"$lt": 2}}) == ["item0", "item1"] + assert await keys({"score": {"$gt": 0, "$lte": 1}}) == ["item1"] + assert await keys({"status": {"$eq": "done"}}) == ["item1", "item3", "item5"] + assert await keys({"status": {"$ne": "done"}}) == ["item0", "item2", "item4"] + + +@pytest.mark.asyncio +async def test_search_filter_applies_limit_and_offset_to_matching_items( + store: AsyncRedisStore, +) -> None: + """limit and offset count the items that pass the filter, not the items Redis returned.""" + namespace = ("paged", str(uuid4())) + for i in range(30): + await store.aput(namespace, f"other{i:02d}", {"keep": False}) + for i in range(3): + await store.aput(namespace, f"match{i}", {"keep": True}) + + first = await store.asearch(namespace, filter={"keep": True}, limit=2) + rest = await store.asearch(namespace, filter={"keep": True}, limit=2, offset=2) + + assert len(first) == 2 + assert len(rest) == 1 + assert {item.key for item in first + rest} == {"match0", "match1", "match2"} + + @pytest.mark.asyncio async def test_search(store: AsyncRedisStore) -> None: """Test search functionality with async store.""" diff --git a/tests/test_store.py b/tests/test_store.py index baa5263..3d12e0a 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -372,6 +372,43 @@ def test_basic_ops(store: RedisStore) -> None: assert deleted_item is None +def test_search_filter_with_comparison_operators(store: RedisStore) -> None: + """Operator filters match like InMemoryStore instead of silently returning nothing.""" + namespace = ("ops", str(uuid4())) + for i in range(6): + store.put( + namespace, f"item{i}", {"score": i, "status": "done" if i % 2 else "open"} + ) + + def keys(filter: dict) -> list[str]: + return sorted(item.key for item in store.search(namespace, filter=filter)) + + assert keys({"score": {"$gt": 3}}) == ["item4", "item5"] + assert keys({"score": {"$gte": 4}}) == ["item4", "item5"] + assert keys({"score": {"$lt": 2}}) == ["item0", "item1"] + assert keys({"score": {"$gt": 0, "$lte": 1}}) == ["item1"] + assert keys({"status": {"$eq": "done"}}) == ["item1", "item3", "item5"] + assert keys({"status": {"$ne": "done"}}) == ["item0", "item2", "item4"] + + +def test_search_filter_applies_limit_and_offset_to_matching_items( + store: RedisStore, +) -> None: + """limit and offset count the items that pass the filter, not the items Redis returned.""" + namespace = ("paged", str(uuid4())) + for i in range(30): + store.put(namespace, f"other{i:02d}", {"keep": False}) + for i in range(3): + store.put(namespace, f"match{i}", {"keep": True}) + + first = store.search(namespace, filter={"keep": True}, limit=2) + rest = store.search(namespace, filter={"keep": True}, limit=2, offset=2) + + assert len(first) == 2 + assert len(rest) == 1 + assert {item.key for item in first + rest} == {"match0", "match1", "match2"} + + def test_search(store: RedisStore) -> None: """Test search functionality.""" # Create test data