Caching for Tools

[logar16]

I am using a search API tool and wanting to keep within the free tier (1k requests/month). As I do development, I will be reusing the same prompt a lot and was wondering if there is a convenient way to do caching for tools? It looks like BaseCache is very much model-focused. That's fine, but what if tools want persistent caching beyond adding a simple @lru_cache? Or is the assumption that tools should always go get latest value (could be expensive)?

[dosubot[bot]]

Hey @logar16! I'm here to help you with any bugs, questions, or contributions. If you need assistance, feel free to ask.

To implement persistent caching for a search API tool beyond using @lru_cache, you can use various caching solutions provided by the LangChain framework. Here are some options:

1. SQLiteCache: Uses SQLite as a backend for persistent storage.
2. CassandraCache: Uses Cassandra for exact or semantic caching.
3. AstraDBCache: Uses Astra DB for exact or semantic caching.
4. AzureCosmosDBSemanticCache: Uses Azure Cosmos DB for semantic caching.
5. ElasticsearchCache: Uses Elasticsearch for caching.
6. OpenSearchSemanticCache: Uses OpenSearch for semantic caching.
7. MongoDBCache: Uses MongoDB for caching.
8. MongoDBAtlasSemanticCache: Uses MongoDB Atlas for semantic caching.

Example with SQLiteCache

from langchain_community.cache import SQLiteCache

# Initialize the SQLiteCache with the desired database path
cache = SQLiteCache(database_path="path_to_your_database.db")

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with CassandraCache

from langchain_community.cache import CassandraCache
from langchain_core.globals import set_llm_cache

set_llm_cache(CassandraCache())

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with AstraDBCache

from langchain_astradb import AstraDBCache
from langchain_core.globals import set_llm_cache

set_llm_cache(
    AstraDBCache(
        api_endpoint="your_astra_db_api_endpoint",
        token="your_astra_db_application_token",
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with AzureCosmosDBSemanticCache

from langchain_community.cache import AzureCosmosDBSemanticCache
from langchain_openai import OpenAIEmbeddings

set_llm_cache(
    AzureCosmosDBSemanticCache(
        cosmosdb_connection_string="your_cosmosdb_connection_string",
        embedding=OpenAIEmbeddings(),
        database_name="your_database_name",
        collection_name="your_collection_name",
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with ElasticsearchCache

from langchain_elasticsearch import ElasticsearchCache
from langchain_core.globals import set_llm_cache

set_llm_cache(
    ElasticsearchCache(
        es_url="http://localhost:9200",
        index_name="llm-chat-cache",
        metadata={"project": "my_chatgpt_project"},
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with OpenSearchSemanticCache

from langchain_community.cache import OpenSearchSemanticCache
from langchain_openai import OpenAIEmbeddings

set_llm_cache(
    OpenSearchSemanticCache(
        opensearch_url="http://localhost:9200", 
        embedding=OpenAIEmbeddings()
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with MongoDBCache

from langchain_mongodb.cache import MongoDBCache
from langchain_core.globals import set_llm_cache

set_llm_cache(
    MongoDBCache(
        connection_string="your_mongodb_connection_string",
        database_name="your_database_name",
        collection_name="your_collection_name",
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

Example with MongoDBAtlasSemanticCache

from langchain_mongodb.cache import MongoDBAtlasSemanticCache
from langchain_openai import OpenAIEmbeddings

set_llm_cache(
    MongoDBAtlasSemanticCache(
        connection_string="your_mongodb_atlas_connection_string",
        embedding=OpenAIEmbeddings(),
        database_name="your_database_name",
        collection_name="your_collection_name",
    )
)

# Example usage
prompt = "example prompt"
llm_string = "example_llm_string"
return_val = ["example_return_value"]

# Update the cache
cache.update(prompt, llm_string, return_val)

# Lookup in the cache
cached_value = cache.lookup(prompt, llm_string)
print(cached_value)

These solutions provide various options for persistent caching, helping you stay within the free tier by reducing the number of API calls. Choose the one that best fits your infrastructure and requirements ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[logar16]

But if the tool I'm using doesn't have an llm_string I'm being forced to put a round peg in a square hole. This feels very hacky and prone to breaking when the BaseCache is changed to better fit llm needs but makes it even harder to use for tools

[dosubot[bot]]

To implement persistent caching for a tool that doesn't use llm_string without relying on BaseCache, you can create a custom cache class using SQLAlchemy for persistence. Here is an example implementation:

from sqlalchemy import create_engine, Column, String, Integer, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
import json
import hashlib

Base = declarative_base()

def _hash(_input: str) -> str:
    """Use a deterministic hashing approach."""
    return hashlib.md5(_input.encode()).hexdigest()

def _dumps_generations(generations) -> str:
    """Serialization for generic RETURN_VAL_TYPE, i.e. sequence of `Generation`"""
    return json.dumps([gen.dict() for gen in generations])

def _loads_generations(generations_str: str):
    """Deserialization of a string into a generic RETURN_VAL_TYPE"""
    try:
        results = json.loads(generations_str)
        return [Generation(**gen_dict) for gen_dict in results]
    except json.JSONDecodeError:
        raise ValueError(f"Could not decode json to list of generations: {generations_str}")

class Generation:
    def __init__(self, text):
        self.text = text

    def dict(self):
        return {"text": self.text}

class FullLLMCache(Base):
    """SQLite table for full LLM Cache (all generations)."""
    __tablename__ = "full_llm_cache"
    prompt = Column(String, primary_key=True)
    idx = Column(Integer, primary_key=True)
    response = Column(String)

class CustomCache:
    """Cache that uses SQLAlchemy as a backend without relying on BaseCache."""

    def __init__(self, database_path: str = ".custom_cache.db"):
        """Initialize by creating the engine and all tables."""
        self.engine = create_engine(f"sqlite:///{database_path}")
        Base.metadata.create_all(self.engine)

    def lookup(self, prompt: str) -> Optional[list]:
        """Look up based on prompt."""
        stmt = select(FullLLMCache.response).where(FullLLMCache.prompt == prompt).order_by(FullLLMCache.idx)
        with Session(self.engine) as session:
            rows = session.execute(stmt).fetchall()
            if rows:
                return _loads_generations(json.dumps([row[0] for row in rows]))
        return None

    def update(self, prompt: str, return_val: list) -> None:
        """Update based on prompt."""
        items = [
            FullLLMCache(prompt=prompt, response=_dumps_generations([gen]), idx=i)
            for i, gen in enumerate(return_val)
        ]
        with Session(self.engine) as session, session.begin():
            for item in items:
                session.merge(item)

    def clear(self) -> None:
        """Clear cache."""
        with Session(self.engine) as session:
            session.query(FullLLMCache).delete()
            session.commit()

This implementation uses SQLAlchemy to create a persistent cache stored in an SQLite database. The CustomCache class provides methods to look up, update, and clear the cache based on a prompt. The llm_string is not used in this implementation ^[1].

To continue talking to Dosu, mention @dosu.

[logar16]

Should every tool in langchain that wants persistence be forced to roll their own caching or should the maintainers add a common caching capability for all tools?

[dosubot[bot]]

To address your concern, it would be beneficial for the maintainers to add a common caching capability for all tools in LangChain. This would ensure consistency and reduce the need for each tool to implement its own caching solution, making development more efficient and less error-prone ^[1][2][3].

This is a tough nut to crack, could really use your help on this, @eyurtsev.

To continue talking to Dosu, mention @dosu.