Error when langgraph AsyncRedisStore's _batch_put_ops method uses a pipeline when in cluster mode

[SunilRath]

I am getting the below error when the langmem memory_store_manager is trying to run a _batch_put_ops on the Redis DB using langgraph redis store. This was working fine when i was using langgraph InMemory Store.

The langgraph AsyncRedisStore's _batch_put_ops method uses a pipeline when in cluster mode. The redis-py library's JSON command helper, get_protocol_version, does not know how to handle a ClusterPipeline object. It expects a regular client object that has a nodes_manager attribute, but the pipeline object doesn't have one.

[redisvl.index.index] Error while loading data to Redis 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'

Have anyone faced this issue? I am using "redis[hiredis]==6.4.0", "redisvl==0.8.0", "langgraph==0.5.3", "langmem==0.0.28",

[rbs333]

@SunilRath thanks for posting! Have not seen this issue before. Have added a ticket on our side to track.

[SunilRath]

I was trying both checkpointer and Store with Redis. When I upgrade langgraph-redis to 0.1.1 I get that error with the checkpointer but not from the store anymore. The store gives error with langgraph-checkpoint-redis==0.0.8. Might be a issue with interaction between redisvl and redis-py.

The checkpointer error is:

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_ver
sion
return client.nodes_manager.connection_kwargs.get("protocol")
^^^^^^^^^^^^^^^^^^^^
AttributeError: 'ClusterPipeline' object has no attribute 'nodes_manager'

[rbs333]

@SunilRath could you provide more details on your redis cluster configuration and how you are creating the client?

For example are you using: from redis.asyncio.cluster import RedisCluster?

If you provide it can help us reproduce.

[SunilRath]

Yes, I am using from redis.asyncio.cluster import RedisCluster and then creating the connection pool using that.
RedisCluster(startup_nodes=startup_nodes_list,...) .
Then, I am creating Langmem memory store manager using create_memory_store_manager and assign that store. After that, I am using the langmem ReflectionExecutor to create/update/delete records in that Redis store.

Appreciate your help with this @rbs333 . Thanks.