Caching
title: Caching Techniques in Python: In-Memory, Disk, and Redis description: Explore caching methods in Python with functools, diskcache, and Redis for improved performance in your applications.
If you want to learn more about concepts in caching and how to use them in your own projects, check out our blog on the topic.
1. functools.cache for Simple In-Memory Caching¶
When to Use: Ideal for functions with immutable arguments, called repeatedly with the same parameters in small to medium-sized applications. This makes sense when we might be reusing the same data within a single session. or in an application where we don't need to persist the cache between sessions.
import time
import functools
import openai
import instructor
from pydantic import BaseModel
client = instructor.from_openai(openai.OpenAI())
class UserDetail(BaseModel):
name: str
age: int
@functools.cache
def extract(data) -> UserDetail:
return client.chat.completions.create(
model="gpt-3.5-turbo",
response_model=UserDetail,
messages=[
{"role": "user", "content": data},
],
)
start = time.perf_counter() # (1)
model = extract("Extract jason is 25 years old")
print(f"Time taken: {time.perf_counter() - start}")
#> Time taken: 0.38183529197704047
start = time.perf_counter()
model = extract("Extract jason is 25 years old") # (2)
print(f"Time taken: {time.perf_counter() - start}")
#> Time taken: 8.75093974173069e-07
- Using
time.perf_counter()to measure the time taken to run the function is better than usingtime.time()because it's more accurate and less susceptible to system clock changes. - The second time we call
extract, the result is returned from the cache, and the function is not called.
Changing the Model does not Invalidate the Cache
Note that changing the model does not invalidate the cache. This is because the cache key is based on the function's name and arguments, not the model. This means that if we change the model, the cache will still return the old result.
Now we can call extract multiple times with the same argument, and the result will be cached in memory for faster access.
Benefits: Easy to implement, provides fast access due to in-memory storage, and requires no additional libraries.
What is a decorator?
A decorator is a function that takes another function and extends the behavior of the latter function without explicitly modifying it. In Python, decorators are functions that take a function as an argument and return a closure.
def decorator(func):
def wrapper(*args, **kwargs):
print("Do something before") # (1)
#> Do something before
result = func(*args, **kwargs)
print("Do something after") # (2)
#> Do something after
return result
return wrapper
@decorator
def say_hello():
#> Hello!
print("Hello!")
#> Hello!
say_hello()
#> "Do something before"
#> "Hello!"
#> "Do something after"
- The code is executed before the function is called
- The code is executed after the function is called
2. diskcache for Persistent, Large Data Caching¶
Copy Caching Code
We'll be using the same instructor_cache decorator for both diskcache and redis caching. You can copy the code below and use it for both examples.
import functools
import inspect
import diskcache
cache = diskcache.Cache('./my_cache_directory') # (1)
def instructor_cache(func):
"""Cache a function that returns a Pydantic model"""
return_type = inspect.signature(func).return_annotation
if not issubclass(return_type, BaseModel): # (2)
raise ValueError("The return type must be a Pydantic model")
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = f"{func.__name__}-{functools._make_key(args, kwargs, typed=False)}"
# Check if the result is already cached
if (cached := cache.get(key)) is not None:
# Deserialize from JSON based on the return type
return return_type.model_validate_json(cached)
# Call the function and cache its result
result = func(*args, **kwargs)
serialized_result = result.model_dump_json()
cache.set(key, serialized_result)
return result
return wrapper
- We create a new
diskcache.Cacheinstance to store the cached data. This will create a new directory calledmy_cache_directoryin the current working directory. - We only want to cache functions that return a Pydantic model to simplify serialization and deserialization logic in this example code
Remember that you can change this code to support non-Pydantic models, or to use a different caching backend. More over, don't forget that this cache does not invalidate when the model changes, so you might want to encode the Model.model_json_schema() as part of the key.
When to Use: Suitable for applications needing cache persistence between sessions or dealing with large datasets. This is useful when we want to reuse the same data across multiple sessions, or when we need to store large amounts of data!
import functools
import inspect
import instructor
import diskcache
from openai import OpenAI
from pydantic import BaseModel
client = instructor.from_openai(OpenAI())
cache = diskcache.Cache('./my_cache_directory')
def instructor_cache(func):
"""Cache a function that returns a Pydantic model"""
return_type = inspect.signature(func).return_annotation # (4)
if not issubclass(return_type, BaseModel): # (1)
raise ValueError("The return type must be a Pydantic model")
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = (
f"{func.__name__}-{functools._make_key(args, kwargs, typed=False)}" # (2)
)
# Check if the result is already cached
if (cached := cache.get(key)) is not None:
# Deserialize from JSON based on the return type (3)
return return_type.model_validate_json(cached)
# Call the function and cache its result
result = func(*args, **kwargs)
serialized_result = result.model_dump_json()
cache.set(key, serialized_result)
return result
return wrapper
class UserDetail(BaseModel):
name: str
age: int
@instructor_cache
def extract(data) -> UserDetail:
return client.chat.completions.create(
model="gpt-3.5-turbo",
response_model=UserDetail,
messages=[
{"role": "user", "content": data},
],
)
- We only want to cache functions that return a Pydantic model to simplify serialization and deserialization logic
- We use functool's
_make_keyto generate a unique key based on the function's name and arguments. This is important because we want to cache the result of each function call separately. - We use Pydantic's
model_validate_jsonto deserialize the cached result into a Pydantic model. - We use
inspect.signatureto get the function's return type annotation, which we use to validate the cached result.
Benefits: Reduces computation time for heavy data processing, provides disk-based caching for persistence.
2. Redis Caching Decorator for Distributed Systems¶
Copy Caching Code
We'll be using the same instructor_cache decorator for both diskcache and redis caching. You can copy the code below and use it for both examples.
import functools
import inspect
import redis
cache = redis.Redis("localhost")
def instructor_cache(func):
"""Cache a function that returns a Pydantic model"""
return_type = inspect.signature(func).return_annotation
if not issubclass(return_type, BaseModel):
raise ValueError("The return type must be a Pydantic model")
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = f"{func.__name__}-{functools._make_key(args, kwargs, typed=False)}"
# Check if the result is already cached
if (cached := cache.get(key)) is not None:
# Deserialize from JSON based on the return type
return return_type.model_validate_json(cached)
# Call the function and cache its result
result = func(*args, **kwargs)
serialized_result = result.model_dump_json()
cache.set(key, serialized_result)
return result
return wrapper
Remember that you can change this code to support non-Pydantic models, or to use a different caching backend. More over, don't forget that this cache does not invalidate when the model changes, so you might want to encode the Model.model_json_schema() as part of the key.
When to Use: Recommended for distributed systems where multiple processes need to access the cached data, or for applications requiring fast read/write access and handling complex data structures.
import redis
import functools
import inspect
import instructor
from pydantic import BaseModel
from openai import OpenAI
client = instructor.from_openai(OpenAI())
cache = redis.Redis("localhost")
def instructor_cache(func):
"""Cache a function that returns a Pydantic model"""
return_type = inspect.signature(func).return_annotation
if not issubclass(return_type, BaseModel): # (1)
raise ValueError("The return type must be a Pydantic model")
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = f"{func.__name__}-{functools._make_key(args, kwargs, typed=False)}" # (2)
# Check if the result is already cached
if (cached := cache.get(key)) is not None:
# Deserialize from JSON based on the return type
return return_type.model_validate_json(cached)
# Call the function and cache its result
result = func(*args, **kwargs)
serialized_result = result.model_dump_json()
cache.set(key, serialized_result)
return result
return wrapper
class UserDetail(BaseModel):
name: str
age: int
@instructor_cache
def extract(data) -> UserDetail:
# Assuming client.chat.completions.create returns a UserDetail instance
return client.chat.completions.create(
model="gpt-3.5-turbo",
response_model=UserDetail,
messages=[
{"role": "user", "content": data},
],
)
- We only want to cache functions that return a Pydantic model to simplify serialization and deserialization logic
- We use functool's
_make_keyto generate a unique key based on the function's name and arguments. This is important because we want to cache the result of each function call separately.
Benefits: Scalable for large-scale systems, supports fast in-memory data storage and retrieval, and is versatile for various data types.
Looking carefully
If you look carefully at the code above you'll notice that we're using the same instructor_cache decorator as before. The implementation is the same, but we're using a different caching backend!