langchain[minor], community[minor], core[minor]: Async Cache support and AsyncRedisCache

libs/community/langchain_community/cache.py

@@ -52,7 +52,7 @@
 except ImportError:
     from sqlalchemy.ext.declarative import declarative_base
 
-from langchain_core.caches import RETURN_VAL_TYPE, BaseCache
+from langchain_core.caches import RETURN_VAL_TYPE, AsyncBaseCache, BaseCache
 from langchain_core.embeddings import Embeddings
 from langchain_core.language_models.llms import LLM, get_prompts
 from langchain_core.load.dump import dumps
@@ -170,7 +170,7 @@ def _loads_generations(generations_str: str) -> Union[RETURN_VAL_TYPE, None]:
         return None
 
 
-class InMemoryCache(BaseCache):
+class InMemoryCache(BaseCache, AsyncBaseCache):
     """Cache that stores things in memory."""
 
     def __init__(self) -> None:
@@ -189,6 +189,17 @@ def clear(self, **kwargs: Any) -> None:
         """Clear cache."""
         self._cache = {}
 
+    async def alookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
+        return self.lookup(prompt, llm_string)
+
+    async def aupdate(
+        self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE
+    ) -> None:
+        return self.update(prompt, llm_string, return_val)
+
+    async def aclear(self, **kwargs: Any) -> None:
+        return self.clear(**kwargs)
+
 
 Base = declarative_base()
 
@@ -349,21 +360,25 @@ def clear(self, **kwargs: Any) -> None:
         self.redis.flushdb(flush_type=asynchronous)
 
 
-class RedisCache(BaseCache):
-    """Cache that uses Redis as a backend."""
+class RedisCache(BaseCache, AsyncBaseCache):
+    """
+    Cache that uses Redis as a backend. Allows to use either sync or
+    async Redis client. Depending on the client passed, you are expected
+    to use either sync or async methods.
+    """
 
     def __init__(self, redis_: Any, *, ttl: Optional[int] = None):
         """
         Initialize an instance of RedisCache.
 
         This method initializes an object with Redis caching capabilities.
         It takes a `redis_` parameter, which should be an instance of a Redis
-        client class, allowing the object to interact with a Redis
-        server for caching purposes.
+        client class (`redis.Redis` or `redis.asyncio.Redis`), allowing the object
+        to interact with a Redis server for caching purposes.
 
         Parameters:
             redis_ (Any): An instance of a Redis client class
-                (e.g., redis.Redis) used for caching.
+                (`redis.Redis` or `redis.asyncio.Redis`) to be used for caching.
                 This allows the object to communicate with a
                 Redis server for caching operations.
             ttl (int, optional): Time-to-live (TTL) for cached items in seconds.
@@ -372,26 +387,49 @@ def __init__(self, redis_: Any, *, ttl: Optional[int] = None):
                 have an automatic expiration.
         """
         try:
-            from redis import Redis
+            from redis import Redis as SyncRedis
+            from redis.asyncio import Redis as AsyncRedis
         except ImportError:
             raise ValueError(
                 "Could not import redis python package. "
                 "Please install it with `pip install redis`."
             )
-        if not isinstance(redis_, Redis):
-            raise ValueError("Please pass in Redis object.")
+        if isinstance(redis_, SyncRedis):
+            self._async = False
+        elif isinstance(redis_, AsyncRedis):
+            self._async = True
+        else:
+            raise ValueError("Please pass a valid Redis client.")
         self.redis = redis_
         self.ttl = ttl
 
     def _key(self, prompt: str, llm_string: str) -> str:
         """Compute key from prompt and llm_string"""
         return _hash(prompt + llm_string)
 
-    def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
-        """Look up based on prompt and llm_string."""
+    def __ensure_sync(self, function_name: str = "function"):
+        if self._async:
+            raise ValueError(f"Cannot use sync {function_name} with async Redis client")
+
+    def __ensure_async(self, function_name: str = "function"):
+        if not self._async:
+            logger.warning(
+                f"Performing async {function_name} with sync Redis client. "
+                "This may block event loop. Consider using `redis.asyncio.Redis`"
+            )
+
+    @staticmethod
+    def __ensure_generation_type(return_val: RETURN_VAL_TYPE):
+        for gen in return_val:
+            if not isinstance(gen, Generation):
+                raise ValueError(
+                    "RedisCache only supports caching of normal LLM generations, "
+                    f"got {type(gen)}"
+                )
+
+    @staticmethod
+    def __get_generations(results: dict[str | bytes, str | bytes]) -> list[Generation]:
         generations = []
-        # Read from a Redis HASH
-        results = self.redis.hgetall(self._key(prompt, llm_string))
         if results:
             for _, text in results.items():
                 try:
@@ -408,35 +446,66 @@ def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
                     generations.append(Generation(text=text))
         return generations if generations else None
 
+    def __configure_pipeline_for_update(self, key, pipe, return_val):
+        pipe.hset(
+            key,
+            mapping={
+                str(idx): dumps(generation) for idx, generation in enumerate(return_val)
+            },
+        )
+        if self.ttl is not None:
+            pipe.expire(key, self.ttl)
+
+    def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
+        """Look up based on prompt and llm_string."""
+        self.__ensure_sync("lookup")
+        # Read from a Redis HASH
+        results = self.redis.hgetall(self._key(prompt, llm_string))
+        return self.__get_generations(results)
+
+    async def alookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
+        """Look up based on prompt and llm_string. Async version."""
+        self.__ensure_async("alookup")
+        results = await self.redis.hgetall(self._key(prompt, llm_string))
+        return self.__get_generations(results)
+
     def update(self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE) -> None:
         """Update cache based on prompt and llm_string."""
-        for gen in return_val:
-            if not isinstance(gen, Generation):
-                raise ValueError(
-                    "RedisCache only supports caching of normal LLM generations, "
-                    f"got {type(gen)}"
-                )
-        # Write to a Redis HASH
+        self.__ensure_sync("update")
+        self.__ensure_generation_type(return_val)
         key = self._key(prompt, llm_string)
 
         with self.redis.pipeline() as pipe:
-            pipe.hset(
-                key,
-                mapping={
-                    str(idx): dumps(generation)
-                    for idx, generation in enumerate(return_val)
-                },
-            )
-            if self.ttl is not None:
-                pipe.expire(key, self.ttl)
-
+            self.__configure_pipeline_for_update(key, pipe, return_val)
             pipe.execute()
 
+    async def aupdate(
+        self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE
+    ) -> None:
+        """Update cache based on prompt and llm_string. Async version."""
+        self.__ensure_async("aupdate")
+        self.__ensure_generation_type(return_val)
+        key = self._key(prompt, llm_string)
+
+        async with self.redis.pipeline() as pipe:
+            self.__configure_pipeline_for_update(key, pipe, return_val)
+            await pipe.execute()
+
     def clear(self, **kwargs: Any) -> None:
         """Clear cache. If `asynchronous` is True, flush asynchronously."""
+        self.__ensure_sync("clear")
         asynchronous = kwargs.get("asynchronous", False)
         self.redis.flushdb(asynchronous=asynchronous, **kwargs)
 
+    async def aclear(self, **kwargs: Any) -> None:
+        """
+        Clear cache. If `asynchronous` is True, flush asynchronously.
+        Async version.
+        """
+        self.__ensure_async("aclear")
+        asynchronous = kwargs.get("asynchronous", False)
+        await self.redis.flushdb(asynchronous=asynchronous, **kwargs)
+
 
 class RedisSemanticCache(BaseCache):
     """Cache that uses Redis as a vector-store backend."""

libs/core/langchain_core/caches.py

@@ -22,3 +22,21 @@ def update(self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE) -> N
     @abstractmethod
     def clear(self, **kwargs: Any) -> None:
         """Clear cache that can take additional keyword arguments."""
+
+
+class AsyncBaseCache(ABC):
+    """Base interface for async cache."""
+
+    @abstractmethod
+    async def alookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
+        """Look up based on prompt and llm_string."""
+
+    @abstractmethod
+    async def aupdate(
+        self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE
+    ) -> None:
+        """Update cache based on prompt and llm_string."""
+
+    @abstractmethod
+    async def aclear(self, **kwargs: Any) -> None:
+        """Clear cache that can take additional keyword arguments."""

libs/core/langchain_core/globals/__init__.py

@@ -1,10 +1,10 @@
 # flake8: noqa
 """Global values and configuration that apply to all of LangChain."""
 import warnings
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Union
 
 if TYPE_CHECKING:
-    from langchain_core.caches import BaseCache
+    from langchain_core.caches import BaseCache, AsyncBaseCache
 
 
 # DO NOT USE THESE VALUES DIRECTLY!
@@ -13,7 +13,7 @@
 # https://github.com/langchain-ai/langchain/pull/11311#issuecomment-1743780004
 _verbose: bool = False
 _debug: bool = False
-_llm_cache: Optional["BaseCache"] = None
+_llm_cache: Optional[Union["BaseCache", "AsyncBaseCache"]] = None
 
 
 def set_verbose(value: bool) -> None:
@@ -134,7 +134,7 @@ def get_debug() -> bool:
     return _debug or old_debug
 
 
-def set_llm_cache(value: Optional["BaseCache"]) -> None:
+def set_llm_cache(value: Optional[Union["BaseCache", "AsyncBaseCache"]]) -> None:
     """Set a new LLM cache, overwriting the previous value, if any."""
     try:
         import langchain  # type: ignore[import]
@@ -162,7 +162,7 @@ def set_llm_cache(value: Optional["BaseCache"]) -> None:
     _llm_cache = value
 
 
-def get_llm_cache() -> "BaseCache":
+def get_llm_cache() -> Union["BaseCache", "AsyncBaseCache"]:
     """Get the value of the `llm_cache` global setting."""
     try:
         import langchain  # type: ignore[import]

libs/core/langchain_core/language_models/chat_models.py

@@ -17,6 +17,7 @@
 )
 
 from langchain_core._api import deprecated
+from langchain_core.caches import AsyncBaseCache, BaseCache
 from langchain_core.callbacks import (
     AsyncCallbackManager,
     AsyncCallbackManagerForLLMRun,
@@ -581,7 +582,12 @@ def _generate_with_cache(
         else:
             llm_string = self._get_llm_string(stop=stop, **kwargs)
             prompt = dumps(messages)
-            cache_val = llm_cache.lookup(prompt, llm_string)
+            if isinstance(llm_cache, BaseCache):
+                cache_val = llm_cache.lookup(prompt, llm_string)
+            elif isinstance(llm_cache, AsyncBaseCache):
+                raise ValueError("Async cache is not supported in this context")
+            else:
+                raise ValueError("Unknown type of llm_cache instance")
             if isinstance(cache_val, list):
                 return ChatResult(generations=cache_val)
             else:
@@ -591,7 +597,12 @@ def _generate_with_cache(
                     )
                 else:
                     result = self._generate(messages, stop=stop, **kwargs)
-                llm_cache.update(prompt, llm_string, result.generations)
+                if isinstance(llm_cache, BaseCache):
+                    llm_cache.update(prompt, llm_string, result.generations)
+                elif isinstance(llm_cache, AsyncBaseCache):
+                    raise ValueError("Async cache is not supported in this context")
+                else:
+                    raise ValueError("Unknown type of llm_cache instance")
                 return result
 
     async def _agenerate_with_cache(
@@ -621,7 +632,12 @@ async def _agenerate_with_cache(
         else:
             llm_string = self._get_llm_string(stop=stop, **kwargs)
             prompt = dumps(messages)
-            cache_val = llm_cache.lookup(prompt, llm_string)
+            if isinstance(llm_cache, AsyncBaseCache):
+                cache_val = await llm_cache.alookup(prompt, llm_string)
+            elif isinstance(llm_cache, BaseCache):
+                cache_val = llm_cache.lookup(prompt, llm_string)
+            else:
+                raise ValueError("Unknown type of llm_cache instance")
             if isinstance(cache_val, list):
                 return ChatResult(generations=cache_val)
             else:
@@ -631,7 +647,10 @@ async def _agenerate_with_cache(
                     )
                 else:
                     result = await self._agenerate(messages, stop=stop, **kwargs)
-                llm_cache.update(prompt, llm_string, result.generations)
+                if isinstance(llm_cache, AsyncBaseCache):
+                    await llm_cache.aupdate(prompt, llm_string, result.generations)
+                elif isinstance(llm_cache, BaseCache):
+                    llm_cache.update(prompt, llm_string, result.generations)
                 return result
 
     @abstractmethod