缓存

标题: Python中的缓存技术：内存、磁盘和Redis 描述: 探索Python中使用functools、diskcache和Redis的缓存方法，以提高应用程序的性能。

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

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

使用 time.perf_counter() 来测量运行函数所需的时间优于使用 time.time()，因为它更准确且不易受到系统时钟变化的影响。
第二次我们调用 extract 时，结果是从缓存中返回的，函数没有被调用。

更改模型不会使缓存失效

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

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

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

What is a decorator?

装饰器是一个函数，它接受另一个函数并扩展后者的行为，而无需显式修改它。在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` 用于持久化、大数据缓存¶

Copy Caching Code

我们将使用相同的instructor_cache装饰器来处理diskcache和redis缓存。你可以复制下面的代码并在两个示例中使用它。

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缓存装饰器¶

Copy Caching Code

我们将使用相同的instructor_cache装饰器来处理diskcache和redis缓存。你可以复制下面的代码并在两个示例中使用它。

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装饰器。实现是相同的，但我们使用了不同的缓存后端！

缓存

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

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

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

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

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