缓存

title: Python 中的缓存技术：内存、磁盘和 Redis description: 探索使用 functools、diskcache 和 Redis 在 Python 中实现缓存的方法，以提高应用程序性能。

如果您想了解更多关于缓存的概念以及如何在您自己的项目中使用它们，请查看我们关于此主题的博客。

1. functools.cache 用于简单的内存缓存¶

何时使用：适用于参数不可变、在中小型应用程序中用相同参数重复调用的函数。这在我们可能在单个会话中重用相同数据时，或在不需要在会话之间持久化缓存的应用程序中非常有用。

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.5008833750034682

start = time.perf_counter()
model = extract("Extract jason is 25 years old")  # (2)
print(f"Time taken: {time.perf_counter() - start}")
#> Time taken: 1.2920063454657793e-06

使用 time.perf_counter() 来测量函数运行所需的时间比使用 time.time() 更好，因为它更准确且不易受系统时钟变化的影响。
第二次调用 extract 时，结果将从缓存中返回，并且不再调用该函数。

更改模型不会使缓存失效

请注意，更改模型不会使缓存失效。这是因为缓存键基于函数的名称和参数，而不是模型。这意味着如果更改模型，缓存仍将返回旧结果。

现在我们可以使用相同的参数多次调用 extract，结果将缓存到内存中以加快访问速度。

优点：易于实现，由于内存存储提供快速访问，并且不需要额外的库。

什么是装饰器？

装饰器是接收另一个函数并扩展该函数行为而不显式修改它的函数。在 Python 中，装饰器是接收函数作为参数并返回闭包的函数。

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"

代码在函数被调用之前执行
代码在函数被调用之后执行

2. diskcache 用于持久化、大数据缓存¶

复制缓存代码

我们将对 diskcache 和 redis 缓存使用相同的 instructor_cache 装饰器。您可以复制下面的代码并将其用于这两个示例。

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

我们创建一个新的 diskcache.Cache 实例来存储缓存数据。这将在当前工作目录中创建一个名为 my_cache_directory 的新目录。
在此示例代码中，我们只希望缓存返回 Pydantic 模型的函数，以简化序列化和反序列化逻辑

请记住，您可以更改此代码以支持非 Pydantic 模型，或使用不同的缓存后端。此外，不要忘记此缓存不会在模型更改时失效，因此您可能希望将 Model.model_json_schema() 作为键的一部分进行编码。

何时使用：适用于需要在会话之间持久化缓存或处理大型数据集的应用程序。这在我们想要跨多个会话重用相同数据，或者需要存储大量数据时非常有用！

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},
        ],
    )

我们只希望缓存返回 Pydantic 模型的函数，以简化序列化和反序列化逻辑
我们使用 functool 的 _make_key 根据函数名称和参数生成唯一键。这很重要，因为我们希望单独缓存每次函数调用的结果。
我们使用 Pydantic 的 model_validate_json 将缓存结果反序列化为 Pydantic 模型。
我们使用 inspect.signature 获取函数的返回类型注解，并用它来验证缓存结果。

优点：减少大数据处理的计算时间，提供基于磁盘的缓存以实现持久化。

2. Redis 缓存装饰器用于分布式系统¶

复制缓存代码

我们将对 diskcache 和 redis 缓存使用相同的 instructor_cache 装饰器。您可以复制下面的代码并将其用于这两个示例。

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

请记住，您可以更改此代码以支持非 Pydantic 模型，或使用不同的缓存后端。此外，不要忘记此缓存不会在模型更改时失效，因此您可能希望将 Model.model_json_schema() 作为键的一部分进行编码。

何时使用：推荐用于需要多个进程访问缓存数据的分布式系统，或需要快速读写访问和处理复杂数据结构的应用程序。

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},
        ],
    )

我们只希望缓存返回 Pydantic 模型的函数，以简化序列化和反序列化逻辑
我们使用 functool 的 _make_key 根据函数名称和参数生成唯一键。这很重要，因为我们希望单独缓存每次函数调用的结果。

优点：可扩展到大型系统，支持快速内存数据存储和检索，并且适用于各种数据类型。

仔细查看

如果您仔细查看上面的代码，您会注意到我们使用的是与之前相同的 instructor_cache 装饰器。实现是相同的，但我们使用的是不同的缓存后端！