fix(store): support filter operators and apply limit after filtering

langgraph/store/redis/__init__.py

@@ -33,13 +33,15 @@
 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,
     BaseRedisStore,
     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:

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:

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 = [

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."""

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