Python 中的缓存介绍

Instructor 使语言模型的使用变得简单，但它们在计算上仍然是昂贵的。

今天，我们将深入探讨如何优化 instructor 代码，同时保持由 Pydantic 模型提供的出色开发者体验 (DX)。我们将解决缓存 Pydantic 模型（通常与 pickle 不兼容）的挑战，并探索使用 装饰器（如 functools.cache）的解决方案。然后，我们将使用 diskcache 和 redis 构建自定义装饰器，以支持持久化缓存和分布式系统。

让我们首先考虑一个典型示例，使用 OpenAI Python 客户端来提取用户详细信息。

import instructor
from openai import OpenAI
from pydantic import BaseModel

# Enables `response_model`
client = instructor.from_openai(OpenAI())


class UserDetail(BaseModel):
    name: str
    age: int


def extract(data) -> UserDetail:
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        response_model=UserDetail,
        messages=[
            {"role": "user", "content": data},
        ],
    )

现在想象一下批量处理数据、运行测试或实验，或者只是在一个工作流程中多次调用 extract 函数。我们会很快遇到性能问题，因为函数可能会被重复调用，并且相同的数据会被一遍又一遍地处理，这会浪费我们的时间和金钱。

1. 使用 functools.cache 进行简单的内存缓存

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

import functools


@functools.cache
def extract(data):
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        response_model=UserDetail,
        messages=[
            {"role": "user", "content": data},
        ],
    )

更改模型不会使缓存失效

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

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

import time

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

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

#> Time taken: 0.92
#> Time taken: 1.20e-06 # (3)

1. 使用 time.perf_counter() 来测量函数运行所需的时间比使用 time.time() 更好，因为它更精确，并且受系统时钟变化的影响更小。
2. 第二次调用 extract 函数时，结果从缓存中返回，函数并未被调用。
3. 第二次调用 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!")


say_hello()
#> "Do something before"
#> "Hello!"
#> "Do something after"

1. 代码在函数调用之前执行
2. 代码在函数调用之后执行

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

1. 我们创建一个新的 diskcache.Cache 实例来存储缓存数据。这将在当前工作目录中创建一个名为 my_cache_directory 的新目录。
2. 在此示例代码中，我们只想缓存返回 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},
        ],
    )

1. 我们只想缓存返回 Pydantic 模型的函数，以简化序列化和反序列化逻辑
2. 我们使用 functool 的 _make_key 根据函数名和参数生成唯一的键。这很重要，因为我们想分别缓存每个函数调用的结果。
3. 我们使用 Pydantic 的 model_validate_json 将缓存结果反序列化为 Pydantic 模型。
4. 我们使用 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},
        ],
    )

1. 我们只想缓存返回 Pydantic 模型的函数，以简化序列化和反序列化逻辑
2. 我们使用 functool 的 _make_key 根据函数名和参数生成唯一的键。这很重要，因为我们想分别缓存每个函数调用的结果。

优点：适用于大型系统的可伸缩性，支持快速内存数据存储和检索，并且对各种数据类型具有通用性。

仔细查看

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

结论

选择合适的缓存策略取决于您应用的具体需求，例如数据的大小和类型、持久化的需求以及系统的架构。无论是优化小型应用中的函数性能，还是在分布式环境中管理大型数据集，Python 都提供了强大的解决方案来提高效率并减少计算开销。

如果您想使用这段代码，可以尝试将其发送给 ChatGPT 以更深入地理解它，并添加对您可能重要的附加功能，例如，当您的 BaseModel 更改时，缓存不会失效，因此您可能希望将 Model.model_json_schema() 作为键的一部分进行编码。

如果您喜欢这些内容，请访问我们的 GitHub 并给我们点赞，同时了解我们的库。