Autogen_core源码：_cache_store.py

_cache_store.py代码

from typing import Dict, Generic, Optional, Protocol, TypeVar

T = TypeVar("T")


class CacheStore(Protocol, Generic[T]):
    """
    This protocol defines the basic interface for store/cache operations.

    Sub-classes should handle the lifecycle of underlying storage.
    """

    def get(self, key: str, default: Optional[T] = None) -> Optional[T]:
        """
        Retrieve an item from the store.

        Args:
            key: The key identifying the item in the store.
            default (optional): The default value to return if the key is not found.
                                Defaults to None.

        Returns:
            The value associated with the key if found, else the default value.
        """
        ...

    def set(self, key: str, value: T) -> None:
        """
        Set an item in the store.

        Args:
            key: The key under which the item is to be stored.
            value: The value to be stored in the store.
        """
        ...


class InMemoryStore(CacheStore[T]):
    def __init__(self) -> None:
        self.store: Dict[str, T] = {}

    def get(self, key: str, default: Optional[T] = None) -> Optional[T]:
        return self.store.get(key, default)

    def set(self, key: str, value: T) -> None:
        self.store[key] = value

代码解释

这段Python代码定义了一个缓存存储的接口协议 CacheStore，并实现了一个基于内存的缓存存储类 InMemoryStore。以下是对代码逻辑和功能的详细解释：

代码逻辑

1. 类型变量定义

T = TypeVar("T")

• TypeVar 是Python typing 模块中的一个工具，用于定义泛型类型变量。这里定义了一个名为 T 的类型变量，它可以代表任意类型。在后续的代码中，T 可以用来表示不同类型的数据，从而实现代码的泛型化。

2. 缓存存储接口协议 CacheStore

class CacheStore(Protocol, Generic[T]):
    """
    This protocol defines the basic interface for store/cache operations.

    Sub-classes should handle the lifecycle of underlying storage.
    """

    def get(self, key: str, default: Optional[T] = None) -> Optional[T]:
        """
        Retrieve an item from the store.

        Args:
            key: The key identifying the item in the store.
            default (optional): The default value to return if the key is not found.
                                Defaults to None.

        Returns:
            The value associated with the key if found, else the default value.
        """
       ...

    def set(self, key: str, value: T) -> None:
        """
        Set an item in the store.

        Args:
            key: The key under which the item is to be stored.
            value: The value to be stored in the store.
        """
       ...

• Protocol 是Python 3.8及以上版本引入的一种特殊类，用于定义协议。协议是一种抽象的接口，它只定义了方法的签名，而不包含具体的实现。
• Generic[T] 表示 CacheStore 是一个泛型类，它可以处理任意类型 T 的数据。
• get 方法用于从存储中检索一个项目。它接受一个字符串类型的 key 和一个可选的默认值 default，如果指定的 key 存在于存储中，则返回与之关联的值；否则返回 default。
• set 方法用于将一个项目存储到存储中。它接受一个字符串类型的 key 和一个类型为 T 的 value，并将 value 存储在 key 下。

3. 基于内存的缓存存储类 InMemoryStore

class InMemoryStore(CacheStore[T]):
    def __init__(self) -> None:
        self.store: Dict[str, T] = {}

    def get(self, key: str, default: Optional[T] = None) -> Optional[T]:
        return self.store.get(key, default)

    def set(self, key: str, value: T) -> None:
        self.store[key] = value

• InMemoryStore 类继承自 CacheStore[T]，表示它实现了 CacheStore 协议。
• __init__ 方法初始化一个空的字典 self.store，用于存储键值对。
• get 方法调用字典的 get 方法，从 self.store 中检索指定 key 的值。如果 key 不存在，则返回 default。
• set 方法将指定的 key 和 value 存储到 self.store 中。

功能总结

这段代码的主要功能是定义了一个通用的缓存存储接口协议 CacheStore，并实现了一个基于内存的缓存存储类 InMemoryStore。CacheStore 协议为缓存存储操作提供了统一的接口，使得不同的存储实现可以遵循相同的规范。InMemoryStore 类使用Python的字典作为底层存储，提供了简单的内存缓存功能，支持通过 get 方法检索数据和通过 set 方法存储数据。这种设计模式使得代码具有良好的可扩展性，可以方便地实现其他类型的缓存存储，如基于文件、数据库等。

代码示例

示例 1：使用 InMemoryStore 存储和获取整数类型数据

from typing import Optional
from autogen_core import InMemoryStore
# 定义 CacheStore 协议和 InMemoryStore 类（此处省略，假设已定义）

# 创建一个存储整数的 InMemoryStore 实例
int_store = InMemoryStore[int]()

# 设置一个键值对
int_store.set("number", 42)

# 获取存储的值
result: Optional[int] = int_store.get("number")
print(f"Stored number: {result}")

# 获取不存在的键的值
default_result: Optional[int] = int_store.get("nonexistent", 0)
print(f"Default result: {default_result}")

Stored number: 42
Default result: 0

示例 2：使用 InMemoryStore 存储和获取字符串类型数据

str_store = InMemoryStore[str]()

# 设置一个键值对
str_store.set("message", "Hello, World!")

# 获取存储的值
result: Optional[str] = str_store.get("message")
print(f"Stored message: {result}")

# 获取不存在的键的值
default_result: Optional[str] = str_store.get("nonexistent", "Default message")
print(f"Default result: {default_result}")

Stored message: Hello, World!
Default result: Default message

示例 3：使用 InMemoryStore 存储和获取自定义对象类型数据

from typing import Optional

# 定义 CacheStore 协议和 InMemoryStore 类（此处省略，假设已定义）

# 定义一个自定义类
class Person:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

    def __str__(self):
        return f"Person(name={self.name}, age={self.age})"

# 创建一个存储 Person 对象的 InMemoryStore 实例
person_store = InMemoryStore[Person]()

# 创建一个 Person 对象
person = Person("Alice", 30)

# 设置一个键值对
person_store.set("person", person)

# 获取存储的值
result: Optional[Person] = person_store.get("person")
if result:
    print(f"Stored person: {result}")

# 获取不存在的键的值
default_result: Optional[Person] = person_store.get("nonexistent")
print(f"Default result: {default_result}")

Stored person: Person(name=Alice, age=30)
Default result: None