Zum Hauptinhalt springen

Caching - In-Memory, Redis, s3, Redis Semantic Cache, Disk

Code anzeigen

Info

Cache initialisieren - In-Memory, Redis, S3 Bucket, Redis Semantic, Disk Cache, Qdrant Semantic​

Redis installieren

pip install redis

Für die gehostete Version können Sie Ihre eigene Redis-DB hier einrichten: https://redis.io/try-free/

import litellm
from litellm import completion
from litellm.caching.caching import Cache

litellm.cache = Cache(type="redis", host=<host>, port=<port>, password=<password>)

# Make completion calls
response1 = completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Tell me a joke."}]
)
response2 = completion(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Tell me a joke."}]
)

# response1 == response2, response 1 is cached

Cache pro LiteLLM-Aufruf ein-/ausschalten​

LiteLLM unterstützt 4 Cache-Kontrollen

  • no-cache: Optional(bool) Wenn True, wird keine gecachte Antwort zurückgegeben, sondern stattdessen der eigentliche Endpunkt aufgerufen.
  • no-store: Optional(bool) Wenn True, wird die Antwort nicht gecacht.
  • ttl: Optional(int) - Die Antwort wird für die vom Benutzer definierte Zeit (in Sekunden) gecacht.
  • s-maxage: Optional(int) Akzeptiert nur gecachte Antworten, die innerhalb des vom Benutzer definierten Bereichs (in Sekunden) liegen.

Lassen Sie uns wissen, wenn Sie mehr benötigen

Beispielverwendung no-cache - Wenn True, wird keine gecachte Antwort zurückgegeben

response = litellm.completion(
        model="gpt-3.5-turbo",
        messages=[
            {
                "role": "user",
                "content": "hello who are you"
            }
        ],
        cache={"no-cache": True},
    )

Cache-Kontextmanager - Aktivieren, Deaktivieren, Cache aktualisieren​

Verwenden Sie den Kontextmanager, um den LiteLLM-Cache einfach zu aktivieren, zu deaktivieren und zu aktualisieren

Cache aktivieren​

Schnellstart Aktivierung

litellm.enable_cache()

Erweiterte Parameter

litellm.enable_cache(
    type: Optional[Literal["local", "redis", "s3", "disk"]] = "local",
    host: Optional[str] = None,
    port: Optional[str] = None,
    password: Optional[str] = None,
    supported_call_types: Optional[
        List[Literal["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"]]
    ] = ["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"],
    **kwargs,
)

Cache deaktivieren​

Caching ausschalten

litellm.disable_cache()

Cache-Parameter aktualisieren (Redis-Host, Port usw.)​

Cache-Parameter aktualisieren

litellm.update_cache(
    type: Optional[Literal["local", "redis", "s3", "disk"]] = "local",
    host: Optional[str] = None,
    port: Optional[str] = None,
    password: Optional[str] = None,
    supported_call_types: Optional[
        List[Literal["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"]]
    ] = ["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"],
    **kwargs,
)

Benutzerdefinierte Cache-Schlüssel:​

Funktion definieren, um den Cache-Schlüssel zurückzugeben

# this function takes in *args, **kwargs and returns the key you want to use for caching
def custom_get_cache_key(*args, **kwargs):
    # return key to use for your cache:
    key = kwargs.get("model", "") + str(kwargs.get("messages", "")) + str(kwargs.get("temperature", "")) + str(kwargs.get("logit_bias", ""))
    print("key for cache", key)
    return key

Setzen Sie Ihre Funktion als litellm.cache.get_cache_key

from litellm.caching.caching import Cache

cache = Cache(type="redis", host=os.environ['REDIS_HOST'], port=os.environ['REDIS_PORT'], password=os.environ['REDIS_PASSWORD'])

cache.get_cache_key = custom_get_cache_key # set get_cache_key function for your cache

litellm.cache = cache # set litellm.cache to your cache

So schreiben Sie benutzerdefinierte Add/Get-Cache-Funktionen​

1. Cache initialisieren​

from litellm.caching.caching import Cache
cache = Cache()

2. Benutzerdefinierte Add/Get-Cache-Funktionen definieren​

def add_cache(self, result, *args, **kwargs):
  your logic
  
def get_cache(self, *args, **kwargs):
  your logic

3. Cache Add/Get-Funktionen auf Ihre Add/Get-Funktionen verweisen​

cache.add_cache = add_cache
cache.get_cache = get_cache

Cache-Initialisierungsparameter​

def __init__(
    self,
    type: Optional[Literal["local", "redis", "redis-semantic", "s3", "disk"]] = "local",
    supported_call_types: Optional[
        List[Literal["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"]]
    ] = ["completion", "acompletion", "embedding", "aembedding", "atranscription", "transcription"],
    ttl: Optional[float] = None,
    default_in_memory_ttl: Optional[float] = None,

    # redis cache params
    host: Optional[str] = None,
    port: Optional[str] = None,
    password: Optional[str] = None,
    namespace: Optional[str] = None,
    default_in_redis_ttl: Optional[float] = None,
    redis_flush_size=None,

    # redis semantic cache params
    similarity_threshold: Optional[float] = None,
    redis_semantic_cache_embedding_model: str = "text-embedding-ada-002",
    redis_semantic_cache_index_name: Optional[str] = None,

    # s3 Bucket, boto3 configuration
    s3_bucket_name: Optional[str] = None,
    s3_region_name: Optional[str] = None,
    s3_api_version: Optional[str] = None,
    s3_path: Optional[str] = None, # if you wish to save to a specific path
    s3_use_ssl: Optional[bool] = True,
    s3_verify: Optional[Union[bool, str]] = None,
    s3_endpoint_url: Optional[str] = None,
    s3_aws_access_key_id: Optional[str] = None,
    s3_aws_secret_access_key: Optional[str] = None,
    s3_aws_session_token: Optional[str] = None,
    s3_config: Optional[Any] = None,

    # disk cache params
    disk_cache_dir=None,

    # qdrant cache params
    qdrant_api_base: Optional[str] = None,
    qdrant_api_key: Optional[str] = None,
    qdrant_collection_name: Optional[str] = None,
    qdrant_quantization_config: Optional[str] = None,
    qdrant_semantic_cache_embedding_model="text-embedding-ada-002",

    **kwargs
):

Protokollierung​

Cache-Treffer werden in Erfolgsereignissen als kwarg["cache_hit"] protokolliert.

Hier ist ein Beispiel für den Zugriff darauf

import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm import completion, acompletion, Cache

# create custom callback for success_events
class MyCustomHandler(CustomLogger):
  async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): 
      print(f"On Success")
      print(f"Value of Cache hit: {kwargs['cache_hit']"})

async def test_async_completion_azure_caching():
  # set custom callback
  customHandler_caching = MyCustomHandler()
  litellm.callbacks = [customHandler_caching]

  # init cache 
  litellm.cache = Cache(type="redis", host=os.environ['REDIS_HOST'], port=os.environ['REDIS_PORT'], password=os.environ['REDIS_PASSWORD'])
  unique_time = time.time()
  response1 = await litellm.acompletion(model="azure/chatgpt-v-2",
                          messages=[{
                              "role": "user",
                              "content": f"Hi 👋 - i'm async azure {unique_time}"
                          }],
                          caching=True)
  await asyncio.sleep(1)
  print(f"customHandler_caching.states pre-cache hit: {customHandler_caching.states}")
  response2 = await litellm.acompletion(model="azure/chatgpt-v-2",
                          messages=[{
                              "role": "user",
                              "content": f"Hi 👋 - i'm async azure {unique_time}"
                          }],
                          caching=True)
  await asyncio.sleep(1) # success callbacks are done in parallel