camel.storages.key_value_storages package

Submodules

camel.storages.key_value_storages.base module

class camel.storages.key_value_storages.base.BaseKeyValueStorage

Bases: ABC

An abstract base class for key-value storage systems. Provides a consistent interface for saving, loading, and clearing data records without any loss of information.

An abstract base class designed to serve as a foundation for various key-value storage systems. The class primarily interacts through Python dictionaries.

This class is meant to be inherited by multiple types of key-value storage implementations, including, but not limited to, JSON file storage, NoSQL databases like MongoDB and Redis, as well as in-memory Python dictionaries.

abstract clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

camel.storages.key_value_storages.in_memory module

class camel.storages.key_value_storages.in_memory.InMemoryKeyValueStorage

Bases: BaseKeyValueStorage

A concrete implementation of the BaseKeyValueStorage using in-memory list. Ideal for temporary storage purposes, as data will be lost when the program ends.

clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

camel.storages.key_value_storages.json module

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)

Bases: JSONEncoder

A custom JSON encoder for serializing specifically enumerated types. Ensures enumerated types can be stored in and retrieved from JSON format.

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

default(obj) → Any

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

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)

Bases: BaseKeyValueStorage

A concrete implementation of the BaseKeyValueStorage using JSON files. Allows for persistent storage of records in a human-readable format.

Parameters:

path (Path, optional) – Path to the desired JSON file. If None, a default path ./chat_history.json will be used. (default: None)

clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

camel.storages.key_value_storages.redis module

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

Bases: BaseKeyValueStorage

A concrete implementation of the BaseCacheStorage using Redis as the backend. This is suitable for distributed cache systems that require persistence and high availability.

clear() → None

Removes all records from the key-value storage system.

property client: Redis | None

Returns the Redis client instance.

Returns:

The Redis client instance.

Return type:

redis.asyncio.Redis

async close() → None

Closes the Redis client asynchronously.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Module contents

class camel.storages.key_value_storages.BaseKeyValueStorage

Bases: ABC

An abstract base class for key-value storage systems. Provides a consistent interface for saving, loading, and clearing data records without any loss of information.

An abstract base class designed to serve as a foundation for various key-value storage systems. The class primarily interacts through Python dictionaries.

This class is meant to be inherited by multiple types of key-value storage implementations, including, but not limited to, JSON file storage, NoSQL databases like MongoDB and Redis, as well as in-memory Python dictionaries.

abstract clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

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)

Bases: JSONEncoder

A custom JSON encoder for serializing specifically enumerated types. Ensures enumerated types can be stored in and retrieved from JSON format.

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

default(obj) → Any

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

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

Bases: BaseKeyValueStorage

A concrete implementation of the BaseKeyValueStorage using in-memory list. Ideal for temporary storage purposes, as data will be lost when the program ends.

clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

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

Bases: BaseKeyValueStorage

A concrete implementation of the BaseKeyValueStorage using JSON files. Allows for persistent storage of records in a human-readable format.

Parameters:

path (Path, optional) – Path to the desired JSON file. If None, a default path ./chat_history.json will be used. (default: None)

clear() → None

Removes all records from the key-value storage system.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

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)

Bases: BaseKeyValueStorage

A concrete implementation of the BaseKeyValueStorage using Mem0 as the backend. This storage system uses Mem0’s text capabilities to store, search, and manage text with context.

Parameters:

• agent_id (str) – Default agent ID to associate memories with.

• api_key (str, optional) – The API key for authentication. If not provided, will try to get from environment variable MEM0_API_KEY (default: None).

• user_id (str, optional) – Default user ID to associate memories with (default: None).

• metadata (Dict[str, Any], optional) – Default metadata to include with all memories (default: None).

References

https://docs.mem0.ai

clear() → None

Removes all records from the Mem0 storage system.

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

Loads all stored records from the Mem0 storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the Mem0 storage system.

Parameters:

records (List[Dict[str, Any]]) – A list of dictionaries, where each dictionary represents a unique record to be stored.

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

Bases: BaseKeyValueStorage

A concrete implementation of the BaseCacheStorage using Redis as the backend. This is suitable for distributed cache systems that require persistence and high availability.

clear() → None

Removes all records from the key-value storage system.

property client: Redis | None

Returns the Redis client instance.

Returns:

The Redis client instance.

Return type:

redis.asyncio.Redis

async close() → None

Closes the Redis client asynchronously.

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

Loads all stored records from the key-value storage system.

Returns:

A list of dictionaries, where each dictionary

represents a stored record.

Return type:

List[Dict[str, Any]]

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

Saves a batch of records to the key-value storage system.