`AttributeError: 'ClusterPipeline' object has no attribute 'nodes_manager'` when using `index.load()` with Redis Cluster

[nkilleen-work]

Description

When using redisvl with a Redis Cluster client, calling the index.load() method results in an AttributeError. The error originates from within the redis-py library, which redisvl calls during its batch writing process.

The traceback indicates that redisvl's storage layer uses a ClusterPipeline object for its awrite operation. An internal helper function in redis-py, get_protocol_version, is then called, which attempts to access client.nodes_manager. However, the ClusterPipeline object does not have this attribute, leading to the crash.

This issue occurs when redisvl is used as a dependency, for example, within langgraph-checkpoint-redis when it attempts to save a checkpoint to a Redis Cluster.

Library Versions

• redisvl: 0.8.0
• redis: 6.4.0 (also observed with other 5.x versions including 5.3.1)
• langgraph-checkpoint-redis: 0.1.1
• python: 3.12

Traceback

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/redisvl/index/index.py", line 1566, in load
    return await self._storage.awrite(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redisvl/index/storage.py", line 477, in awrite
    await self._aset(pipe, key, obj)
  File "/usr/local/lib/python3.12/site-packages/redisvl/index/storage.py", line 679, in _aset
    client.json().set(key, "$", obj)  # type: ignore[return-value, misc]
    ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/commands/redismodules.py", line 24, in json
    jj = JSON(client=self, encoder=encoder, decoder=decoder)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/commands/json/__init__.py", line 71, in __init__
    if get_protocol_version(self.client) in ["3", 3]:
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/commands/helpers.py", line 118, in get_protocol_version
    return client.nodes_manager.connection_kwargs.get("protocol")
           ^^^^^^^^^^^^^^^^^^^^
AttributeError: 'ClusterPipeline' object has no attribute 'nodes_manager'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/loop.py", line 1123, in _checkpointer_put_after_previous
    await prev
  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/loop.py", line 1125, in _checkpointer_put_after_previous
    await cast(BaseCheckpointSaver, self.checkpointer).aput(
  File "/usr/local/lib/python3.12/site-packages/langgraph/checkpoint/redis/aio.py", line 1079, in aput
    raise e
  File "/usr/local/lib/python3.12/site-packages/langgraph/checkpoint/redis/aio.py", line 998, in aput
    await self.checkpoints_index.load(
  File "/usr/local/lib/python3.12/site-packages/redisvl/index/index.py", line 1583, in load
    raise RedisVLError(f"Failed to load data: {str(e)}") from e
redisvl.exceptions.RedisVLError: Failed to load data: 'ClusterPipeline' object has no attribute 'nodes_manager'

[rbs333]

@nkilleen-work thanks for posting! We've created a ticket on our side and are looking into it

[rbs333]

@nkilleen-work could you provide some more details about how you're establish and maintaining your redis cluster client?

For example, are you using from redis.asyncio.cluster import RedisCluster to create the connection and what is the underlying configuration for those instances?

[nkilleen-work]

Hi @rbs333, thanks for looking into this. I thought you might ask this question! Let me give you the important parts you need.

Yes, I am using from redis.asyncio.cluster import RedisCluster to create the connection when more than one node is configured.

The client is created and managed within a RedisManager class. Here is the relevant code showing how the RedisCluster instance is configured and instantiated:

from redis.asyncio.cluster import ClusterNode, RedisCluster

class RedisManager:
    # ...

    async def initialize(self) -> None:
        # ...
        try:
            # self._config is a Pydantic model for configuration
            nodes_str = self._config.cluster.nodes
            password = self._config.password.get_secret_value() if self._config.password else None
            username = self._config.user if self._config.user else None

            # This part parses a comma-separated string of "host:port" into `parsed_nodes`
            # ... (parsing logic omitted for brevity)

            if len(parsed_nodes) > 1:
                startup_nodes_list = [
                    ClusterNode(host=node["host"], port=node["port"]) for node in parsed_nodes
                ]
                
                self.redis = RedisCluster(
                    startup_nodes=startup_nodes_list,
                    username=username,
                    password=password,
                    decode_responses=True,
                    ssl=True,
                )
            # ...
        # ...

The configuration for the cluster client is driven by a configuration object that provides the node list, username, and password. decode_responses and ssl are both set to True.

Let me know if you need any more details.

PS. SunilRath works with me and we're using the same setup. The issue he made is similar to this one: redis-developer/langgraph-redis#96. The issue I reference here comes from using the AsyncRedisSaver in the latest version of langgraph-checkpoint-redis==0.1.1.

I do not get this issue with AsyncRedisSaver on langgraph-checkpoint-redis==0.0.8, but to prevent the issue I have to downgrade a number of other deps as well:

"google-cloud-aiplatform==1.105.0",
"langchain-community>=0.0.31",
"langchain-core==0.3.69",
"langchain-google-vertexai==2.0.27",
"langchain==0.3.26",
"langgraph==0.5.3",
"redis[hiredis]==6.4.0",
"redisvl==0.8.0",

Even with the downgraded deps, the issue using the AsyncRedisStore still persists in redis-developer/langgraph-redis#96