Repository: run-llama/llama_index
Pull request: Implement async methods in RedisKVStore
# Description

Async methods were not implemented in RedisKVStore. This is necessary if RedisKVStore is used as an ingestion cache or docstore in an async ingestion pipeline.

Fixes [#10248](https://github.com/run-llama/llama_index/issues/10248)

Implement async methods.

## New Package?

Did I fill in the `tool.llamahub` section in the `pyproject.toml` and provide a detailed README.md for my new integration or package?

- [ ] Yes
- [x] No

## Version Bump?

Did I bump the version in the `pyproject.toml` file of the package I am updating? (Except for the `llama-index-core` package)

- [ ] Yes
- [x] No

## Type of Change

Please delete options that are not relevant.

- [x] Bug fix (non-breaking change which fixes an issue)

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

- [x] I stared at the code and made sure it makes sense
- [x] Used in my own project code and it works properly 

## Suggested Checklist:

- [x] I have performed a self-review of my own code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] I have added Google Colab support for the newly added notebooks.
- [x] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] I ran `make format; make lint` to appease the lint gods


Problem statement:
[Bug]: IngestionPipeline 0.9.36 pipline.arun to Redis Not implemented Error 
### Bug Description

using IngestionPipeline  with docstore=RedisDocumentStore,   then running pipline.arun to Redis Not implemented Error  get Not implemented Error using llama-index 0.9.36. Downgrade to 0.9.30 allows me to run pipline.arun.  just using simpleDirectoryRead load_data documents and passing it to run or arun.

### Version

0.9.36 llama-index

### Steps to Reproduce

install llama-index 0.9.36 
create IngestionPipeline:
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        embed_model,
    ],
    docstore=RedisDocumentStore.from_host_and_port(
        "localhost", 6379, namespace="document_store"
    ),
    vector_store=RedisVectorStore(
        index_name="redis_vector_store",
        index_prefix="vectore_store",
        redis_url="redis://localhost:6379",
    ),
    cache=IngestionCache(
        cache=RedisCache.from_host_and_port("localhost", 6379),
        collection="redis_cache",
    ),
    docstore_strategy=DocstoreStrategy.UPSERTS,
)

attempt pipline.arun to a redis vectorstore
Error "Not implemented"

Workaround:
downgrade to llama-index 0.9.30  and both pipline.arun  and pipline.run  work



### Relevant Logs/Tracbacks

_No response_

Hints from issue discussion:
_🤖_
<!-- Greeting -->
Hello @JoeSF49! 

I'm [Dosu](https://dosu.dev), an AI here to assist you with your inquiries and issues relating to the LlamaIndex repository. I'm here to help you navigate through bugs, answer your questions, and guide you in contributing to the project while we're waiting for a human maintainer.

I've taken notice of the issue you've described and I'm currently looking into it. I'll get back to you with a thorough answer shortly.



Ah yea, need to implement the async methods here
https://github.com/run-llama/llama_index/blob/773a2fdcb4ecb06ceb26ee2f8322ec38b990d344/llama_index/storage/kvstore/redis_kvstore.py#L118
_🤖_
<!-- Answer -->
Thank you for reporting this issue. It seems like there's a problem with the implementation of the async methods in the Redis key-value store in the LlamaIndex version 0.9.36. 

To better understand and address this issue, could you please provide the following additional information:

1. The complete error message you're seeing when you attempt to run `pipeline.arun`.
2. The version of Redis you're using.
3. The operating system and its version on which you're running your LlamaIndex application.
4. Any other relevant configuration details that might help us understand the issue better.

This information will help us in diagnosing the problem more accurately. Thank you for your cooperation.


---

This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

i believe it has to do with Docstore strategy in the pipeling.  i also traced it to start failing at 0.9.33.  it's works up to llama-index 0.9.32

ERROR:
```python
---------------------------------------------------------------------------

NotImplementedError                       Traceback (most recent call last)

Cell In[18], line 2

      1 ## Async

----> 2 nodes = await pipeline.arun(documents=documents,show_progress=True)

      3 # nodes = pipeline.run(documents=documents,show_progress=True)

      4 print(f"Ingested {len(nodes)} Nodes")

 

File /opt/miniforge/envs/pytorch/lib/python3.10/site-packages/llama_index/ingestion/pipeline.py:570, in IngestionPipeline.arun(self, show_progress, documents, nodes, cache_collection, in_place, store_doc_text, num_workers, **kwargs)

    565 if self.docstore is not None and self.vector_store is not None:

    566     if self.docstore_strategy in (

    567         DocstoreStrategy.UPSERTS,

    568         DocstoreStrategy.UPSERTS_AND_DELETE,

    569     ):

--> 570         nodes_to_run = await self._ahandle_upserts(

    571             input_nodes, store_doc_text=store_doc_text

    572         )

    573     elif self.docstore_strategy == DocstoreStrategy.DUPLICATES_ONLY:

    574         nodes_to_run = await self._ahandle_duplicates(

    575             input_nodes, store_doc_text=store_doc_text

    576         )

 

File /opt/miniforge/envs/pytorch/lib/python3.10/site-packages/llama_index/ingestion/pipeline.py:513, in IngestionPipeline._ahandle_upserts(self, nodes, store_doc_text)

    509 """Handle docstore upserts by checking hashes and ids."""

    510 assert self.docstore is not None

    512 existing_doc_ids_before = set(

--> 513     (await self.docstore.aget_all_document_hashes()).values()

    514 )

    515 doc_ids_from_nodes = set()

    516 deduped_nodes_to_run = {}

 

File /opt/miniforge/envs/pytorch/lib/python3.10/site-packages/llama_index/storage/docstore/keyval_docstore.py:524, in KVDocumentStore.aget_all_document_hashes(self)

    522 """Get the stored hash for all documents."""

    523 hashes = {}

--> 524 for doc_id in await self._kvstore.aget_all(

    525     collection=self._metadata_collection

    526 ):

    527     hash = await self.aget_document_hash(doc_id)

    528     if hash is not None:

 

File /opt/miniforge/envs/pytorch/lib/python3.10/site-packages/llama_index/storage/kvstore/redis_kvstore.py:140, in RedisKVStore.aget_all(self, collection)

    138 async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:

    139     """Get all values from the store."""

--> 140     raise NotImplementedError

 

NotImplementedError:
```


As far as OSes:  i use docker workloads on an ML hardware platform
Redis-stack docker image.  redis version 7.2.3
Python env is basically ubuntu 22.04.  It's really using a nvidia/cuda-12.x-ubuntu22.04 image.  These has nvidia drivers and toolkits for hardware (GPU) passthrough.  i install mambaforge env within this workload "OS" and libs.  llama-index, langchain etc
Yea as mentioned above, async redis docstore operations just need to be implemented 
@dosu-bot  Hi and sorry I rarely make bug so i'm new to this process.   Do I have to provided any more information or will this fix be implemented in upcoming releases?  I cannot run the pipeline without the async as it fills the GPU memory during reg ".run".  Thanks
_🤖_
<!-- Answer -->
Thank you for the detailed report, JoeSF49. It seems like the issue is indeed related to the async methods not being implemented for the Redis docstore operations in the versions after 0.9.32. 

To help us further investigate and address this issue, could you please provide the following additional information:

1. The exact version of Python you are using.
2. The specific versions of the dependencies you are using, especially for Redis and any related packages.
3. Any custom configurations or settings you are using for Redis or LlamaIndex.
4. If possible, a minimal code snippet that reproduces the error. This will help us to replicate the issue on our end.

This information will help us understand the context better and expedite the process of finding a solution. Thank you for your cooperation.


---

This response is meant to be useful and save you time. It is **not meant to be a precise solution**, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

Commit message: implement async methods in RedisKVStore

Patch to review:
diff --git a/llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-redis/llama_index/storage/kvstore/redis/base.py b/llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-redis/llama_index/storage/kvstore/redis/base.py
index ad5812feb1..7bb1eff0be 100644
--- a/llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-redis/llama_index/storage/kvstore/redis/base.py
+++ b/llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-redis/llama_index/storage/kvstore/redis/base.py
@@ -11,7 +11,8 @@
 
 
 class RedisKVStore(BaseKVStore):
-    """Redis KV Store.
+    """
+    Redis KV Store.
 
     Args:
         redis_client (Any): Redis client
@@ -48,7 +49,8 @@ def __init__(
             raise ValueError("Either 'redis_client' or redis_url must be provided.")
 
     def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None:
-        """Put a key-value pair into the store.
+        """
+        Put a key-value pair into the store.
 
         Args:
             key (str): key
@@ -61,7 +63,8 @@ def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None
     async def aput(
         self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
     ) -> None:
-        """Put a key-value pair into the store.
+        """
+        Put a key-value pair into the store.
 
         Args:
             key (str): key
@@ -69,7 +72,7 @@ async def aput(
             collection (str): collection name
 
         """
-        raise NotImplementedError
+        return self.put(key=key, val=val, collection=collection)
 
     def put_all(
         self,
@@ -77,7 +80,8 @@ def put_all(
         collection: str = DEFAULT_COLLECTION,
         batch_size: int = DEFAULT_BATCH_SIZE,
     ) -> None:
-        """Put a dictionary of key-value pairs into the store.
+        """
+        Put a dictionary of key-value pairs into the store.
 
         Args:
             kv_pairs (List[Tuple[str, dict]]): key-value pairs
@@ -98,7 +102,8 @@ def put_all(
                 pipe.execute()
 
     def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
-        """Get a value from the store.
+        """
+        Get a value from the store.
 
         Args:
             key (str): key
@@ -113,14 +118,15 @@ def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
     async def aget(
         self, key: str, collection: str = DEFAULT_COLLECTION
     ) -> Optional[dict]:
-        """Get a value from the store.
+        """
+        Get a value from the store.
 
         Args:
             key (str): key
             collection (str): collection name
 
         """
-        raise NotImplementedError
+        return self.get(key=key, collection=collection)
 
     def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
         """Get all values from the store."""
@@ -132,10 +138,11 @@ def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
 
     async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
         """Get all values from the store."""
-        raise NotImplementedError
+        return self.get_all(collection=collection)
 
     def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
-        """Delete a value from the store.
+        """
+        Delete a value from the store.
 
         Args:
             key (str): key
@@ -146,14 +153,15 @@ def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
         return bool(deleted_num > 0)
 
     async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
-        """Delete a value from the store.
+        """
+        Delete a value from the store.
 
         Args:
             key (str): key
             collection (str): collection name
 
         """
-        raise NotImplementedError
+        return self.delete(key=key, collection=collection)
 
     @classmethod
     def from_host_and_port(
@@ -161,7 +169,8 @@ def from_host_and_port(
         host: str,
         port: int,
     ) -> "RedisKVStore":
-        """Load a RedisKVStore from a Redis host and port.
+        """
+        Load a RedisKVStore from a Redis host and port.
 
         Args:
             host (str): Redis host
@@ -172,7 +181,8 @@ def from_host_and_port(
 
     @classmethod
     def from_redis_client(cls, redis_client: Any) -> "RedisKVStore":
-        """Load a RedisKVStore from a Redis Client.
+        """
+        Load a RedisKVStore from a Redis Client.
 
         Args:
             redis_client (Redis): Redis client


Code review comments:
diff_hunk: @@ -61,23 +63,25 @@ def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None
     async def aput(
         self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
     ) -> None:
-        """Put a key-value pair into the store.
+        """
+        Put a key-value pair into the store.
 
         Args:
             key (str): key
             val (dict): value
             collection (str): collection name
 
         """
-        raise NotImplementedError
+        return self.put(key=key, val=val, collection=collection)
path: llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-redis/llama_index/storage/kvstore/redis/base.py
text: @user1:
Should we take the time to actually implement the async method though?

@user1:
fake-async will still block the thread

@user1:
@author light bump on this

@author:
Unfortunately I don't have time to implement true async methods. I've used this fake async implementation with async ingestion pipeline and have not experienced any major performance degradation for my own use cases.