feat: Support Redis Cluster (#35)

Add support for Redis Cluster in the Store implementations:

Detect when the client is connected to a Redis Cluster or Enterprise node
If using Cluster, use single-key commands instead of multi-key commands, including transactions.
Allow users to override cluster detection. This is useful with something like the Redis Enterprise proxy, which can be used from a non-cluster-aware client.
This is "Part One" because our lower-level dependency, RedisVL, also needs to be updated to support cluster-aware clients. However, with the changes in this PR, users who use Redis clusters through a proxy (with a non-cluster-aware client, e.g., Redis, not RedisCluster) can set cluster_mode=True so we change our multi-key command behavior.

Author: abrookins

langgraph/checkpoint/redis/aio.py

@@ -131,8 +131,10 @@ async def __aexit__(
     ) -> None:
         """Async context manager exit."""
         if self._owns_its_client:
-            await self._redis.aclose()  # type: ignore[attr-defined]
-            await self._redis.connection_pool.disconnect()
+            await self._redis.aclose()
+            coro = self._redis.connection_pool.disconnect()
+            if coro:
+                await coro
 
             # Prevent RedisVL from attempting to close the client
             # on an event loop in a separate thread.

langgraph/checkpoint/redis/ashallow.py

@@ -134,8 +134,10 @@ async def __aexit__(
         tb: Optional[TracebackType],
     ) -> None:
         if self._owns_its_client:
-            await self._redis.aclose()  # type: ignore[attr-defined]
-            await self._redis.connection_pool.disconnect()
+            await self._redis.aclose()
+            coro = self._redis.connection_pool.disconnect()
+            if coro:
+                await coro
 
             # Prevent RedisVL from attempting to close the client
             # on an event loop in a separate thread.

langgraph/checkpoint/redis/base.py

@@ -154,7 +154,7 @@ def set_client_info(self) -> None:
 
         try:
             # Try to use client_setinfo command if available
-            self._redis.client_setinfo("LIB-NAME", __full_lib_name__)  # type: ignore
+            self._redis.client_setinfo("LIB-NAME", __full_lib_name__)
         except (ResponseError, AttributeError):
             # Fall back to a simple echo if client_setinfo is not available
             try:
@@ -174,7 +174,7 @@ async def aset_client_info(self) -> None:
 
         try:
             # Try to use client_setinfo command if available
-            await self._redis.client_setinfo("LIB-NAME", client_info)  # type: ignore
+            await self._redis.client_setinfo("LIB-NAME", client_info)
         except (ResponseError, AttributeError):
             # Fall back to a simple echo if client_setinfo is not available
             try:
@@ -468,17 +468,17 @@ def put_writes(
                     # UPSERT case - only update specific fields
                     if key_exists:
                         # Update only channel, type, and blob fields
-                        pipeline.set(key, "$.channel", write_obj["channel"])  # type: ignore[arg-type]
-                        pipeline.set(key, "$.type", write_obj["type"])  # type: ignore[arg-type]
-                        pipeline.set(key, "$.blob", write_obj["blob"])  # type: ignore[arg-type]
+                        pipeline.set(key, "$.channel", write_obj["channel"])
+                        pipeline.set(key, "$.type", write_obj["type"])
+                        pipeline.set(key, "$.blob", write_obj["blob"])
                     else:
                         # For new records, set the complete object
-                        pipeline.set(key, "$", write_obj)  # type: ignore[arg-type]
+                        pipeline.set(key, "$", write_obj)
                         created_keys.append(key)
                 else:
                     # INSERT case - only insert if doesn't exist
                     if not key_exists:
-                        pipeline.set(key, "$", write_obj)  # type: ignore[arg-type]
+                        pipeline.set(key, "$", write_obj)
                         created_keys.append(key)
 
             pipeline.execute()

langgraph/checkpoint/redis/shallow.py

@@ -555,10 +555,10 @@ def put_writes(
                         pipeline.set(key, "$.blob", write_obj["blob"])
                     else:
                         # For new records, set the complete object
-                        pipeline.set(key, "$", write_obj)  # type: ignore[arg-type]
+                        pipeline.set(key, "$", write_obj)
                 else:
                     # INSERT case
-                    pipeline.set(key, "$", write_obj)  # type: ignore[arg-type]
+                    pipeline.set(key, "$", write_obj)
 
             pipeline.execute()