💾 存储模块

evoagentx.storages

BaseModule

BaseModule(**kwargs)

Bases: BaseModel

Base module class that serves as the foundation for all modules in the EvoAgentX framework.

This class provides serialization/deserialization capabilities, supports creating instances from dictionaries, JSON, or files, and exporting instances to these formats.

Attributes:

Name	Type	Description
class_name	str	The class name, defaults to None but is automatically set during subclass initialization
model_config		Pydantic model configuration that controls type matching and behavior

Initializes a BaseModule instance.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments used to initialize the instance	{}

Raises:

Type	Description
ValidationError	When parameter validation fails
Exception	When other errors occur during initialization

Source code in evoagentx/core/module.py

def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

kwargs property

kwargs: dict

Returns the extra fields of the model.

Returns:

Name	Type	Description
dict	dict	Dictionary containing all extra keyword arguments

__init_subclass__

__init_subclass__(**kwargs)

Subclass initialization method that automatically sets the class_name attribute.

Parameters:

Name	Type	Description	Default
cls	Type	The subclass being initialized	required
**kwargs	Any	Additional keyword arguments	{}

Source code in evoagentx/core/module.py

def __init_subclass__(cls, **kwargs):
    """
    Subclass initialization method that automatically sets the class_name attribute.

    Args:
        cls (Type): The subclass being initialized
        **kwargs (Any): Additional keyword arguments
    """
    super().__init_subclass__(**kwargs)
    cls.class_name = cls.__name__

init_module

init_module()

Module initialization method that subclasses can override to provide additional initialization logic.

Source code in evoagentx/core/module.py

def init_module(self):
    """
    Module initialization method that subclasses can override to provide additional initialization logic.
    """
    pass

__str__

__str__() -> str

Returns a string representation of the object.

Returns:

Name	Type	Description
str	str	String representation of the object

Source code in evoagentx/core/module.py

def __str__(self) -> str:
    """
    Returns a string representation of the object.

    Returns:
        str: String representation of the object
    """
    return self.to_str()

from_dict classmethod

from_dict(data: Dict[str, Any], **kwargs) -> BaseModule

Instantiate the BaseModule from a dictionary.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary containing instance data	required
**kwargs	Any	Additional keyword arguments, can include log to control logging output	{}

Returns:

Name	Type	Description
BaseModule	BaseModule	The created module instance

Raises:

Type	Description
Exception	When errors occur during initialization

Source code in evoagentx/core/module.py

@classmethod
def from_dict(cls, data: Dict[str, Any], **kwargs) -> "BaseModule":
    """
    Instantiate the BaseModule from a dictionary.

    Args:
        data: Dictionary containing instance data
        **kwargs (Any): Additional keyword arguments, can include log to control logging output

    Returns:
        BaseModule: The created module instance

    Raises:
        Exception: When errors occur during initialization
    """
    use_logger = kwargs.get("log", True)
    with exception_buffer() as buffer:
        try:
            class_name = data.get("class_name", None)
            if class_name:
                cls = MODULE_REGISTRY.get_module(class_name)
            module = cls._create_instance(data)
            # module = cls.model_validate(data)
            if len(buffer.exceptions) > 0:
                error_message = get_base_module_init_error_message(cls, data, buffer.exceptions)
                if use_logger:
                    logger.error(error_message)
                raise Exception(get_error_message(buffer.exceptions))
        finally:
            pass
    return module

from_json classmethod

from_json(content: str, **kwargs) -> BaseModule

Construct the BaseModule from a JSON string.

This method uses yaml.safe_load to parse the JSON string into a Python object, which supports more flexible parsing than standard json.loads (including handling single quotes, trailing commas, etc). The parsed data is then passed to from_dict to create the instance.

Parameters:

Name	Type	Description	Default
content	str	JSON string	required
**kwargs	Any	Additional keyword arguments, can include log to control logging output	{}

Returns:

Name	Type	Description
BaseModule	BaseModule	The created module instance

Raises:

Type	Description
ValueError	When the input is not a valid JSON string

Source code in evoagentx/core/module.py

@classmethod
def from_json(cls, content: str, **kwargs) -> "BaseModule":
    """
    Construct the BaseModule from a JSON string.

    This method uses yaml.safe_load to parse the JSON string into a Python object,
    which supports more flexible parsing than standard json.loads (including handling
    single quotes, trailing commas, etc). The parsed data is then passed to from_dict
    to create the instance.

    Args:
        content: JSON string
        **kwargs (Any): Additional keyword arguments, can include `log` to control logging output

    Returns:
        BaseModule: The created module instance

    Raises:
        ValueError: When the input is not a valid JSON string
    """
    use_logger = kwargs.get("log", True)
    try:
        data = yaml.safe_load(content)
    except Exception:
        error_message = f"Can not instantiate {cls.__name__}. The input to {cls.__name__}.from_json is not a valid JSON string."
        if use_logger:
            logger.error(error_message)
        raise ValueError(error_message)

    if not isinstance(data, (list, dict)):
        error_message = f"Can not instantiate {cls.__name__}. The input to {cls.__name__}.from_json is not a valid JSON string."
        if use_logger:
            logger.error(error_message)
        raise ValueError(error_message)

    return cls.from_dict(data, log=use_logger)

from_str classmethod

from_str(content: str, **kwargs) -> BaseModule

Construct the BaseModule from a string that may contain JSON.

This method is more forgiving than from_json as it can extract valid JSON objects embedded within larger text. It uses parse_json_from_text to extract all potential JSON strings from the input text, then tries to create an instance from each extracted JSON string until successful.

Parameters:

Name	Type	Description	Default
content	str	Text that may contain JSON strings	required
**kwargs	Any	Additional keyword arguments, can include log to control logging output	{}

Returns:

Name	Type	Description
BaseModule	BaseModule	The created module instance

Raises:

Type	Description
ValueError	When the input does not contain valid JSON strings or the JSON is incompatible with the class

Source code in evoagentx/core/module.py

@classmethod
def from_str(cls, content: str, **kwargs) -> "BaseModule":
    """
    Construct the BaseModule from a string that may contain JSON.

    This method is more forgiving than `from_json` as it can extract valid JSON
    objects embedded within larger text. It uses `parse_json_from_text` to extract 
    all potential JSON strings from the input text, then tries to create an instance 
    from each extracted JSON string until successful.

    Args:
        content: Text that may contain JSON strings
        **kwargs (Any): Additional keyword arguments, can include `log` to control logging output

    Returns:
        BaseModule: The created module instance

    Raises:
        ValueError: When the input does not contain valid JSON strings or the JSON is incompatible with the class
    """
    use_logger = kwargs.get("log", True)

    extracted_json_list = parse_json_from_text(content)
    if len(extracted_json_list) == 0:
        error_message = f"The input to {cls.__name__}.from_str does not contain any valid JSON str."
        if use_logger:
            logger.error(error_message)
        raise ValueError(error_message)

    module = None
    for json_str in extracted_json_list:
        try:
            module = cls.from_json(json_str, log=False)
        except Exception:
            continue
        break

    if module is None:
        error_message = f"Can not instantiate {cls.__name__}. The input to {cls.__name__}.from_str either does not contain a valide JSON str, or the JSON str is incomplete or incompatable (incorrect variables or types) with {cls.__name__}."
        error_message += f"\nInput:\n{content}"
        if use_logger:
            logger.error(error_message)
        raise ValueError(error_message)

    return module

load_module classmethod

load_module(path: str, **kwargs) -> dict

Load the values for a module from a file.

By default, it opens the specified file and uses yaml.safe_load to parse its contents into a Python object (typically a dictionary).

Parameters:

Name	Type	Description	Default
path	str	The path of the file	required
**kwargs	Any	Additional keyword arguments	{}

Returns:

Name	Type	Description
dict	dict	The JSON object instantiated from the file

Source code in evoagentx/core/module.py

@classmethod 
def load_module(cls, path: str, **kwargs) -> dict:
    """
    Load the values for a module from a file.

    By default, it opens the specified file and uses `yaml.safe_load` to parse its contents 
    into a Python object (typically a dictionary).

    Args:
        path: The path of the file
        **kwargs (Any): Additional keyword arguments

    Returns:
        dict: The JSON object instantiated from the file
    """
    with open(path, mode="r", encoding="utf-8") as file:
        content = yaml.safe_load(file.read())
    return content

from_file classmethod

from_file(path: str, load_function: Callable = None, **kwargs) -> BaseModule

Construct the BaseModule from a file.

This method reads and parses a file into a data structure, then creates a module instance from that data. It first verifies that the file exists, then uses either the provided load_function or the default load_module method to read and parse the file content, and finally calls from_dict to create the instance.

Parameters:

Name	Type	Description	Default
path	str	The path of the file	required
load_function	Callable	The function used to load the data, takes a file path as input and returns a JSON object	None
**kwargs	Any	Additional keyword arguments, can include log to control logging output	{}

Returns:

Name	Type	Description
BaseModule	BaseModule	The created module instance

Raises:

Type	Description
ValueError	When the file does not exist

Source code in evoagentx/core/module.py

@classmethod
def from_file(cls, path: str, load_function: Callable=None, **kwargs) -> "BaseModule":
    """
    Construct the BaseModule from a file.

    This method reads and parses a file into a data structure, then creates
    a module instance from that data. It first verifies that the file exists,
    then uses either the provided `load_function` or the default `load_module`
    method to read and parse the file content, and finally calls `from_dict`
    to create the instance.

    Args:
        path: The path of the file
        load_function: The function used to load the data, takes a file path as input and returns a JSON object
        **kwargs (Any): Additional keyword arguments, can include `log` to control logging output

    Returns:
        BaseModule: The created module instance

    Raises:
        ValueError: When the file does not exist
    """
    use_logger = kwargs.get("log", True)
    if not os.path.exists(path):
        error_message = f"File \"{path}\" does not exist!"
        if use_logger:
            logger.error(error_message)
        raise ValueError(error_message)

    function = load_function or cls.load_module
    content = function(path, **kwargs)
    module = cls.from_dict(content, log=use_logger)

    return module

to_dict

to_dict(exclude_none: bool = True, ignore: List[str] = [], **kwargs) -> dict

Convert the BaseModule to a dictionary.

Parameters:

Name	Type	Description	Default
exclude_none	bool	Whether to exclude fields with None values	True
ignore	List[str]	List of field names to ignore	[]
**kwargs	Any	Additional keyword arguments	{}

Returns:

Name	Type	Description
dict	dict	Dictionary containing the object data

Source code in evoagentx/core/module.py

def to_dict(self, exclude_none: bool = True, ignore: List[str] = [], **kwargs) -> dict:
    """
    Convert the BaseModule to a dictionary.

    Args:
        exclude_none: Whether to exclude fields with None values
        ignore: List of field names to ignore
        **kwargs (Any): Additional keyword arguments

    Returns:
        dict: Dictionary containing the object data
    """
    data = {}
    for field_name, _ in type(self).model_fields.items():
        if field_name in ignore:
            continue
        field_value = getattr(self, field_name, None)
        if exclude_none and field_value is None:
            continue
        if isinstance(field_value, BaseModule):
            data[field_name] = field_value.to_dict(exclude_none=exclude_none, ignore=ignore)
        elif isinstance(field_value, list):
            data[field_name] = [
                item.to_dict(exclude_none=exclude_none, ignore=ignore) if isinstance(item, BaseModule) else item
                for item in field_value
            ]
        elif isinstance(field_value, dict):
            data[field_name] = {
                key: value.to_dict(exclude_none=exclude_none, ignore=ignore) if isinstance(value, BaseModule) else value
                for key, value in field_value.items()
            }
        else:
            data[field_name] = field_value

    return data

to_json

to_json(use_indent: bool = False, ignore: List[str] = [], **kwargs) -> str

Convert the BaseModule to a JSON string.

Parameters:

Name	Type	Description	Default
use_indent	bool	Whether to use indentation	False
ignore	List[str]	List of field names to ignore	[]
**kwargs	Any	Additional keyword arguments	{}

Returns:

Name	Type	Description
str	str	The JSON string

Source code in evoagentx/core/module.py

def to_json(self, use_indent: bool=False, ignore: List[str] = [], **kwargs) -> str:
    """
    Convert the BaseModule to a JSON string.

    Args:
        use_indent: Whether to use indentation
        ignore: List of field names to ignore
        **kwargs (Any): Additional keyword arguments

    Returns:
        str: The JSON string
    """
    if use_indent:
        kwargs["indent"] = kwargs.get("indent", 4)
    else:
        kwargs.pop("indent", None)
    if kwargs.get("default", None) is None:
        kwargs["default"] = custom_serializer
    data = self.to_dict(exclude_none=True)
    for ignore_field in ignore:
        data.pop(ignore_field, None)
    return json.dumps(data, **kwargs)

to_str

to_str(**kwargs) -> str

Convert the BaseModule to a string. Use .to_json to output JSON string by default.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments	{}

Returns:

Name	Type	Description
str	str	The string

Source code in evoagentx/core/module.py

def to_str(self, **kwargs) -> str:
    """
    Convert the BaseModule to a string. Use .to_json to output JSON string by default.

    Args:
        **kwargs (Any): Additional keyword arguments

    Returns:
        str: The string
    """
    return self.to_json(use_indent=False)

save_module

save_module(path: str, ignore: List[str] = [], **kwargs) -> str

Save the BaseModule to a file.

This method will set non-serializable objects to None by default. If you want to save non-serializable objects, override this method. Remember to also override the load_module function to ensure the loaded object can be correctly parsed by cls.from_dict.

Parameters:

Name	Type	Description	Default
path	str	The path to save the file	required
ignore	List[str]	List of field names to ignore	[]
**kwargs	Any	Additional keyword arguments	{}

Returns:

Name	Type	Description
str	str	The path where the file is saved, same as the input path

Source code in evoagentx/core/module.py

def save_module(self, path: str, ignore: List[str] = [], **kwargs)-> str:
    """
    Save the BaseModule to a file.

    This method will set non-serializable objects to None by default.
    If you want to save non-serializable objects, override this method.
    Remember to also override the `load_module` function to ensure the loaded
    object can be correctly parsed by `cls.from_dict`.

    Args:
        path: The path to save the file
        ignore: List of field names to ignore
        **kwargs (Any): Additional keyword arguments

    Returns:
        str: The path where the file is saved, same as the input path
    """
    logger.info("Saving {} to {}", self.__class__.__name__, path)
    return save_json(self.to_json(use_indent=True, default=lambda x: None, ignore=ignore), path=path)

deepcopy

deepcopy()

Deep copy the module.

This is a tweak to the default python deepcopy that only deep copies self.parameters(), and for other attributes, we just do the shallow copy.

Source code in evoagentx/core/module.py

def deepcopy(self):
    """Deep copy the module.

    This is a tweak to the default python deepcopy that only deep copies `self.parameters()`, and for other
    attributes, we just do the shallow copy.
    """
    try:
        # If the instance itself is copyable, we can just deep copy it.
        # Otherwise we will have to create a new instance and copy over the attributes one by one.
        return copy.deepcopy(self)
    except Exception:
        pass

    # Create an empty instance.
    new_instance = self.__class__.__new__(self.__class__)
    # Set attribuetes of the copied instance.
    for attr, value in self.__dict__.items():
        if isinstance(value, BaseModule):
            setattr(new_instance, attr, value.deepcopy())
        else:
            try:
                # Try to deep copy the attribute
                setattr(new_instance, attr, copy.deepcopy(value))
            except Exception:
                logging.warning(
                    f"Failed to deep copy attribute '{attr}' of {self.__class__.__name__}, "
                    "falling back to shallow copy or reference copy."
                )
                try:
                    # Fallback to shallow copy if deep copy fails
                    setattr(new_instance, attr, copy.copy(value))
                except Exception:
                    # If even the shallow copy fails, we just copy over the reference.
                    setattr(new_instance, attr, value)

    return new_instance

DBStoreFactory

Factory class for creating database store instances based on provider and configuration. Maps provider names to specific database store classes.

create classmethod

create(provider_name: str, config: DBConfig)

Create a database store instance for the specified provider.

Attributes:

Name	Type	Description
provider_name	str	Name of the database provider (e.g., 'sqlite', 'posgre_sql').
config	DBConfig	Configuration for the database store.

Returns:

Name	Type	Description
DBStoreBase		An instance of the database store.

Raises:

Type	Description
ValueError	If the provider is not supported.

Source code in evoagentx/utils/factory.py

@classmethod
def create(cls, provider_name: str, config: DBConfig):
    """
    Create a database store instance for the specified provider.

    Attributes:
        provider_name (str): Name of the database provider (e.g., 'sqlite', 'posgre_sql').
        config (DBConfig): Configuration for the database store.

    Returns:
        DBStoreBase: An instance of the database store.

    Raises:
        ValueError: If the provider is not supported.
    """
    class_type = cls.provider_to_class.get(provider_name)
    if class_type:
        if not isinstance(config, dict):
            config = config.model_dump()
        db_store_class = load_class(class_type)
        return db_store_class(**config)
    else:
        raise ValueError(f"Unsupported Database provider: {provider_name}")

VectorStoreFactory

Factory class for creating vector store instances based on configuration. Maps provider names to specific vector store classes.

create

create(config: VectorStoreConfig)

Create a vector store instance based on the provided configuration.

Attributes:

Name	Type	Description
config	VectorStoreConfig	Configuration for the vector store.

Returns:

Name	Type	Description
VectorStoreBase		An instance of the vector store.

Source code in evoagentx/utils/factory.py

def create(cls, config: VectorStoreConfig):
    """
    Create a vector store instance based on the provided configuration.

    Attributes:
        config (VectorStoreConfig): Configuration for the vector store.

    Returns:
        VectorStoreBase: An instance of the vector store.
    """
    # TODO: Implement vector store creation logic
    pass

GraphStoreFactory

Factory class for creating graph store instances based on configuration. Maps provider names to specific graph store classes.

create classmethod

create(config: VectorStoreConfig)

Create a graph store instance based on the provided configuration.

Attributes:

Name	Type	Description
config	VectorStoreConfig	Configuration for the graph store.

Returns:

Name	Type	Description
GraphStoreBase		An instance of the graph store.

Source code in evoagentx/utils/factory.py

@classmethod
def create(cls, config: VectorStoreConfig):
    """
    Create a graph store instance based on the provided configuration.

    Attributes:
        config (VectorStoreConfig): Configuration for the graph store.

    Returns:
        GraphStoreBase: An instance of the graph store.
    """
    # TODO: Implement graph store creation logic
    pass

StoreConfig

StoreConfig(**kwargs)

Bases: BaseConfig

Aggregates database, vector, file, and graph store configurations.

Source code in evoagentx/core/module.py

def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

TableType

Bases: str, Enum

Enum representing the default table types for the database. Each value corresponds to a specific table name used for storing different types of data.

AgentStore

Bases: BaseModel

Stores agent metadata with a unique name and content dictionary.

WorkflowStore

Bases: BaseModel

Stores workflow metadata with a unique name and content dictionary.

MemoryStore

Bases: BaseModel

Stores memory-related metadata with optional fields for keywords, entities, and embeddings.

HistoryStore

Bases: BaseModel

Stores changes to memory with event details and timestamps.

StorageHandler

StorageHandler(**kwargs)

Bases: BaseModule

Implementation of a storage handler for managing various storage backends.

StorageHandler provides an abstraction for reading and writing data (e.g., memory, agents, workflows). It supports multiple storage types, including database, vector, and graph storage, initialized via factories.

Source code in evoagentx/core/module.py

def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

init_module

init_module()

Initialize all storage backends based on the provided configuration. Calls individual initialization methods for database, vector, and graph stores.

Source code in evoagentx/storages/base.py

def init_module(self):
    """
    Initialize all storage backends based on the provided configuration.
    Calls individual initialization methods for database, vector, and graph stores.
    """
    self._init_db_store()
    self._init_vector_store()
    self._init_graph_store()

load

load(tables: Optional[List[str]] = None, *args, **kwargs) -> Dict[str, Any]

Load all data from the database storage.

Attributes:

Name	Type	Description
tables	Optional[List[str]]	List of table names to load; if None, loads all tables.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Dict[str, str]]: A dictionary with table names as keys and lists of records as values. You should parse the values by yourself.

Source code in evoagentx/storages/base.py

def load(self, tables: Optional[List[str]] = None, *args, **kwargs) -> Dict[str, Any]:
    """
    Load all data from the database storage.

    Attributes:
        tables (Optional[List[str]]): List of table names to load; if None, loads all tables.

    Returns:
        Dict[str, Dict[str, str]]: A dictionary with table names as keys and lists of records as values. You should parse the values by yourself.
    """
    result = {}
    table_info = self.storageDB.col_info()

    if tables is None:
        tables_to_load = [t.value for t in TableType]
    else:
        tables_to_load = tables

    # Load data for each table
    for table_name in tables_to_load:
        table_data = []
        # Check if the table exists
        if any(t["table_name"] == table_name for t in table_info):
            cursor = self.storageDB.connection.cursor()
            cursor.execute(f"SELECT * FROM {table_name}")
            # Get column names from the columns dictionary
            columns = next(t["columns"].keys() for t in table_info if t["table_name"] == table_name)
            rows = cursor.fetchall()
            table_data = [dict(zip(columns, row)) for row in rows]
        result[table_name] = table_data

    return result

save

save(data: Dict[str, Any], *args, **kwargs)

Save all provided data to the database storage.

Attributes:

Name	Type	Description
data	Dict[str, Any]	Dictionary with table names as keys and lists of records to save.

Raises:

Type	Description
ValueError	If an unknown table name is provided.

Source code in evoagentx/storages/base.py

def save(self, data: Dict[str, Any], *args, **kwargs):
    """
    Save all provided data to the database storage.

    Attributes:
        data (Dict[str, Any]): Dictionary with table names as keys and lists of records to save.

    Raises:
        ValueError: If an unknown table name is provided.
    """
    for table_name, records in data.items():
        store_type = None
        # Map table name to store_type
        for st in TableType:
            if st.value == table_name:
                store_type = st
                break
        if store_type is None:
            raise ValueError(f"Unknown table: {table_name}")
        # Insert each record
        for record in records:
            self.storageDB.insert(metadata=record, store_type=store_type, table=table_name)

parse_result

parse_result(results: Dict[str, str], store: Union[AgentStore, WorkflowStore, MemoryStore, HistoryStore]) -> Dict[str, Any]

Parse database results, converting JSON strings to Python objects where applicable.

Attributes:

Name	Type	Description
results	Dict[str, str]	Raw database results with column names as keys.
store	Union[AgentStore, WorkflowStore, MemoryStore, HistoryStore]	Pydantic model for validation.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Parsed results with JSON strings deserialized to Python objects.

Source code in evoagentx/storages/base.py

def parse_result(self, results: Dict[str, str], 
                 store: Union[AgentStore, WorkflowStore, MemoryStore, HistoryStore]) -> Dict[str, Any]:
    """
    Parse database results, converting JSON strings to Python objects where applicable.

    Attributes:
        results (Dict[str, str]): Raw database results with column names as keys.
        store (Union[AgentStore, WorkflowStore, MemoryStore, HistoryStore]): Pydantic model for validation.

    Returns:
        Dict[str, Any]: Parsed results with JSON strings deserialized to Python objects.
    """
    for k, v in store.model_fields.items():
        if v.annotation not in [Optional[str], str]:
            try:
                results[k] = json.loads(results[k])
            except (json.JSONDecodeError, KeyError, TypeError):
                results[k] = results.get(k)
    return results

load_memory

load_memory(memory_id: str, table: Optional[str] = None, **kwargs) -> Dict[str, Any]

Load a single long-term memory data.

Attributes:

Name	Type	Description
memory_id	str	The ID of the long-term memory.
table	Optional[str]	The table name; defaults to 'memory' if None.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The data that can be used to create a LongTermMemory instance.

Source code in evoagentx/storages/base.py

def load_memory(self, memory_id: str, table: Optional[str]=None, **kwargs) -> Dict[str, Any]:
    """
    Load a single long-term memory data.

    Attributes:
        memory_id (str): The ID of the long-term memory.
        table (Optional[str]): The table name; defaults to 'memory' if None.

    Returns:
        Dict[str, Any]: The data that can be used to create a LongTermMemory instance.
    """
    pass

save_memory

save_memory(memory_data: Dict[str, Any], table: Optional[str] = None, **kwargs)

Save or update a single memory.

Attributes:

Name	Type	Description
memory_data	Dict[str, Any]	The long-term memory's data.
table	Optional[str]	The table name; defaults to 'memory' if None.

Source code in evoagentx/storages/base.py

def save_memory(self, memory_data: Dict[str, Any], table: Optional[str]=None, **kwargs):
    """
    Save or update a single memory.

    Attributes:
        memory_data (Dict[str, Any]): The long-term memory's data.
        table (Optional[str]): The table name; defaults to 'memory' if None.

    """
    pass

load_agent

load_agent(agent_name: str, table: Optional[str] = None, *args, **kwargs) -> Dict[str, Any]

Load a single agent's data.

Attributes:

Name	Type	Description
agent_name	str	The unique name of the agent to retrieve.
table	Optional[str]	The table name; defaults to 'agent' if None.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The data that can be used to create an Agent instance, or None if not found.

Source code in evoagentx/storages/base.py

def load_agent(self, agent_name: str, table: Optional[str]=None, *args, **kwargs) -> Dict[str, Any]:
    """
    Load a single agent's data.

    Attributes:
        agent_name (str): The unique name of the agent to retrieve.
        table (Optional[str]): The table name; defaults to 'agent' if None.

    Returns:
        Dict[str, Any]: The data that can be used to create an Agent instance, or None if not found.
    """
    table = table or TableType.store_agent.value
    result = self.storageDB.get_by_id(agent_name, store_type="agent", table=table)
    # Parse the result to convert JSON strings to Python objects
    if result is not None:
        result = self.parse_result(result, AgentStore)
    return result

remove_agent

remove_agent(agent_name: str, table: Optional[str] = None, *args, **kwargs)

Remove an agent from storage if the agent exists.

Attributes:

Name	Type	Description
agent_name	str	The name of the agent to be deleted.
table	Optional[str]	The table name; defaults to 'agent' if None.

Raises:

Type	Description
ValueError	If the agent does not exist in the specified table.

Source code in evoagentx/storages/base.py

def remove_agent(self, agent_name: str, table: Optional[str]=None, *args, **kwargs):
    """
    Remove an agent from storage if the agent exists.

    Attributes:
        agent_name (str): The name of the agent to be deleted.
        table (Optional[str]): The table name; defaults to 'agent' if None.

    Raises:
        ValueError: If the agent does not exist in the specified table.
    """
    table = table or TableType.store_agent.value
    success = self.storageDB.delete(agent_name, store_type="agent", table=table)
    if not success:
        raise ValueError(f"Agent with name {agent_name} not found in table {table}")

save_agent

save_agent(agent_data: Dict[str, Any], table: Optional[str] = None, *args, **kwargs)

Save or update a single agent's data.

Attributes:

Name	Type	Description
agent_data	Dict[str, Any]	The agent's data, must include 'name' and 'content' keys.
table	Optional[str]	The table name; defaults to 'agent' if None.

Raises:

Type	Description
ValueError	If 'name' field is missing or if Pydantic validation fails.

Source code in evoagentx/storages/base.py

def save_agent(self, agent_data: Dict[str, Any], table: Optional[str]=None, *args, **kwargs):
    """
    Save or update a single agent's data.

    Attributes:
        agent_data (Dict[str, Any]): The agent's data, must include 'name' and 'content' keys.
        table (Optional[str]): The table name; defaults to 'agent' if None.

    Raises:
        ValueError: If 'name' field is missing or if Pydantic validation fails.
    """
    table = table or TableType.store_agent.value
    agent_name = agent_data.get("name")
    if not agent_name:
        raise ValueError("Agent data must include a 'name' field")

    existing = self.storageDB.get_by_id(agent_name, store_type="agent", table=table)
    if existing:
        self.storageDB.update(agent_name, new_metadata=agent_data, store_type="agent", table=table)
    else:
        self.storageDB.insert(metadata=agent_data, store_type="agent", table=table)

load_workflow

load_workflow(workflow_id: str, table: Optional[str] = None, *args, **kwargs) -> Dict[str, Any]

Load a single workflow's data.

Attributes:

Name	Type	Description
workflow_id	str	The ID of the workflow.
table	Optional[str]	The table name; defaults to 'workflow' if None.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The data that can be used to create a WorkFlow instance, or None if not found.

Source code in evoagentx/storages/base.py

def load_workflow(self, workflow_id: str, table: Optional[str] = None, *args, **kwargs) -> Dict[str, Any]:
    """
    Load a single workflow's data.

    Attributes:
        workflow_id (str): The ID of the workflow.
        table (Optional[str]): The table name; defaults to 'workflow' if None.

    Returns:
        Dict[str, Any]: The data that can be used to create a WorkFlow instance, or None if not found.
    """
    table = table or TableType.store_workflow.value
    result = self.storageDB.get_by_id(workflow_id, store_type="workflow", table=table)
    # Parse the result to convert JSON strings to Python objects
    if result is not None:
        result = self.parse_result(result, WorkflowStore)
    return result

save_workflow

save_workflow(workflow_data: Dict[str, Any], table: Optional[str] = None, *args, **kwargs)

Save or update a workflow's data.

Attributes:

Name	Type	Description
workflow_data	Dict[str, Any]	The workflow's data, must include 'name' field.
table	Optional[str]	The table name; defaults to 'workflow' if None.

Raises:

Type	Description
ValueError	If 'name' field is missing or if Pydantic validation fails.

Source code in evoagentx/storages/base.py

def save_workflow(self, workflow_data: Dict[str, Any], table: Optional[str] = None, *args, **kwargs):
    """
    Save or update a workflow's data.

    Attributes:
        workflow_data (Dict[str, Any]): The workflow's data, must include 'name' field.
        table (Optional[str]): The table name; defaults to 'workflow' if None.

    Raises:
        ValueError: If 'name' field is missing or if Pydantic validation fails.
    """
    table = table or TableType.store_workflow.value
    workflow_id = workflow_data.get("name")
    if not workflow_id:
        raise ValueError("Workflow data must include a 'name' field")
    # Check if workflow exists to decide between insert or update
    existing = self.storageDB.get_by_id(workflow_id, store_type="workflow", table=table)
    if existing:

        self.storageDB.update(workflow_id, new_metadata=workflow_data, store_type="workflow", table=table)
    else:
        self.storageDB.insert(metadata=workflow_data, store_type="workflow", table=table)

load_history

load_history(memory_id: str, table: Optional[str] = None, *args, **kwargs) -> Dict[str, Any]

Load a single history entry.

Attributes:

Name	Type	Description
memory_id	str	The ID of the memory associated with the history entry.
table	Optional[str]	The table name; defaults to 'history' if None.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The history data, or None if not found.

Source code in evoagentx/storages/base.py

def load_history(self, memory_id: str, table: Optional[str] = None, *args, **kwargs) -> Dict[str, Any]:
    """
    Load a single history entry.

    Attributes:
        memory_id (str): The ID of the memory associated with the history entry.
        table (Optional[str]): The table name; defaults to 'history' if None.

    Returns:
        Dict[str, Any]: The history data, or None if not found.
    """
    table = table or TableType.store_history.value
    result = self.storageDB.get_by_id(memory_id, store_type="history", table=table)
    # Parse the result to convert JSON strings to Python objects (if any)
    if result is not None:
        result = self.parse_result(result, HistoryStore)
    return result

save_history

save_history(history_data: Dict[str, Any], table: Optional[str] = None, *args, **kwargs)

Save or update a single history entry.

Attributes:

Name	Type	Description
history_data	Dict[str, Any]	The history data, must include 'memory_id' field.
table	Optional[str]	The table name; defaults to 'history' if None.

Raises:

Type	Description
ValueError	If 'memory_id' field is missing or if Pydantic validation fails.

Source code in evoagentx/storages/base.py

def save_history(self, history_data: Dict[str, Any], table: Optional[str] = None, *args, **kwargs):
    """
    Save or update a single history entry.

    Attributes:
        history_data (Dict[str, Any]): The history data, must include 'memory_id' field.
        table (Optional[str]): The table name; defaults to 'history' if None.

    Raises:
        ValueError: If 'memory_id' field is missing or if Pydantic validation fails.
    """
    table = table or TableType.store_history.value
    memory_id = history_data.get("memory_id")
    if not memory_id:
        raise ValueError("History data must include a 'memory_id' field")
    # Check if history entry exists to decide between insert or update
    existing = self.storageDB.get_by_id(memory_id, store_type="history", table=table)
    if existing:
        # parse the history, then change the old_hisotry
        result = HistoryStore.model_validate(self.parse_result(existing, HistoryStore))
        history_data["old_memory"] = result.old_memory
        self.storageDB.update(memory_id, new_metadata=history_data, store_type="history", table=table)
    else:
        self.storageDB.insert(metadata=history_data, store_type="history", table=table)