autogen_ext.models.cache

class ChatCompletionCache(client: ChatCompletionClient, store: CacheStore[CreateResult | List[str | CreateResult]] | None = None)

基类：ChatCompletionClient, Component[ChatCompletionCacheConfig]

对 ChatCompletionClient 的包装器，用于缓存底层客户端的创建结果。 缓存命中不会计入原始客户端的令牌使用量。

典型用法：

以使用 openai 客户端进行磁盘缓存为例。 首先安装带有必要包的 autogen-ext：

pip install -U "autogen-ext[openai, diskcache]"

并按如下方式使用：

import asyncio
import tempfile

from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.models.cache import ChatCompletionCache, CHAT_CACHE_VALUE_TYPE
from autogen_ext.cache_store.diskcache import DiskCacheStore
from diskcache import Cache


async def main():
    with tempfile.TemporaryDirectory() as tmpdirname:
        # 初始化原始客户端
        openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")

        # 然后初始化 CacheStore，这里使用 diskcache.Cache。
        # 也可以使用 redis 例如：
        # from autogen_ext.cache_store.redis import RedisStore
        # import redis
        # redis_instance = redis.Redis()
        # cache_store = RedisCacheStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
        cache_store = DiskCacheStore[CHAT_CACHE_VALUE_TYPE](Cache(tmpdirname))
        cache_client = ChatCompletionCache(openai_model_client, cache_store)

        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # 应打印来自 OpenAI 的响应
        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # 应打印缓存的响应


asyncio.run(main())

现在可以像使用原始客户端一样使用 cached_client，但启用了缓存功能。

参数:

• client (ChatCompletionClient) -- 要包装的原始 ChatCompletionClient。

• store (CacheStore) -- 实现 get 和 set 方法的存储对象。 用户需负责管理存储的生命周期及清理（如果需要）。 默认使用内存缓存。

classmethod _from_config(config: ChatCompletionCacheConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) -- 配置对象。

Returns:

Self -- 组件的新实例。

_to_config() → ChatCompletionCacheConfig

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:

T -- 组件的配置。

actual_usage() → RequestUsage

property capabilities: ModelCapabilities

async close() → None

component_config_schema

ChatCompletionCacheConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_ext.models.cache.ChatCompletionCache'

覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

component_type: ClassVar[ComponentType] = 'chat_completion_cache'

组件的逻辑类型。

count_tokens(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = []) → int

async create(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = [], json_output: bool | type[BaseModel] | None = None, extra_create_args: Mapping[str, Any] = {}, cancellation_token: CancellationToken | None = None) → CreateResult

ChatCompletionClient.create 的缓存版本。 如果 create 调用的结果已被缓存，将立即返回缓存结果 而不会调用底层客户端。

注意：对于缓存结果，cancellation_token 将被忽略。

create_stream(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = [], json_output: bool | type[BaseModel] | None = None, extra_create_args: Mapping[str, Any] = {}, cancellation_token: CancellationToken | None = None) → AsyncGenerator[str | CreateResult, None]

ChatCompletionClient.create_stream 的缓存版本。 如果调用 create_stream 的结果已被缓存，将直接返回缓存结果 而不会从底层客户端进行流式传输。

注意：对于缓存结果，cancellation_token 将被忽略。

property model_info: ModelInfo

remaining_tokens(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = []) → int

total_usage() → RequestUsage