camel.storages.key_value_storages 包

子模块

camel.storages.key_value_storages.base 模块

class camel.storages.key_value_storages.base.BaseKeyValueStorage

基类: ABC

一个用于键值存储系统的抽象基类。提供保存、加载和清除数据记录的一致接口，且不会丢失任何信息。

一个抽象基类，旨在作为各种键值存储系统的基础。该类主要通过Python字典进行交互。

这个类旨在被多种键值存储实现继承，包括但不限于JSON文件存储、NoSQL数据库（如MongoDB和Redis）以及内存中的Python字典。

abstract clear() → None

从键值存储系统中移除所有记录。

abstract load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

abstract save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

camel.storages.key_value_storages.in_memory 模块

class camel.storages.key_value_storages.in_memory.InMemoryKeyValueStorage

基类: BaseKeyValueStorage

BaseKeyValueStorage的具体实现，使用内存中的列表。非常适合临时存储用途，因为当程序结束时数据会丢失。

clear() → None

从键值存储系统中移除所有记录。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

camel.storages.key_value_storages.json 模块

class camel.storages.key_value_storages.json.CamelJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

基类: JSONEncoder

一个用于序列化特定枚举类型的自定义JSON编码器。 确保枚举类型可以存储到JSON格式中并从JSON格式中检索。

CAMEL_ENUMS: ClassVar[Dict[str, EnumMeta]] = {'ModelType': <enum 'ModelType'>, 'OpenAIBackendRole': <enum 'OpenAIBackendRole'>, 'RoleType': <enum 'RoleType'>, 'TaskType': <enum 'TaskType'>}

default(obj) → Any

在子类中实现此方法，使其返回o的可序列化对象，或调用基类实现（以引发TypeError）。

例如，为了支持任意迭代器，您可以这样实现默认值：

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class camel.storages.key_value_storages.json.JsonStorage(path: Path | None = None)

基类: BaseKeyValueStorage

一个基于JSON文件的具体BaseKeyValueStorage实现。允许以人类可读的格式持久化存储记录。

Parameters:

path (Path, 可选) – 目标JSON文件的路径。如果为None， 将使用默认路径./chat_history.json。 (默认值: None)

clear() → None

从键值存储系统中移除所有记录。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

camel.storages.key_value_storages.redis 模块

class camel.storages.key_value_storages.redis.RedisStorage(sid: str, url: str = 'redis://localhost:6379', loop: AbstractEventLoop | None = None, **kwargs)

基类: BaseKeyValueStorage

一个基于Redis后端的BaseCacheStorage具体实现。这适用于需要持久化和高可用性的分布式缓存系统。

clear() → None

从键值存储系统中移除所有记录。

property client: Redis | None

返回Redis客户端实例。

Returns:

Redis客户端实例。

Return type:

redis.asyncio.Redis

async close() → None

异步关闭Redis客户端。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]], expire: int | None = None) → None

将一批记录保存到键值存储系统中。

模块内容

class camel.storages.key_value_storages.BaseKeyValueStorage

基类: ABC

一个用于键值存储系统的抽象基类。提供保存、加载和清除数据记录的一致接口，且不会丢失任何信息。

一个抽象基类，旨在作为各种键值存储系统的基础。该类主要通过Python字典进行交互。

这个类旨在被多种键值存储实现继承，包括但不限于JSON文件存储、NoSQL数据库（如MongoDB和Redis）以及内存中的Python字典。

abstract clear() → None

从键值存储系统中移除所有记录。

abstract load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

abstract save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

class camel.storages.key_value_storages.CamelJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

基类: JSONEncoder

一个用于序列化特定枚举类型的自定义JSON编码器。 确保枚举类型可以存储到JSON格式中并从JSON格式中检索。

CAMEL_ENUMS: ClassVar[Dict[str, EnumMeta]] = {'ModelType': <enum 'ModelType'>, 'OpenAIBackendRole': <enum 'OpenAIBackendRole'>, 'RoleType': <enum 'RoleType'>, 'TaskType': <enum 'TaskType'>}

default(obj) → Any

在子类中实现此方法，使其返回o的可序列化对象，或调用基类实现（以引发TypeError）。

例如，为了支持任意迭代器，您可以这样实现默认值：

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class camel.storages.key_value_storages.InMemoryKeyValueStorage

基类: BaseKeyValueStorage

BaseKeyValueStorage的具体实现，使用内存中的列表。非常适合临时存储用途，因为当程序结束时数据将会丢失。

clear() → None

从键值存储系统中移除所有记录。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

class camel.storages.key_value_storages.JsonStorage(path: Path | None = None)

基类: BaseKeyValueStorage

BaseKeyValueStorage的具体实现，使用JSON文件。允许以人类可读的格式持久化存储记录。

Parameters:

path (Path, optional) – 目标JSON文件的路径。如果为None， 将使用默认路径./chat_history.json。 (默认值: None)

clear() → None

从键值存储系统中移除所有记录。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]]) → None

将一批记录保存到键值存储系统中。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

class camel.storages.key_value_storages.Mem0Storage(agent_id: str, api_key: str | None = None, user_id: str | None = None, metadata: Dict[str, Any] | None = None)

基类: BaseKeyValueStorage

BaseKeyValueStorage的具体实现，使用Mem0作为后端。该存储系统利用Mem0的文本功能来存储、搜索和管理带上下文的文本。

Parameters:

• agent_id (str) - 默认关联记忆的代理ID。

• api_key (str, optional) – 用于身份验证的API密钥。如果未提供，将尝试从环境变量MEM0_API_KEY获取(默认: None)。

• user_id (str, optional) – 默认关联记忆的用户ID (默认值: None).

• metadata (Dict[str, Any], optional) - 默认包含在所有记忆中的元数据（默认：None）。

参考文献

https://docs.mem0.ai

clear() → None

从Mem0存储系统中移除所有记录。

load() → List[Dict[str, Any]]

从Mem0存储系统加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]]) → None

将一批记录保存到Mem0存储系统。

Parameters:

records (List[Dict[str, Any]]) – 一个字典列表，其中每个字典代表要存储的唯一记录。

class camel.storages.key_value_storages.RedisStorage(sid: str, url: str = 'redis://localhost:6379', loop: AbstractEventLoop | None = None, **kwargs)

基类: BaseKeyValueStorage

一个基于Redis后端的BaseCacheStorage具体实现。这适用于需要持久化和高可用性的分布式缓存系统。

clear() → None

从键值存储系统中移除所有记录。

property client: Redis | None

返回Redis客户端实例。

Returns:

Redis客户端实例。

Return type:

redis.asyncio.Redis

async close() → None

异步关闭Redis客户端。

load() → List[Dict[str, Any]]

从键值存储系统中加载所有存储的记录。

Returns:

一个字典列表，其中每个字典

代表一条存储的记录。

Return type:

List[Dict[str, Any]]

save(records: List[Dict[str, Any]], expire: int | None = None) → None

将一批记录保存到键值存储系统中。