From 16b363fbb023d0e377c06a66e24584264e35478b Mon Sep 17 00:00:00 2001 From: Sam Crowder Date: Mon, 11 Aug 2025 09:19:34 -0700 Subject: [PATCH] feat(langgraph): implement redis node level cache (#5834) ### Description Adds Redis as a supported cache backend for LangGraph node-level caching, enabling distributed caching across multiple processes/servers. This implementation follows the same patterns as existing InMemoryCache and SqliteCache. ### Key changes - New RedisCache class implementing the BaseCache interface - Support for TTL-based expiration and batch operations - Worker-specific cache prefixes for parallel test isolation ### Dependencies - redis package (already included in dev dependencies) ### Test Plan - Unit tests: Added Redis cache tests covering basic operations, TTL, batch operations, and error handling - Integration tests: Redis cache integrated into existing LangGraph test suite, tested with all checkpointer combinations --- libs/checkpoint-postgres/uv.lock | 1 + libs/checkpoint-sqlite/uv.lock | 1 + .../langgraph/cache/redis/__init__.py | 144 ++++++++ libs/checkpoint/pyproject.toml | 1 + libs/checkpoint/tests/test_redis_cache.py | 313 ++++++++++++++++++ libs/checkpoint/uv.lock | 23 ++ libs/langgraph/Makefile | 20 +- libs/langgraph/pyproject.toml | 1 + libs/langgraph/tests/compose-redis.yml | 16 + libs/langgraph/tests/conftest.py | 26 +- libs/langgraph/uv.lock | 24 ++ libs/prebuilt/Makefile | 16 +- libs/prebuilt/tests/compose-redis.yml | 16 + libs/prebuilt/uv.lock | 2 + 14 files changed, 585 insertions(+), 19 deletions(-) create mode 100644 libs/checkpoint/langgraph/cache/redis/__init__.py create mode 100644 libs/checkpoint/tests/test_redis_cache.py create mode 100644 libs/langgraph/tests/compose-redis.yml create mode 100644 libs/prebuilt/tests/compose-redis.yml diff --git a/libs/checkpoint-postgres/uv.lock b/libs/checkpoint-postgres/uv.lock index 343c49a2d..efd85491e 100644 --- a/libs/checkpoint-postgres/uv.lock +++ b/libs/checkpoint-postgres/uv.lock @@ -329,6 +329,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ] diff --git a/libs/checkpoint-sqlite/uv.lock b/libs/checkpoint-sqlite/uv.lock index c99bed22b..c079777d0 100644 --- a/libs/checkpoint-sqlite/uv.lock +++ b/libs/checkpoint-sqlite/uv.lock @@ -341,6 +341,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ] diff --git a/libs/checkpoint/langgraph/cache/redis/__init__.py b/libs/checkpoint/langgraph/cache/redis/__init__.py new file mode 100644 index 000000000..ea4f71480 --- /dev/null +++ b/libs/checkpoint/langgraph/cache/redis/__init__.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +from langgraph.cache.base import BaseCache, FullKey, Namespace, ValueT +from langgraph.checkpoint.serde.base import SerializerProtocol + + +class RedisCache(BaseCache[ValueT]): + """Redis-based cache implementation with TTL support.""" + + def __init__( + self, + redis: Any, + *, + serde: SerializerProtocol | None = None, + prefix: str = "langgraph:cache:", + ) -> None: + """Initialize the cache with a Redis client. + + Args: + redis: Redis client instance (sync or async) + serde: Serializer to use for values + prefix: Key prefix for all cached values + """ + super().__init__(serde=serde) + self.redis = redis + self.prefix = prefix + + def _make_key(self, ns: Namespace, key: str) -> str: + """Create a Redis key from namespace and key.""" + ns_str = ":".join(ns) if ns else "" + return f"{self.prefix}{ns_str}:{key}" if ns_str else f"{self.prefix}{key}" + + def _parse_key(self, redis_key: str) -> tuple[Namespace, str]: + """Parse a Redis key back to namespace and key.""" + if not redis_key.startswith(self.prefix): + raise ValueError( + f"Key {redis_key} does not start with prefix {self.prefix}" + ) + + remaining = redis_key[len(self.prefix) :] + if ":" in remaining: + parts = remaining.split(":") + key = parts[-1] + ns_parts = parts[:-1] + return (tuple(ns_parts), key) + else: + return (tuple(), remaining) + + def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: + """Get the cached values for the given keys.""" + if not keys: + return {} + + # Build Redis keys + redis_keys = [self._make_key(ns, key) for ns, key in keys] + + # Get values from Redis using MGET + try: + raw_values = self.redis.mget(redis_keys) + except Exception: + # If Redis is unavailable, return empty dict + return {} + + values: dict[FullKey, ValueT] = {} + for i, raw_value in enumerate(raw_values): + if raw_value is not None: + try: + # Deserialize the value + encoding, data = raw_value.split(b":", 1) + values[keys[i]] = self.serde.loads_typed((encoding.decode(), data)) + except Exception: + # Skip corrupted entries + continue + + return values + + async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: + """Asynchronously get the cached values for the given keys.""" + return self.get(keys) + + def set(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: + """Set the cached values for the given keys and TTLs.""" + if not mapping: + return + + # Use pipeline for efficient batch operations + pipe = self.redis.pipeline() + + for (ns, key), (value, ttl) in mapping.items(): + redis_key = self._make_key(ns, key) + encoding, data = self.serde.dumps_typed(value) + + # Store as "encoding:data" format + serialized_value = f"{encoding}:".encode() + data + + if ttl is not None: + pipe.setex(redis_key, ttl, serialized_value) + else: + pipe.set(redis_key, serialized_value) + + try: + pipe.execute() + except Exception: + # Silently fail if Redis is unavailable + pass + + async def aset(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: + """Asynchronously set the cached values for the given keys and TTLs.""" + self.set(mapping) + + def clear(self, namespaces: Sequence[Namespace] | None = None) -> None: + """Delete the cached values for the given namespaces. + If no namespaces are provided, clear all cached values.""" + try: + if namespaces is None: + # Clear all keys with our prefix + pattern = f"{self.prefix}*" + keys = self.redis.keys(pattern) + if keys: + self.redis.delete(*keys) + else: + # Clear specific namespaces + keys_to_delete = [] + for ns in namespaces: + ns_str = ":".join(ns) if ns else "" + pattern = ( + f"{self.prefix}{ns_str}:*" if ns_str else f"{self.prefix}*" + ) + keys = self.redis.keys(pattern) + keys_to_delete.extend(keys) + + if keys_to_delete: + self.redis.delete(*keys_to_delete) + except Exception: + # Silently fail if Redis is unavailable + pass + + async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None: + """Asynchronously delete the cached values for the given namespaces. + If no namespaces are provided, clear all cached values.""" + self.clear(namespaces) diff --git a/libs/checkpoint/pyproject.toml b/libs/checkpoint/pyproject.toml index 647e757e0..6df6ab54b 100644 --- a/libs/checkpoint/pyproject.toml +++ b/libs/checkpoint/pyproject.toml @@ -32,6 +32,7 @@ dev = [ "numpy", "pandas", "pandas-stubs>=2.2.2.240807", + "redis", ] [tool.hatch.build.targets.wheel] diff --git a/libs/checkpoint/tests/test_redis_cache.py b/libs/checkpoint/tests/test_redis_cache.py new file mode 100644 index 000000000..0eac2072d --- /dev/null +++ b/libs/checkpoint/tests/test_redis_cache.py @@ -0,0 +1,313 @@ +"""Unit tests for Redis cache implementation.""" + +import time + +import pytest +import redis + +from langgraph.cache.redis import RedisCache + + +class TestRedisCache: + @pytest.fixture(autouse=True) + def setup(self): + """Set up test Redis client and cache.""" + self.client = redis.Redis( + host="localhost", port=6379, db=0, decode_responses=False + ) + try: + self.client.ping() + except redis.ConnectionError: + pytest.skip("Redis server not available") + + self.cache = RedisCache(self.client, prefix="test:cache:") + + # Clean up before each test + self.client.flushdb() + + def teardown_method(self): + """Clean up after each test.""" + try: + self.client.flushdb() + except Exception: + pass + + def test_basic_set_and_get(self): + """Test basic set and get operations.""" + keys = [(("graph", "node"), "key1")] + values = {keys[0]: ({"result": 42}, None)} + + # Set value + self.cache.set(values) + + # Get value + result = self.cache.get(keys) + assert len(result) == 1 + assert result[keys[0]] == {"result": 42} + + def test_batch_operations(self): + """Test batch set and get operations.""" + keys = [ + (("graph", "node1"), "key1"), + (("graph", "node2"), "key2"), + (("other", "node"), "key3"), + ] + values = { + keys[0]: ({"result": 1}, None), + keys[1]: ({"result": 2}, 60), # With TTL + keys[2]: ({"result": 3}, None), + } + + # Set values + self.cache.set(values) + + # Get all values + result = self.cache.get(keys) + assert len(result) == 3 + assert result[keys[0]] == {"result": 1} + assert result[keys[1]] == {"result": 2} + assert result[keys[2]] == {"result": 3} + + def test_ttl_behavior(self): + """Test TTL (time-to-live) functionality.""" + key = (("graph", "node"), "ttl_key") + values = {key: ({"data": "expires_soon"}, 1)} # 1 second TTL + + # Set with TTL + self.cache.set(values) + + # Should be available immediately + result = self.cache.get([key]) + assert len(result) == 1 + assert result[key] == {"data": "expires_soon"} + + # Wait for expiration + time.sleep(1.1) + + # Should be expired + result = self.cache.get([key]) + assert len(result) == 0 + + def test_namespace_isolation(self): + """Test that different namespaces are isolated.""" + key1 = (("graph1", "node"), "same_key") + key2 = (("graph2", "node"), "same_key") + + values = {key1: ({"graph": 1}, None), key2: ({"graph": 2}, None)} + + self.cache.set(values) + + result = self.cache.get([key1, key2]) + assert result[key1] == {"graph": 1} + assert result[key2] == {"graph": 2} + + def test_clear_all(self): + """Test clearing all cached values.""" + keys = [(("graph", "node1"), "key1"), (("graph", "node2"), "key2")] + values = {keys[0]: ({"result": 1}, None), keys[1]: ({"result": 2}, None)} + + self.cache.set(values) + + # Verify data exists + result = self.cache.get(keys) + assert len(result) == 2 + + # Clear all + self.cache.clear() + + # Verify data is gone + result = self.cache.get(keys) + assert len(result) == 0 + + def test_clear_by_namespace(self): + """Test clearing cached values by namespace.""" + keys = [ + (("graph1", "node"), "key1"), + (("graph2", "node"), "key2"), + (("graph1", "other"), "key3"), + ] + values = { + keys[0]: ({"result": 1}, None), + keys[1]: ({"result": 2}, None), + keys[2]: ({"result": 3}, None), + } + + self.cache.set(values) + + # Clear only graph1 namespace + self.cache.clear([("graph1", "node"), ("graph1", "other")]) + + # graph1 should be cleared, graph2 should remain + result = self.cache.get(keys) + assert len(result) == 1 + assert result[keys[1]] == {"result": 2} + + def test_empty_operations(self): + """Test behavior with empty keys/values.""" + # Empty get + result = self.cache.get([]) + assert result == {} + + # Empty set + self.cache.set({}) # Should not raise error + + def test_nonexistent_keys(self): + """Test getting keys that don't exist.""" + keys = [(("graph", "node"), "nonexistent")] + result = self.cache.get(keys) + assert len(result) == 0 + + @pytest.mark.asyncio + async def test_async_operations(self): + """Test async set and get operations with sync Redis client.""" + # Create sync Redis client and cache (like main integration tests) + client = redis.Redis( + host="localhost", port=6379, db=1, decode_responses=False + ) + try: + client.ping() + except Exception: + pytest.skip("Redis not available") + + cache = RedisCache(client, prefix="test:async:") + + keys = [(("graph", "node"), "async_key")] + values = {keys[0]: ({"async": True}, None)} + + # Async set (delegates to sync) + await cache.aset(values) + + # Async get (delegates to sync) + result = await cache.aget(keys) + assert len(result) == 1 + assert result[keys[0]] == {"async": True} + + # Cleanup + client.flushdb() + + @pytest.mark.asyncio + async def test_async_clear(self): + """Test async clear operations with sync Redis client.""" + # Create sync Redis client and cache (like main integration tests) + client = redis.Redis( + host="localhost", port=6379, db=1, decode_responses=False + ) + try: + client.ping() + except Exception: + pytest.skip("Redis not available") + + cache = RedisCache(client, prefix="test:async:") + + keys = [(("graph", "node"), "key")] + values = {keys[0]: ({"data": "test"}, None)} + + await cache.aset(values) + + # Verify data exists + result = await cache.aget(keys) + assert len(result) == 1 + + # Clear all (delegates to sync) + await cache.aclear() + + # Verify data is gone + result = await cache.aget(keys) + assert len(result) == 0 + + # Cleanup + client.flushdb() + + def test_redis_unavailable_get(self): + """Test behavior when Redis is unavailable during get operations.""" + # Create cache with non-existent Redis server + bad_client = redis.Redis( + host="nonexistent", port=9999, socket_connect_timeout=0.1 + ) + cache = RedisCache(bad_client, prefix="test:cache:") + + keys = [(("graph", "node"), "key")] + result = cache.get(keys) + + # Should return empty dict when Redis unavailable + assert result == {} + + def test_redis_unavailable_set(self): + """Test behavior when Redis is unavailable during set operations.""" + # Create cache with non-existent Redis server + bad_client = redis.Redis( + host="nonexistent", port=9999, socket_connect_timeout=0.1 + ) + cache = RedisCache(bad_client, prefix="test:cache:") + + keys = [(("graph", "node"), "key")] + values = {keys[0]: ({"data": "test"}, None)} + + # Should not raise exception when Redis unavailable + cache.set(values) # Should silently fail + + @pytest.mark.asyncio + async def test_redis_unavailable_async(self): + """Test async behavior when Redis is unavailable.""" + # Create sync cache with non-existent Redis server (like main integration tests) + bad_client = redis.Redis( + host="nonexistent", port=9999, socket_connect_timeout=0.1 + ) + cache = RedisCache(bad_client, prefix="test:cache:") + + keys = [(("graph", "node"), "key")] + values = {keys[0]: ({"data": "test"}, None)} + + # Should return empty dict for get (delegates to sync) + result = await cache.aget(keys) + assert result == {} + + # Should not raise exception for set (delegates to sync) + await cache.aset(values) # Should silently fail + + def test_corrupted_data_handling(self): + """Test handling of corrupted data in Redis.""" + # Set some valid data first + keys = [(("graph", "node"), "valid_key")] + values = {keys[0]: ({"data": "valid"}, None)} + self.cache.set(values) + + # Manually insert corrupted data + corrupted_key = self.cache._make_key(("graph", "node"), "corrupted_key") + self.client.set(corrupted_key, b"invalid:data:format:too:many:colons") + + # Should skip corrupted entry and return only valid ones + all_keys = [keys[0], (("graph", "node"), "corrupted_key")] + result = self.cache.get(all_keys) + + assert len(result) == 1 + assert result[keys[0]] == {"data": "valid"} + + def test_key_parsing_edge_cases(self): + """Test key parsing with edge cases.""" + # Test empty namespace + key1 = ((), "empty_ns") + values = {key1: ({"data": "empty_ns"}, None)} + self.cache.set(values) + result = self.cache.get([key1]) + assert result[key1] == {"data": "empty_ns"} + + # Test namespace with special characters + key2 = (("graph:with:colons", "node-with-dashes"), "key_with_underscores") + values = {key2: ({"data": "special_chars"}, None)} + self.cache.set(values) + result = self.cache.get([key2]) + assert result[key2] == {"data": "special_chars"} + + def test_large_data_serialization(self): + """Test handling of large data objects.""" + # Create a large data structure + large_data = {"large_list": list(range(1000)), "nested": {"data": "x" * 1000}} + key = (("graph", "node"), "large_key") + values = {key: (large_data, None)} + + self.cache.set(values) + result = self.cache.get([key]) + + assert len(result) == 1 + assert result[key] == large_data diff --git a/libs/checkpoint/uv.lock b/libs/checkpoint/uv.lock index 4a39de103..1454bd275 100644 --- a/libs/checkpoint/uv.lock +++ b/libs/checkpoint/uv.lock @@ -32,6 +32,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a1/ee/48ca1a7c89ffec8b6a0c5d02b89c305671d5ffd8d3c94acf8b8c408575bb/anyio-4.9.0-py3-none-any.whl", hash = "sha256:9f76d541cad6e36af7beb62e978876f3b41e3e04f2c1fbf0884604c0a9c4d93c", size = 100916, upload-time = "2025-03-17T00:02:52.713Z" }, ] +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + [[package]] name = "certifi" version = "2025.7.9" @@ -345,6 +354,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ] @@ -366,6 +376,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ] @@ -1153,6 +1164,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/19/87/5124b1c1f2412bb95c59ec481eaf936cd32f0fe2a7b16b97b81c4c017a6a/PyYAML-6.0.2-cp39-cp39-win_amd64.whl", hash = "sha256:39693e1f8320ae4f43943590b49779ffb98acb81f788220ea932a6b6c51004d8", size = 162312, upload-time = "2024-08-06T20:33:49.073Z" }, ] +[[package]] +name = "redis" +version = "6.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout", marker = "python_full_version < '3.11.3'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/21/cd/030274634a1a052b708756016283ea3d84e91ae45f74d7f5dcf55d753a0f/redis-6.3.0.tar.gz", hash = "sha256:3000dbe532babfb0999cdab7b3e5744bcb23e51923febcfaeb52c8cfb29632ef", size = 4647275, upload-time = "2025-08-05T08:12:31.648Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/a7/2fe45801534a187543fc45d28b3844d84559c1589255bc2ece30d92dc205/redis-6.3.0-py3-none-any.whl", hash = "sha256:92f079d656ded871535e099080f70fab8e75273c0236797126ac60242d638e9b", size = 280018, upload-time = "2025-08-05T08:12:30.093Z" }, +] + [[package]] name = "requests" version = "2.32.4" diff --git a/libs/langgraph/Makefile b/libs/langgraph/Makefile index fd4574a39..e6d56c3f9 100644 --- a/libs/langgraph/Makefile +++ b/libs/langgraph/Makefile @@ -37,11 +37,11 @@ coverage: --cov-report xml \ --cov-report term-missing:skip-covered -start-postgres: - docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait --remove-orphans +start-services: + docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml up -V --force-recreate --wait --remove-orphans -stop-postgres: - docker compose -f tests/compose-postgres.yml down -v +stop-services: + docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml down -v start-dev-server: LOG_LEVEL=warning uv run langgraph dev --config tests/example_app/langgraph.json --no-browser & echo "$$!" > .devserver.pid @@ -60,11 +60,11 @@ NO_DOCKER ?= $(sh command -v docker >/dev/null 2>&1 && echo "false" || echo "tru test: if [ "$(NO_DOCKER)" = "false" ]; then \ - make start-postgres &&\ + make start-services &&\ make start-dev-server &&\ uv run pytest $(TEST); \ EXIT_CODE=$$?; \ - make stop-postgres; \ + make stop-services; \ make stop-dev-server; \ exit $$EXIT_CODE; \ else \ @@ -74,11 +74,11 @@ test: fi test_parallel: - make start-postgres &&\ + make start-services &&\ make start-dev-server &&\ uv run pytest -n auto --dist worksteal $(TEST); \ EXIT_CODE=$$?; \ - make stop-postgres; \ + make stop-services; \ make stop-dev-server; \ exit $$EXIT_CODE @@ -93,11 +93,11 @@ MAXFAIL_ARGS := $(if $(MAXFAIL),--maxfail $(MAXFAIL),) XDIST_ARGS := $(if $(WORKERS),-x $(XDIST_ARGS),) test_watch: - make start-postgres &&\ + make start-services &&\ make start-dev-server &&\ uv run ptw . -- --ff -vv $(XDIST_ARGS) $(MAXFAIL_ARGS) $(TEST); \ EXIT_CODE=$$?; \ - make stop-postgres; \ + make stop-services; \ make stop-dev-server; \ exit $$EXIT_CODE diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 04905075b..13c1c277c 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -49,6 +49,7 @@ dev = [ "types-requests", "pycryptodome", "langgraph-cli[inmem]", + "redis", ] [tool.uv] diff --git a/libs/langgraph/tests/compose-redis.yml b/libs/langgraph/tests/compose-redis.yml new file mode 100644 index 000000000..be4343709 --- /dev/null +++ b/libs/langgraph/tests/compose-redis.yml @@ -0,0 +1,16 @@ +name: langgraph-tests +services: + redis-test: + image: redis:7-alpine + ports: + - "6379:6379" + command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru + healthcheck: + test: redis-cli ping + start_period: 10s + timeout: 1s + retries: 5 + interval: 5s + start_interval: 1s + tmpfs: + - /data # Use tmpfs for faster testing diff --git a/libs/langgraph/tests/conftest.py b/libs/langgraph/tests/conftest.py index d82239aa5..4ef30b918 100644 --- a/libs/langgraph/tests/conftest.py +++ b/libs/langgraph/tests/conftest.py @@ -3,10 +3,12 @@ from collections.abc import AsyncIterator, Iterator from uuid import UUID import pytest +import redis from pytest_mock import MockerFixture from langgraph.cache.base import BaseCache from langgraph.cache.memory import InMemoryCache +from langgraph.cache.redis import RedisCache from langgraph.cache.sqlite import SqliteCache from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.store.base import BaseStore @@ -55,12 +57,34 @@ def durability(request: pytest.FixtureRequest) -> Durability: return request.param -@pytest.fixture(scope="function", params=["sqlite", "memory"]) +@pytest.fixture( + scope="function", + params=["sqlite", "memory"] if NO_DOCKER else ["sqlite", "memory", "redis"], +) def cache(request: pytest.FixtureRequest) -> Iterator[BaseCache]: if request.param == "sqlite": yield SqliteCache(path=":memory:") elif request.param == "memory": yield InMemoryCache() + elif request.param == "redis": + # Get worker ID for parallel test isolation + worker_id = getattr(request.config, "workerinput", {}).get("workerid", "master") + + redis_client = redis.Redis( + host="localhost", port=6379, db=0, decode_responses=False + ) + # Use worker-specific prefix to avoid cache pollution between parallel tests + cache = RedisCache(redis_client, prefix=f"test:cache:{worker_id}:") + yield cache + + try: + # Only clear keys with our specific prefix + pattern = f"test:cache:{worker_id}:*" + keys = redis_client.keys(pattern) + if keys: + redis_client.delete(*keys) + except Exception: + pass else: raise ValueError(f"Unknown cache type: {request.param}") diff --git a/libs/langgraph/uv.lock b/libs/langgraph/uv.lock index 0a4052e3b..2d00caf41 100644 --- a/libs/langgraph/uv.lock +++ b/libs/langgraph/uv.lock @@ -119,6 +119,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/03/49/d10027df9fce941cb8184e78a02857af36360d33e1721df81c5ed2179a1a/async_lru-2.0.5-py3-none-any.whl", hash = "sha256:ab95404d8d2605310d345932697371a5f40def0487c03d6d0ad9138de52c9943", size = 6069, upload-time = "2025-03-16T17:25:35.422Z" }, ] +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + [[package]] name = "attrs" version = "25.3.0" @@ -1225,6 +1234,7 @@ dev = [ { name = "pytest-repeat" }, { name = "pytest-watcher" }, { name = "pytest-xdist", extra = ["psutil"] }, + { name = "redis" }, { name = "ruff" }, { name = "syrupy" }, { name = "types-requests" }, @@ -1263,6 +1273,7 @@ dev = [ { name = "pytest-repeat" }, { name = "pytest-watcher" }, { name = "pytest-xdist", extras = ["psutil"] }, + { name = "redis" }, { name = "ruff" }, { name = "syrupy" }, { name = "types-requests" }, @@ -1326,6 +1337,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ] @@ -2628,6 +2640,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/51/8b/619a9ee2fa4d3c724fbadde946427735ade64da03894b071bbdc3b789d83/pyzmq-27.0.0-pp39-pypy39_pp73-win_amd64.whl", hash = "sha256:096af9e133fec3a72108ddefba1e42985cb3639e9de52cfd336b6fc23aa083e9", size = 544715, upload-time = "2025-06-13T14:09:05.579Z" }, ] +[[package]] +name = "redis" +version = "6.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout", marker = "python_full_version < '3.11.3'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/21/cd/030274634a1a052b708756016283ea3d84e91ae45f74d7f5dcf55d753a0f/redis-6.3.0.tar.gz", hash = "sha256:3000dbe532babfb0999cdab7b3e5744bcb23e51923febcfaeb52c8cfb29632ef", size = 4647275, upload-time = "2025-08-05T08:12:31.648Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/a7/2fe45801534a187543fc45d28b3844d84559c1589255bc2ece30d92dc205/redis-6.3.0-py3-none-any.whl", hash = "sha256:92f079d656ded871535e099080f70fab8e75273c0236797126ac60242d638e9b", size = 280018, upload-time = "2025-08-05T08:12:30.093Z" }, +] + [[package]] name = "referencing" version = "0.36.2" diff --git a/libs/prebuilt/Makefile b/libs/prebuilt/Makefile index aef3028e5..d8c0abc58 100644 --- a/libs/prebuilt/Makefile +++ b/libs/prebuilt/Makefile @@ -7,11 +7,11 @@ all: help # TESTING AND COVERAGE ###################### -start-postgres: - docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait --remove-orphans +start-services: + docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml up -V --force-recreate --wait --remove-orphans -stop-postgres: - docker compose -f tests/compose-postgres.yml down -v +stop-services: + docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml down -v TEST ?= . @@ -19,15 +19,15 @@ test-fast: LANGGRAPH_TEST_FAST=1 uv run pytest $(TEST) test: - make start-postgres && LANGGRAPH_TEST_FAST=0 uv run pytest $(TEST); \ + make start-services && LANGGRAPH_TEST_FAST=0 uv run pytest $(TEST); \ EXIT_CODE=$$?; \ - make stop-postgres; \ + make stop-services; \ exit $$EXIT_CODE test_watch: - make start-postgres && LANGGRAPH_TEST_FAST=0 uv run ptw $(TEST); \ + make start-services && LANGGRAPH_TEST_FAST=0 uv run ptw $(TEST); \ EXIT_CODE=$$?; \ - make stop-postgres; \ + make stop-services; \ exit $$EXIT_CODE ###################### diff --git a/libs/prebuilt/tests/compose-redis.yml b/libs/prebuilt/tests/compose-redis.yml new file mode 100644 index 000000000..18862fd28 --- /dev/null +++ b/libs/prebuilt/tests/compose-redis.yml @@ -0,0 +1,16 @@ +name: langgraph-tests-redis +services: + redis-test: + image: redis:7-alpine + ports: + - "6379:6379" + command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru + healthcheck: + test: redis-cli ping + start_period: 10s + timeout: 1s + retries: 5 + interval: 5s + start_interval: 1s + tmpfs: + - /data # Use tmpfs for faster testing diff --git a/libs/prebuilt/uv.lock b/libs/prebuilt/uv.lock index 19fcbd6c8..3bd2217bf 100644 --- a/libs/prebuilt/uv.lock +++ b/libs/prebuilt/uv.lock @@ -359,6 +359,7 @@ dev = [ { name = "pytest-repeat" }, { name = "pytest-watcher" }, { name = "pytest-xdist", extras = ["psutil"] }, + { name = "redis" }, { name = "ruff" }, { name = "syrupy" }, { name = "types-requests" }, @@ -392,6 +393,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-watcher" }, + { name = "redis" }, { name = "ruff" }, ]