Skip to content

πŸ€– Agents

evoagentx.agents

Agent

Agent(**kwargs)

Bases: BaseModule

Base class for all agents.

Attributes:

Name Type Description
name str

Unique identifier for the agent

description str

Human-readable description of the agent's purpose

llm_config Optional[LLMConfig]

Configuration for the language model. If provided, a new LLM instance will be created. Otherwise, the existing LLM instance specified in the llm field will be used.

llm Optional[BaseLLM]

Language model instance. If provided, the existing LLM instance will be used.

agent_id Optional[str]

Unique ID for the agent, auto-generated if not provided

system_prompt Optional[str]

System prompt for the Agent.

actions List[Action]

List of available actions

n Optional[int]

Number of latest messages used to provide context for action execution. It uses all the messages in short term memory by default.

is_human bool

Whether this agent represents a human user

version int

Version number of the agent, default is 0.

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)

__call__

__call__(*args: Any, **kwargs: Any) -> Union[dict, Coroutine[Any, Any, dict]]

Make the operator callable and automatically choose between sync and async execution.

Source code in evoagentx/agents/agent.py
def __call__(self, *args: Any, **kwargs: Any) -> Union[dict, Coroutine[Any, Any, dict]]:
    """Make the operator callable and automatically choose between sync and async execution."""
    try:
        # Safe way to check if we're inside an async environment
        asyncio.get_running_loop()
        return self.async_execute(*args, **kwargs)
    except RuntimeError:
        # No running loop β€” likely in sync context or worker thread
        return self.execute(*args, **kwargs)

async_execute async

async_execute(action_name: str, msgs: Optional[List[Message]] = None, action_input_data: Optional[dict] = None, return_msg_type: Optional[MessageType] = UNKNOWN, return_action_input_data: Optional[bool] = False, **kwargs) -> Union[Message, Tuple[Message, dict]]

Execute an action asynchronously with the given context and return results.

This is the async version of the execute method, allowing it to perform actions based on the current conversation context.

Parameters:

Name Type Description Default
action_name str

The name of the action to execute

required
msgs Optional[List[Message]]

Optional list of messages providing context for the action

None
action_input_data Optional[dict]

Optional pre-extracted input data for the action

None
return_msg_type Optional[MessageType]

Message type for the return message

UNKNOWN
**kwargs Any

Additional parameters, may include workflow information

{}

Returns:

Name Type Description
Message Union[Message, Tuple[Message, dict]]

A message containing the execution results

Source code in evoagentx/agents/agent.py
async def async_execute(
    self, 
    action_name: str, 
    msgs: Optional[List[Message]] = None, 
    action_input_data: Optional[dict] = None, 
    return_msg_type: Optional[MessageType] = MessageType.UNKNOWN,
    return_action_input_data: Optional[bool] = False, 
    **kwargs
) -> Union[Message, Tuple[Message, dict]]:
    """Execute an action asynchronously with the given context and return results.

    This is the async version of the execute method, allowing it to perform actions
    based on the current conversation context.

    Args:
        action_name: The name of the action to execute
        msgs: Optional list of messages providing context for the action
        action_input_data: Optional pre-extracted input data for the action
        return_msg_type: Message type for the return message
        **kwargs (Any): Additional parameters, may include workflow information

    Returns:
        Message: A message containing the execution results
    """
    action, action_input_data = self._prepare_execution(
        action_name=action_name,
        msgs=msgs,
        action_input_data=action_input_data,
        **kwargs
    )

    if is_method_overridden(action, Action, "async_execute"):
        execute_function = action.async_execute
    elif is_method_overridden(action, Action, "execute"):
        execute_function = action.execute
    else:
        raise NotImplementedError(
            f"The action '{type(action).__name__}' must implement `execute` or `async_execute`."
        )

    execution_results = await call_maybe_async(
        execute_function,
        llm=self.llm,
        inputs=action_input_data,
        sys_msg=self.system_prompt,
        return_prompt=True,
        **kwargs
    )
    action_output, prompt = execution_results

    message = self._create_output_message(
        action_output=action_output,
        prompt=prompt,
        action_name=action_name,
        return_msg_type=return_msg_type,
        **kwargs
    )
    if return_action_input_data:
        return message, action_input_data
    return message

execute

execute(action_name: str, msgs: Optional[List[Message]] = None, action_input_data: Optional[dict] = None, return_msg_type: Optional[MessageType] = UNKNOWN, return_action_input_data: Optional[bool] = False, **kwargs) -> Union[Message, Tuple[Message, dict]]

Execute an action with the given context and return results.

This is the core method for agent functionality, allowing it to perform actions based on the current conversation context.

Parameters:

Name Type Description Default
action_name str

The name of the action to execute

required
msgs Optional[List[Message]]

Optional list of messages providing context for the action

None
action_input_data Optional[dict]

Optional pre-extracted input data for the action

None
return_msg_type Optional[MessageType]

Message type for the return message

UNKNOWN
**kwargs Any

Additional parameters, may include workflow information

{}

Returns:

Name Type Description
Message Union[Message, Tuple[Message, dict]]

A message containing the execution results

Source code in evoagentx/agents/agent.py
def execute(
    self, 
    action_name: str, 
    msgs: Optional[List[Message]] = None, 
    action_input_data: Optional[dict] = None, 
    return_msg_type: Optional[MessageType] = MessageType.UNKNOWN,
    return_action_input_data: Optional[bool] = False, 
    **kwargs
) -> Union[Message, Tuple[Message, dict]]:
    """Execute an action with the given context and return results.

    This is the core method for agent functionality, allowing it to perform actions
    based on the current conversation context.

    Args:
        action_name: The name of the action to execute
        msgs: Optional list of messages providing context for the action
        action_input_data: Optional pre-extracted input data for the action
        return_msg_type: Message type for the return message
        **kwargs (Any): Additional parameters, may include workflow information

    Returns:
        Message: A message containing the execution results
    """
    action, action_input_data = self._prepare_execution(
        action_name=action_name,
        msgs=msgs,
        action_input_data=action_input_data,
        **kwargs
    )

    # execute action
    execution_results = action.execute(
        llm=self.llm, 
        inputs=action_input_data, 
        sys_msg=self.system_prompt,
        return_prompt=True,
        **kwargs
    )
    action_output, prompt = execution_results

    message = self._create_output_message(
        action_output=action_output,
        prompt=prompt,
        action_name=action_name,
        return_msg_type=return_msg_type,
        **kwargs
    )
    if return_action_input_data:
        return message, action_input_data
    return message

set_llm_config

set_llm_config(llm_config: LLMConfig)

Set a new LLM config and rebuild the LLM instance from it.

Source code in evoagentx/agents/agent.py
def set_llm_config(self, llm_config: LLMConfig):
    """Set a new LLM config and rebuild the LLM instance from it."""
    self.llm_config = llm_config
    llm_cls = MODEL_REGISTRY.get_model(llm_config.llm_type)
    self.llm = llm_cls(config=llm_config)

set_llm

set_llm(llm: BaseLLM)

Set a new LLM instance and sync llm_config from it.

Source code in evoagentx/agents/agent.py
def set_llm(self, llm: BaseLLM):
    """Set a new LLM instance and sync llm_config from it."""
    self.llm = llm
    self.llm_config = llm.config

init_llm

init_llm()

Initialize the language model for the agent.

Source code in evoagentx/agents/agent.py
def init_llm(self):
    """
    Initialize the language model for the agent.
    """
    if self.is_human:
        # if human, no need to initialize LLM
        return

    # 1. Handle the case where both llm and llm_config are set
    if self.llm and self.llm_config:
        if self.llm.config != self.llm_config:
            raise ValueError(
                f"Inconsistent LLM setup for agent '{self.name}': "
                "The provided `llm` does not match `llm_config`. "
                "Ensure they match or only provide one."
            )
        return

    # 2. Handle the case where only one (llm or llm_config) is provided
    if self.llm_config:
        self.set_llm_config(self.llm_config)
    elif self.llm:
        self.set_llm(self.llm)
    else:
        raise ValueError(f"Must provide `llm_config` or `llm` for agent '{self.name}'.")

init_long_term_memory

init_long_term_memory()

Initialize long-term memory components.

Source code in evoagentx/agents/agent.py
def init_long_term_memory(self):
    """
    Initialize long-term memory components.
    """
    assert self.storage_handler is not None, "must provide ``storage_handler`` when use_long_term_memory=True"
    # TODO revise the initialisation of long_term_memory and long_term_memory_manager
    if not self.long_term_memory:
        self.long_term_memory = LongTermMemory()
    if not self.long_term_memory_manager:
        self.long_term_memory_manager = MemoryManager(
            storage_handler=self.storage_handler,
            memory=self.long_term_memory
        )

init_context_extractor

init_context_extractor()

Initialize the context extraction action.

Source code in evoagentx/agents/agent.py
def init_context_extractor(self):
    """
    Initialize the context extraction action.
    """
    cext_action = ContextExtraction()
    self.cext_action_name = cext_action.name
    self.add_action(cext_action)

add_action

add_action(action: Type[Action])

Add a new action to the agent's available actions.

Parameters:

Name Type Description Default
action Type[Action]

The action instance to add

required
Source code in evoagentx/agents/agent.py
def add_action(self, action: Type[Action]):
    """
    Add a new action to the agent's available actions.

    Args:
        action: The action instance to add
    """
    action_name  = action.name
    if action_name in self._action_map:
        return
    self.actions.append(action)
    self._action_map[action_name] = action

check_action_name

check_action_name(action_name: str)

Check if an action name is valid for this agent.

Parameters:

Name Type Description Default
action_name str

Name of the action to check

required
Source code in evoagentx/agents/agent.py
def check_action_name(self, action_name: str):
    """
    Check if an action name is valid for this agent.

    Args:
        action_name: Name of the action to check
    """
    if action_name not in self._action_map:
        raise KeyError(f"'{action_name}' is an invalid action for {self.name}! Available action names: {list(self._action_map.keys())}")

get_action

get_action(action_name: str) -> Action

Retrieves the Action instance associated with the given name.

Parameters:

Name Type Description Default
action_name str

Name of the action to retrieve

required

Returns:

Type Description
Action

The Action instance with the specified name

Source code in evoagentx/agents/agent.py
def get_action(self, action_name: str) -> Action:
    """
    Retrieves the Action instance associated with the given name.

    Args:
        action_name: Name of the action to retrieve

    Returns:
        The Action instance with the specified name
    """
    self.check_action_name(action_name=action_name)
    return self._action_map[action_name]

get_action_name

get_action_name(action_cls: Type[Action]) -> str

Searches through the agent's actions to find one matching the specified type.

Parameters:

Name Type Description Default
action_cls Type[Action]

The Action class type to search for

required

Returns:

Type Description
str

The name of the matching action

Source code in evoagentx/agents/agent.py
def get_action_name(self, action_cls: Type[Action]) -> str:
    """
    Searches through the agent's actions to find one matching the specified type.

    Args:
        action_cls: The Action class type to search for

    Returns:
        The name of the matching action
    """
    for name, action in self._action_map.items():
        if isinstance(action, action_cls):
            return name
    raise ValueError(f"Couldn't find an action that matches Type '{action_cls.__name__}'")

get_action_inputs

get_action_inputs(action: Action) -> Union[dict, None]

Uses the context extraction action to determine appropriate inputs for the specified action based on the conversation history.

Parameters:

Name Type Description Default
action Action

The action for which to extract inputs

required

Returns:

Type Description
Union[dict, None]

Dictionary of extracted input data, or None if extraction fails

Source code in evoagentx/agents/agent.py
def get_action_inputs(self, action: Action) -> Union[dict, None]:
    """
    Uses the context extraction action to determine appropriate inputs
    for the specified action based on the conversation history.

    Args:
        action: The action for which to extract inputs

    Returns:
        Dictionary of extracted input data, or None if extraction fails
    """
    # return the input data of an action.
    context = self.short_term_memory.get(n=self.n)
    cext_action = self.get_action(self.cext_action_name)
    action_inputs = cext_action.execute(llm=self.llm, action=action, context=context)
    return action_inputs

get_all_actions

get_all_actions() -> List[Action]

Get all actions except the context extraction action.

Returns:

Type Description
List[Action]

List of Action instances available for execution

Source code in evoagentx/agents/agent.py
def get_all_actions(self) -> List[Action]:
    """Get all actions except the context extraction action.

    Returns:
        List of Action instances available for execution
    """
    actions = [action for action in self.actions if action.name != self.cext_action_name]
    return actions

get_agent_profile

get_agent_profile(action_names: List[str] = None) -> str

Generate a human-readable profile of the agent and its capabilities.

Parameters:

Name Type Description Default
action_names List[str]

Optional list of action names to include in the profile. If None, all actions are included.

None

Returns:

Type Description
str

A formatted string containing the agent profile

Source code in evoagentx/agents/agent.py
def get_agent_profile(self, action_names: List[str] = None) -> str:
    """Generate a human-readable profile of the agent and its capabilities.

    Args:
        action_names: Optional list of action names to include in the profile.
                      If None, all actions are included.

    Returns:
        A formatted string containing the agent profile
    """
    all_actions = self.get_all_actions()
    if action_names is None:
        # if `action_names` is None, return description of all actions 
        action_descriptions = "\n".join([f"  - {action.name}: {action.description}" for action in all_actions])
    else: 
        # otherwise, only return description of actions that matches `action_names`
        action_descriptions = "\n".join([f"  - {action.name}: {action.description}" for action in all_actions if action.name in action_names])
    profile = f"Agent Name: {self.name}\nDescription: {self.description}\nAvailable Actions:\n{action_descriptions}"
    return profile

clear_short_term_memory

clear_short_term_memory()

Remove all content from the agent's short-term memory.

Source code in evoagentx/agents/agent.py
def clear_short_term_memory(self):
    """
    Remove all content from the agent's short-term memory.
    """
    pass 

get_prompts

get_prompts() -> dict

Get all the prompts of the agent.

Returns:

Name Type Description
dict dict

A dictionary with keys in the format 'agent_name::action_name' and values containing the system_prompt and action prompt.

Source code in evoagentx/agents/agent.py
def get_prompts(self) -> dict:
    """
    Get all the prompts of the agent.

    Returns:
        dict: A dictionary with keys in the format 'agent_name::action_name' and values
            containing the system_prompt and action prompt.
    """
    prompts = {}
    for action in self.get_all_actions():
        prompts[action.name] = {
            "system_prompt": self.system_prompt, 
            "prompt": action.prompt
        }
    return prompts

set_prompt

set_prompt(action_name: str, prompt: str, system_prompt: Optional[str] = None) -> bool

Set the prompt for a specific action of this agent.

Parameters:

Name Type Description Default
action_name str

Name of the action whose prompt should be updated

required
prompt str

New prompt text to set for the action

required
system_prompt Optional[str]

Optional new system prompt to set for the agent

None

Returns:

Name Type Description
bool bool

True if the prompt was successfully updated, False otherwise

Raises:

Type Description
KeyError

If the action_name does not exist for this agent

Source code in evoagentx/agents/agent.py
def set_prompt(self, action_name: str, prompt: str, system_prompt: Optional[str] = None) -> bool:
    """
    Set the prompt for a specific action of this agent.

    Args:
        action_name: Name of the action whose prompt should be updated
        prompt: New prompt text to set for the action
        system_prompt: Optional new system prompt to set for the agent

    Returns:
        bool: True if the prompt was successfully updated, False otherwise

    Raises:
        KeyError: If the action_name does not exist for this agent
    """
    try:
        action = self.get_action(action_name)
        action.prompt = prompt

        if system_prompt is not None:
            self.system_prompt = system_prompt

        return True
    except KeyError:
        raise KeyError(f"Action '{action_name}' not found in agent '{self.name}'")

set_prompts

set_prompts(prompts: dict) -> bool

Set the prompts for all actions of this agent.

Parameters:

Name Type Description Default
prompts dict

A dictionary with keys in the format 'action_name' and values containing the system_prompt and action prompt.

required

Returns:

Name Type Description
bool bool

True if the prompts were successfully updated, False otherwise

Source code in evoagentx/agents/agent.py
def set_prompts(self, prompts: dict) -> bool:
    """
    Set the prompts for all actions of this agent.

    Args:
        prompts: A dictionary with keys in the format 'action_name' and values
            containing the system_prompt and action prompt.

    Returns:
        bool: True if the prompts were successfully updated, False otherwise
    """
    for action_name, prompt_data in prompts.items():
        # self.set_prompt(action_name, prompt_data["prompt"], prompt_data["system_prompt"])
        if not isinstance(prompt_data, dict):
            raise ValueError(f"Invalid prompt data for action '{action_name}'. Expected a dictionary with 'prompt' and 'system_prompt' (optional) keys.")
        if "prompt" not in prompt_data:
            raise ValueError(f"Missing 'prompt' key in prompt data for action '{action_name}'.")
        self.set_prompt(action_name, prompt_data["prompt"], prompt_data.get("system_prompt", None))
    return True

save_module

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

Save the agent to persistent storage.

Parameters:

Name Type Description Default
path str

Path where the agent should be saved

required
ignore List[str]

List of field names to exclude from serialization

[]
**kwargs Any

Additional parameters for the save operation

{}

Returns:

Type Description
str

The path where the agent was saved

Source code in evoagentx/agents/agent.py
def save_module(self, path: str, ignore: List[str] = [], **kwargs)-> str:
    """Save the agent to persistent storage.

    Args:
        path: Path where the agent should be saved
        ignore: List of field names to exclude from serialization
        **kwargs (Any): Additional parameters for the save operation

    Returns:
        The path where the agent was saved
    """
    ignore_fields = self._save_ignore_fields + ignore
    super().save_module(path=path, ignore=ignore_fields, **kwargs)

from_dict classmethod

from_dict(data: Dict, llm_config: Optional[LLMConfig] = None, **kwargs) -> Agent

Create an agent instance from a dictionary.

Parameters:

Name Type Description Default
data Dict

The dictionary containing all necessary configuration to recreate this agent

required
llm_config Optional[LLMConfig]

The LLMConfig instance

None

Returns:

Name Type Description
Agent Agent

The agent instance

Source code in evoagentx/agents/agent.py
@classmethod
def from_dict(cls, data: Dict, llm_config: Optional[LLMConfig] = None, **kwargs) -> 'Agent':
    """
    Create an agent instance from a dictionary.

    Args:
        data: The dictionary containing all necessary configuration to recreate this agent
        llm_config: The LLMConfig instance

    Returns:
        Agent: The agent instance
    """
    data = add_llm_config_to_agent_dict(data, llm_config)
    agent = cls._create_instance(data)
    return agent 

get_config

get_config() -> dict

Get a dictionary containing all necessary configuration to recreate this agent.

Returns:

Name Type Description
dict dict

A configuration dictionary that can be used to initialize a new Agent instance

dict

with the same properties as this one.

Source code in evoagentx/agents/agent.py
def get_config(self) -> dict:
    """
    Get a dictionary containing all necessary configuration to recreate this agent.

    Returns:
        dict: A configuration dictionary that can be used to initialize a new Agent instance
        with the same properties as this one.
    """
    config = self.to_dict()
    return config

CustomizeAgent

CustomizeAgent(name: str, description: str, prompt: Optional[str] = None, prompt_template: Optional[PromptTemplate] = None, llm_config: Optional[LLMConfig] = None, inputs: Optional[List[Union[dict, Parameter]]] = None, outputs: Optional[List[Union[dict, Parameter]]] = None, system_prompt: Optional[str] = None, output_parser: Optional[Type[ActionOutput]] = None, parse_mode: Optional[str] = 'title', parse_func: Optional[Callable] = None, title_format: Optional[str] = None, tools: Optional[List[Union[Toolkit, Tool]]] = None, custom_output_format: Optional[str] = None, max_steps: int = 20, max_tool_call_concurrency: int = 5, **kwargs)

Bases: Agent

CustomizeAgent provides a flexible framework for creating specialized LLM-powered agents without writing custom code. It enables the creation of agents with well-defined inputs and outputs, custom prompt templates, and configurable parsing strategies.

Attributes:

Name Type Description
name str

The name of the agent.

description str

A description of the agent's purpose and capabilities.

prompt_template PromptTemplate

The prompt template that will be used for the agent's primary action.

prompt str

The prompt template that will be used for the agent's primary action. Should contain placeholders in the format {input_name} for each input parameter.

llm_config LLMConfig

Configuration for the language model.

inputs List[Union[dict, Parameter]]

List of input specifications as dicts or Parameter objects. Each dict (e.g., {"name": str, "type": str, "description": str, ["required": bool, "json_schema": dict]}) contains: - name (str): Name of the input parameter - type (str): Type of the input - description (str): Description of what the input represents - required (bool, optional): Whether this input is required (default: True) - json_schema (dict, optional): The json schema of the input, recommended when type is object or array.

outputs List[Union[dict, Parameter]]

List of output specifications as dicts or Parameter objects. Each dict (e.g., {"name": str, "type": str, "description": str, ["required": bool, "json_schema": dict]}) contains: - name (str): Name of the output field - type (str): Type of the output - description (str): Description of what the output represents - required (bool, optional): Whether this output is required (default: True) - json_schema (dict, optional): The json schema of the output, recommended when type is object or array.

system_prompt str

The system prompt for the LLM. Defaults to DEFAULT_SYSTEM_PROMPT.

output_parser Type[ActionOutput]

A custom class for parsing the LLM's output. Must be a subclass of ActionOutput.

parse_mode str

Mode for parsing LLM output. Options are: - "title": Parse outputs using section titles (default) - "str": Parse as plain text - "json": Parse as JSON - "xml": Parse as XML - "custom": Use a custom parsing function

parse_func Callable

Custom function for parsing LLM output when parse_mode is "custom". Must accept a "content" parameter and return a dictionary.

title_format str

Format string for title parsing mode with {title} placeholder. Default is "## {title}".

tools list[Toolkit]

List of tools to be used by the agent.

custom_output_format str

Specify the output format. Only used when prompt_template is used. If not provided, the output format will be constructed from the outputs specification and parse_mode.

Source code in evoagentx/agents/customize_agent.py
def __init__(
    self,
    name: str,
    description: str,
    prompt: Optional[str] = None,
    prompt_template: Optional[PromptTemplate] = None,
    llm_config: Optional[LLMConfig] = None,
    inputs: Optional[List[Union[dict, Parameter]]] = None,
    outputs: Optional[List[Union[dict, Parameter]]] = None,
    system_prompt: Optional[str] = None,
    output_parser: Optional[Type[ActionOutput]] = None,
    parse_mode: Optional[str] = "title",
    parse_func: Optional[Callable] = None,
    title_format: Optional[str] = None,
    tools: Optional[List[Union[Toolkit, Tool]]] = None,
    custom_output_format: Optional[str] = None,
    max_steps: int = 20,
    max_tool_call_concurrency: int = 5,
    **kwargs
):
    system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT
    inputs = inputs or []
    outputs = outputs or []

    if prompt is not None and prompt_template is not None:
        logger.warning("Both `prompt` and `prompt_template` are provided in `CustomizeAgent`. `prompt_template` will be used.")
        prompt = None

    if isinstance(parse_func, str):
        if not PARSE_FUNCTION_REGISTRY.has_function(parse_func):
            raise ValueError(f"parse function `{parse_func}` is not registered! To instantiate a CustomizeAgent from a file, you should use decorator `@register_parse_function` to register the parse function.")
        parse_func = PARSE_FUNCTION_REGISTRY.get_function(parse_func)

    if isinstance(output_parser, str):
        output_parser = MODULE_REGISTRY.get_module(output_parser)

    # set default title format
    if parse_mode == "title" and title_format is None:
        title_format = "## {title}"

    # validate the data and normalize inputs/outputs to Parameter objects
    valid_inputs, valid_outputs, parse_mode = self.validate_data(
        prompt=prompt,
        prompt_template=prompt_template,
        inputs=inputs,
        outputs=outputs,
        output_parser=output_parser,
        parse_mode=parse_mode,
        parse_func=parse_func,
        title_format=title_format
    )

    customize_action = CustomizeAgent.create_customize_action(
        name=name,
        desc=description,
        prompt=prompt,
        prompt_template=prompt_template,
        inputs=valid_inputs,
        outputs=valid_outputs,
        parse_mode=parse_mode,
        parse_func=parse_func,
        output_parser=output_parser,
        title_format=title_format,
        custom_output_format=custom_output_format,
        tools=tools,
        max_steps=max_steps,
        max_tool_call_concurrency=max_tool_call_concurrency,
    )
    super().__init__(
        name=name,
        description=description,
        llm_config=llm_config,
        system_prompt=system_prompt,
        actions=[customize_action],
        **kwargs
    )

    # Set backing attributes after super().__init__ so Pydantic doesn't wipe them.
    # parse_func must be stored before parse_mode because the parse_mode setter reads it.
    self._inputs  = valid_inputs
    self._outputs = valid_outputs
    self.output_parser = output_parser
    self._parse_func = parse_func
    self.parse_mode = parse_mode
    self._title_format = title_format
    self.custom_output_format = custom_output_format

customize_action_name property

customize_action_name: str

Get the name of the primary custom action for this agent.

Returns:

Type Description
str

The name of the primary custom action

action property

action: Action

Get the primary custom action for this agent.

Returns:

Type Description
Action

The primary custom action

prompt property

prompt: str

Get the prompt for the primary custom action.

Returns:

Type Description
str

The prompt for the primary custom action

prompt_template property

prompt_template: PromptTemplate

Get the prompt template for the primary custom action.

Returns:

Type Description
PromptTemplate

The prompt template for the primary custom action

validate_data

validate_data(prompt: str, prompt_template: PromptTemplate, inputs: List[Union[dict, Parameter]], outputs: List[Union[dict, Parameter]], output_parser: Type[ActionOutput], parse_mode: str, parse_func: Callable, title_format: str) -> tuple

Validate and normalize agent configuration, auto-correcting parse_mode where needed.

Converts inputs and outputs to Parameter objects, validates all parsing-related options, and (for prompt_template-based agents only) auto-corrects parse_mode to "json" when the output schema contains object/array types or a json_schema without a custom parse function. Raw-prompt agents keep their parse_mode, since the prompt is sent verbatim and dictates its own output format.

Returns:

Type Description
tuple

A tuple containing:

tuple
  • valid_inputs (List[Parameter]): normalized input parameters.
tuple
  • valid_outputs (List[Parameter]): normalized output parameters.
tuple
  • parse_mode (str): validated (and possibly auto-corrected) parse mode.
Source code in evoagentx/agents/customize_agent.py
def validate_data(
    self,
    prompt: str,
    prompt_template: PromptTemplate,
    inputs: List[Union[dict, Parameter]],
    outputs: List[Union[dict, Parameter]],
    output_parser: Type[ActionOutput],
    parse_mode: str,
    parse_func: Callable,
    title_format: str,
) -> tuple:
    """Validate and normalize agent configuration, auto-correcting parse_mode where needed.

    Converts `inputs` and `outputs` to `Parameter` objects, validates all
    parsing-related options, and (for `prompt_template`-based agents only)
    auto-corrects `parse_mode` to `"json"` when the output schema contains
    `object`/`array` types or a `json_schema` without a custom parse function.
    Raw-`prompt` agents keep their `parse_mode`, since the prompt is sent verbatim
    and dictates its own output format.

    Returns:
        A tuple containing:
        - `valid_inputs` (List[Parameter]): normalized input parameters.
        - `valid_outputs` (List[Parameter]): normalized output parameters.
        - `parse_mode` (str): validated (and possibly auto-corrected) parse mode.
    """
    # Normalize inputs and outputs to Parameter objects first so all subsequent code
    # can safely use attribute access regardless of what the caller passed in.
    valid_inputs  = self._check_params_types(inputs,  "inputs")  or []
    valid_outputs = self._check_params_types(outputs, "outputs") or []

    # check if the prompt is provided
    if prompt is None and prompt_template is None:
        raise ValueError("`prompt` or `prompt_template` is required when creating a CustomizeAgent.")

    # check if all the inputs are in the prompt (only used when prompt_template is not provided)
    if prompt_template is None and valid_inputs:
        all_input_names = [input_item.name for input_item in valid_inputs]
        inputs_names_not_in_prompt = [name for name in all_input_names if f'{{{name}}}' not in prompt]
        if inputs_names_not_in_prompt:
            raise KeyError(f"The following inputs are not found in the prompt: {inputs_names_not_in_prompt}.")

    # check if the output_parser is valid
    if output_parser is not None:
        self._check_output_parser(valid_outputs, output_parser)

    # check the parse_mode value itself
    if parse_mode not in PARSER_VALID_MODE:
        raise ValueError(f"'{parse_mode}' is an invalid value for `parse_mode`. Available choices: {PARSER_VALID_MODE}.")

    # Auto-correct parse_mode to "json" when outputs require it and no custom parse_func is provided.
    # Only do this for prompt_template-based agents: the template renders the output-format section
    # (and injects the JSON schema) so the model is actually instructed to emit JSON. A raw `prompt`
    # is sent verbatim, so the model follows whatever format the prompt itself specifies; forcing
    # json here would make the parser disagree with the prompt's requested format.
    if prompt_template is not None and CustomizeAgent._outputs_require_json_mode(valid_outputs, parse_func) and parse_mode != "json":
        logger.warning(
            f"parse_mode='{parse_mode}' is not compatible with the current outputs (object/array types or "
            f"json_schema). Auto-correcting to parse_mode='json'. To suppress this warning, explicitly set "
            f"parse_mode='json'."
        )
        return valid_inputs, valid_outputs, "json"

    if parse_mode == "custom" and parse_func is None:
        raise ValueError("`parse_func` (a callable function with an input argument `content`) must be provided when `parse_mode` is 'custom'.")

    CustomizeAgent._validate_parse_func(parse_func)
    CustomizeAgent._validate_title_format(title_format, parse_mode)

    return valid_inputs, valid_outputs, parse_mode

create_customize_action staticmethod

create_customize_action(name: str, desc: str, prompt: str, prompt_template: PromptTemplate, inputs: List[Union[dict, Parameter]], outputs: List[Union[dict, Parameter]], parse_mode: str, parse_func: Optional[Callable] = None, output_parser: Optional[ActionOutput] = None, title_format: Optional[str] = '## {title}', custom_output_format: Optional[str] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, max_steps: int = 20, max_tool_call_concurrency: int = 5, **kwargs) -> Action

Create a custom action based on the provided specifications.

This method dynamically generates an Action class and instance with: - Input parameters defined by the inputs specification - Output format defined by the outputs specification - Custom execution logic using the customize_action_execute function - If tools is provided, returns a CustomizeAction action instead

Parameters:

Name Type Description Default
name str

Base name for the action

required
desc str

Description of the action

required
prompt str

Prompt template for the action

required
prompt_template PromptTemplate

Prompt template for the action

required
inputs List[Union[dict, Parameter]]

List of input field specifications

required
outputs List[Union[dict, Parameter]]

List of output field specifications

required
parse_mode str

Mode to use for parsing LLM output

required
parse_func Optional[Callable]

Optional custom parsing function

None
output_parser Optional[ActionOutput]

Optional custom output parser class

None
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools

None
max_steps int

Maximum number of steps the agent can take

20
max_tool_call_concurrency int

Maximum number of concurrent tool calls

5

Returns:

Type Description
Action

A newly created Action instance

Source code in evoagentx/agents/customize_agent.py
@staticmethod
def create_customize_action(
    name: str,
    desc: str,
    prompt: str,
    prompt_template: PromptTemplate,
    inputs: List[Union[dict, Parameter]],
    outputs: List[Union[dict, Parameter]],
    parse_mode: str,
    parse_func: Optional[Callable] = None,
    output_parser: Optional[ActionOutput] = None,
    title_format: Optional[str] = "## {title}",
    custom_output_format: Optional[str] = None,
    tools: Optional[List[Union[Tool, Toolkit]]] = None,
    max_steps: int = 20,
    max_tool_call_concurrency: int = 5,
    **kwargs
) -> Action:
    """Create a custom action based on the provided specifications.

    This method dynamically generates an Action class and instance with:
    - Input parameters defined by the inputs specification
    - Output format defined by the outputs specification
    - Custom execution logic using the customize_action_execute function
    - If tools is provided, returns a CustomizeAction action instead

    Args:
        name: Base name for the action
        desc: Description of the action
        prompt: Prompt template for the action
        prompt_template: Prompt template for the action
        inputs: List of input field specifications
        outputs: List of output field specifications
        parse_mode: Mode to use for parsing LLM output
        parse_func: Optional custom parsing function
        output_parser: Optional custom output parser class
        tools: Optional list of tools
        max_steps: Maximum number of steps the agent can take
        max_tool_call_concurrency: Maximum number of concurrent tool calls

    Returns:
        A newly created Action instance
    """
    assert prompt is not None or prompt_template is not None, "must provide `prompt` or `prompt_template` when creating CustomizeAgent"

    inputs: List[Parameter] = to_params(inputs)
    outputs: List[Parameter] = to_params(outputs)

    action_input_type = CustomizeAgent.create_action_input(inputs, name)

    if output_parser is None:
        action_output_type = CustomizeAgent.create_action_output(outputs, name)
    else:
        action_output_type = output_parser

    action_cls_name = get_unique_class_name(
        generate_dynamic_class_name(name + " action")
    )

    customize_action = CustomizeAction(
        name=action_cls_name,
        description=desc,
        prompt=prompt,
        prompt_template=prompt_template,
        inputs_format=action_input_type,
        outputs_format=action_output_type,
        parse_mode=parse_mode,
        parse_func=parse_func,
        title_format=title_format,
        custom_output_format=custom_output_format,
        tools=tools,
        max_steps=max_steps,
        max_tool_call_concurrency=max_tool_call_concurrency,
    )

    return customize_action

__call__

__call__(inputs: dict = None, return_msg_type: MessageType = UNKNOWN, **kwargs) -> Message

Call the customize action.

Parameters:

Name Type Description Default
inputs dict

The inputs to the customize action.

None
**kwargs Any

Additional keyword arguments.

{}

Returns:

Name Type Description
ActionOutput Message

The output of the customize action.

Source code in evoagentx/agents/customize_agent.py
def __call__(self, inputs: dict = None, return_msg_type: MessageType = MessageType.UNKNOWN, **kwargs) -> Message:
    """
    Call the customize action.

    Args:
        inputs (dict): The inputs to the customize action.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        ActionOutput: The output of the customize action.
    """
    # return self.execute(action_name=self.customize_action_name, action_input_data=inputs, **kwargs) 
    inputs = inputs or {} 
    return super().__call__(action_name=self.customize_action_name, action_input_data=inputs, return_msg_type=return_msg_type, **kwargs)

get_customize_agent_info

get_customize_agent_info() -> dict

Get the information of the customize agent.

Source code in evoagentx/agents/customize_agent.py
def get_customize_agent_info(self) -> dict:
    """
    Get the information of the customize agent.
    """
    customize_action = self.get_action(self.customize_action_name)

    config = {
        "class_name": "CustomizeAgent",
        "name": self.name,
        "description": self.description,
        "prompt": customize_action.prompt,
        "prompt_template": customize_action.prompt_template.to_dict() if customize_action.prompt_template is not None else None, 
        # "llm_config": self.llm_config.to_dict(exclude_none=True),
        "inputs": [p.to_dict(ignore=["class_name"]) for p in self.inputs] if self.inputs else [],
        "outputs": [p.to_dict(ignore=["class_name"]) for p in self.outputs] if self.outputs else [],
        "system_prompt": self.system_prompt,
        "output_parser": self.output_parser.__name__ if self.output_parser is not None else None,
        "parse_mode": self.parse_mode,
        "parse_func": self.parse_func.__name__ if self.parse_func is not None else None,
        "title_format": self.title_format,
        "tool_names": [tool.name for tool in customize_action.tools] if customize_action.tools else [],
        "custom_output_format": self.custom_output_format,
        "max_steps": customize_action.max_steps,
        "max_tool_call_concurrency": customize_action.max_tool_call_concurrency,
    }
    return config

save_module

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

Save the customize agent's configuration to a JSON file.

Parameters:

Name Type Description Default
path str

File path where the configuration should be saved

required
ignore List[str]

List of keys to exclude from the saved configuration

[]
**kwargs Any

Additional parameters for the save operation

{}

Returns:

Type Description
str

The path where the configuration was saved

Source code in evoagentx/agents/customize_agent.py
def save_module(self, path: str, ignore: List[str] = [], **kwargs)-> str:
    """Save the customize agent's configuration to a JSON file.

    Args:
        path: File path where the configuration should be saved
        ignore: List of keys to exclude from the saved configuration
        **kwargs (Any): Additional parameters for the save operation

    Returns:
        The path where the configuration was saved
    """
    config = self.get_customize_agent_info()

    for ignore_key in ignore:
        config.pop(ignore_key, None)

    # Save to JSON file
    make_parent_folder(path)
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(config, f, indent=4, ensure_ascii=False)

    return path

get_config

get_config() -> dict

Get a dictionary containing all necessary configuration to recreate this agent.

Returns:

Name Type Description
dict dict

A configuration dictionary that can be used to initialize a new Agent instance

dict

with the same properties as this one.

Source code in evoagentx/agents/customize_agent.py
def get_config(self) -> dict:
    """
    Get a dictionary containing all necessary configuration to recreate this agent.

    Returns:
        dict: A configuration dictionary that can be used to initialize a new Agent instance
        with the same properties as this one.
    """
    config = self.get_customize_agent_info()
    config["llm_config"] = self.llm_config.to_dict()
    return config

ActionAgent

ActionAgent(name: str, description: str, inputs: List[dict], outputs: List[dict], execute_func: Callable, async_execute_func: Optional[Callable] = None, llm_config: Optional[LLMConfig] = None, **kwargs)

Bases: Agent

ActionAgent is a specialized agent that executes a provided function directly without LLM. It creates an action that uses the provided function as the execution backbone.

Attributes:

Name Type Description
name str

The name of the agent.

description str

A description of the agent's purpose and capabilities.

inputs List[dict]

List of input specifications, where each dict contains: - name (str): Name of the input parameter - type (str): Type of the input - description (str): Description of what the input represents - required (bool, optional): Whether this input is required (default: True)

outputs List[dict]

List of output specifications, where each dict contains: - name (str): Name of the output field - type (str): Type of the output - description (str): Description of what the output represents - required (bool, optional): Whether this output is required (default: True)

execute_func Callable

The function to execute the agent.

async_execute_func (Callable, Optional)

Async version of the function. If not provided, an async wrapper will be automatically created around execute_func.

llm_config LLMConfig

Configuration for the language model (minimal usage).

Source code in evoagentx/agents/action_agent.py
def __init__(
    self,
    name: str,
    description: str,
    inputs: List[dict],
    outputs: List[dict],
    execute_func: Callable,
    async_execute_func: Optional[Callable] = None,
    llm_config: Optional[LLMConfig] = None,
    **kwargs
):
    # Validate inputs
    if not callable(execute_func):
        raise ValueError("execute_func must be callable")

    if async_execute_func is not None and not callable(async_execute_func):
        raise ValueError("async_execute_func must be callable")

    # Validate inputs and outputs
    self._validate_inputs_outputs(inputs, outputs)

    # Set is_human based on LLM availability
    is_human = llm_config is None

    # Initialize parent directly
    super().__init__(
        name=name,
        description=description,
        llm_config=llm_config,
        is_human=is_human,
        **kwargs
    )

    # Store function references and metadata
    self.execute_func = execute_func
    self.async_execute_func = async_execute_func
    self.inputs = inputs
    self.outputs = outputs

    # Create and add the function-based action
    action = self._create_function_action_with_params(
        name, execute_func, async_execute_func, inputs, outputs
    )
    self.add_action(action)

main_action_name property

main_action_name: str

Get the name of the main function action for this agent.

Returns:

Type Description
str

The name of the main function action

get_config

get_config() -> dict

Get configuration for the ActionAgent.

Source code in evoagentx/agents/action_agent.py
def get_config(self) -> dict:
    """Get configuration for the ActionAgent."""
    # Get base config from Agent
    config = super().get_config()

    # Add ActionAgent-specific information
    config.update({
        "class_name": "ActionAgent",
        "execute_func_name": self.execute_func.__name__ if self.execute_func else None,
        "async_execute_func_name": self.async_execute_func.__name__ if self.async_execute_func else None,
        "inputs": self.inputs,
        "outputs": self.outputs
    })
    return config

save_module

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

Save the ActionAgent configuration to a JSON file.

Parameters:

Name Type Description Default
path str

File path where the configuration should be saved

required
ignore List[str]

List of keys to exclude from the saved configuration

[]
**kwargs Any

Additional parameters for the save operation

{}

Returns:

Type Description
str

The path where the configuration was saved

Source code in evoagentx/agents/action_agent.py
def save_module(self, path: str, ignore: List[str] = [], **kwargs) -> str:
    """Save the ActionAgent configuration to a JSON file.

    Args:
        path: File path where the configuration should be saved
        ignore: List of keys to exclude from the saved configuration
        **kwargs (Any): Additional parameters for the save operation

    Returns:
        The path where the configuration was saved
    """
    config = self.get_config()

    # Add ActionAgent-specific information
    config.update({
        "class_name": "ActionAgent",
        "execute_func_name": self.execute_func.__name__ if self.execute_func else None,
        "async_execute_func_name": self.async_execute_func.__name__ if self.async_execute_func else None,
        "inputs": self.inputs,
        "outputs": self.outputs
    })

    # Remove non-serializable items
    for ignore_key in ignore:
        config.pop(ignore_key, None)

    # Save to JSON file
    make_parent_folder(path)
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(config, f, indent=4, ensure_ascii=False)

    return path

load_module classmethod

load_module(path: str, llm_config: LLMConfig = None, **kwargs) -> ActionAgent

Load the ActionAgent from a JSON file.

Parameters:

Name Type Description Default
path str

The path of the file

required
llm_config LLMConfig

The LLMConfig instance (optional)

None
**kwargs

Additional keyword arguments

{}

Returns:

Name Type Description
ActionAgent ActionAgent

The loaded agent instance

Raises:

Type Description
KeyError

If required functions are not found in the registry

Source code in evoagentx/agents/action_agent.py
@classmethod
def load_module(cls, path: str, llm_config: LLMConfig = None, **kwargs) -> "ActionAgent":
    """Load the ActionAgent from a JSON file.

    Args:
        path: The path of the file
        llm_config: The LLMConfig instance (optional)
        **kwargs: Additional keyword arguments

    Returns:
        ActionAgent: The loaded agent instance

    Raises:
        KeyError: If required functions are not found in the registry
    """
    # Load configuration
    with open(path, 'r', encoding='utf-8') as f:
        config = json.load(f)

    # Extract function names
    execute_func_name = config.get("execute_func_name")
    async_execute_func_name = config.get("async_execute_func_name")

    # Retrieve functions from registry
    execute_func = None
    async_execute_func = None

    if execute_func_name:
        if not ACTION_FUNCTION_REGISTRY.has_function(execute_func_name):
            raise KeyError(f"Function '{execute_func_name}' not found in registry. Please register it first.")
        execute_func = ACTION_FUNCTION_REGISTRY.get_function(execute_func_name)

    if async_execute_func_name:
        if not ACTION_FUNCTION_REGISTRY.has_function(async_execute_func_name):
            raise KeyError(f"Function '{async_execute_func_name}' not found in registry. Please register it first.")
        async_execute_func = ACTION_FUNCTION_REGISTRY.get_function(async_execute_func_name)

    # Create agent
    agent = cls(
        name=config["name"],
        description=config["description"],
        inputs=config["inputs"],
        outputs=config["outputs"],
        execute_func=execute_func,
        async_execute_func=async_execute_func,
        llm_config=llm_config,
        **kwargs
    )

    return agent

__call__

__call__(inputs: dict = None, return_msg_type: MessageType = UNKNOWN, **kwargs) -> Message

Call the main function action.

Parameters:

Name Type Description Default
inputs dict

The inputs to the function action.

None
return_msg_type MessageType

The type of message to return.

UNKNOWN
**kwargs Any

Additional keyword arguments.

{}

Returns:

Name Type Description
Message Message

The output of the function action.

Source code in evoagentx/agents/action_agent.py
def __call__(self, inputs: dict = None, return_msg_type: MessageType = MessageType.UNKNOWN, **kwargs) -> Message:
    """
    Call the main function action.

    Args:
        inputs (dict): The inputs to the function action.
        return_msg_type (MessageType): The type of message to return.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        Message: The output of the function action.
    """
    inputs = inputs or {} 
    return super().__call__(action_name=self.main_action_name, action_input_data=inputs, return_msg_type=return_msg_type, **kwargs)

AgentManager

AgentManager(**kwargs)

Bases: BaseModule

Responsible for creating and managing all Agent objects required for workflow operation.

Attributes:

Name Type Description
storage_handler StorageHandler

Used to load and save agents from/to storage.

agents List[Agent]

A list to keep track of all managed Agent instances.

agent_states Dict[str, AgentState]

A dictionary to track the state of each Agent by name.

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)

size property

size

Get the total number of agents managed by this manager.

check_agents

check_agents()

Validate agent list integrity and state consistency.

Performs thorough validation of the agent manager's internal state: 1. Checks for duplicate agent names 2. Verifies that agent states exist for all agents 3. Ensures agent list and state dictionary sizes match

Source code in evoagentx/agents/agent_manager.py
def check_agents(self):
    """Validate agent list integrity and state consistency.

    Performs thorough validation of the agent manager's internal state:
    1. Checks for duplicate agent names
    2. Verifies that agent states exist for all agents
    3. Ensures agent list and state dictionary sizes match
    """
    # check that the names of self.agents should be unique
    duplicate_agent_names = self.find_duplicate_agents(self.agents)
    if duplicate_agent_names:
        raise ValueError(f"The agents should be unique. Found duplicate agent names: {duplicate_agent_names}!")
    # check agent states
    if len(self.agents) != len(self.agent_states):
        raise ValueError(f"The lengths of self.agents ({len(self.agents)}) and self.agent_states ({len(self.agent_states)}) are different!")
    missing_agents = self.find_missing_agent_states()
    if missing_agents:
        raise ValueError(f"The following agents' states were not found: {missing_agents}")

has_agent

has_agent(agent_name: str) -> bool

Check if an agent with the given name exists in the manager.

Parameters:

Name Type Description Default
agent_name str

The name of the agent to check

required

Returns:

Type Description
bool

True if an agent with the given name exists, False otherwise

Source code in evoagentx/agents/agent_manager.py
def has_agent(self, agent_name: str) -> bool:
    """Check if an agent with the given name exists in the manager.

    Args:
        agent_name: The name of the agent to check

    Returns:
        True if an agent with the given name exists, False otherwise
    """
    all_agent_names = self.list_agents()
    return agent_name in all_agent_names

load_agent

load_agent(agent_name: str, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs) -> Agent

Load an agent from local storage through storage_handler.

Retrieves agent data from storage and creates an Agent instance.

Parameters:

Name Type Description Default
agent_name str

The name of the agent to load

required
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools available for the agent

None
**kwargs Any

Additional parameters for agent creation

{}

Returns:

Type Description
Agent

Agent instance with data loaded from storage

Source code in evoagentx/agents/agent_manager.py
def load_agent(self, agent_name: str, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs) -> Agent:
    """Load an agent from local storage through storage_handler.

    Retrieves agent data from storage and creates an Agent instance.

    Args:
        agent_name: The name of the agent to load
        tools: Optional list of tools available for the agent
        **kwargs (Any): Additional parameters for agent creation

    Returns:
        Agent instance with data loaded from storage
    """
    if not self.storage_handler:
        raise ValueError("must provide ``self.storage_handler`` to use ``load_agent``")
    agent_data = self.storage_handler.load_agent(agent_name=agent_name)
    agent: Agent = create_agent_from_dict(agent_dict=agent_data, tools=tools, **kwargs)
    return agent

load_all_agents

load_all_agents(**kwargs)

Load all agents from storage and add them to the manager.

Retrieves all available agents from storage and adds them to the managed agents collection.

Parameters:

Name Type Description Default
**kwargs Any

Additional parameters passed to storage handler

{}
Source code in evoagentx/agents/agent_manager.py
def load_all_agents(self, **kwargs):
    """Load all agents from storage and add them to the manager.

    Retrieves all available agents from storage and adds them to the
    managed agents collection.

    Args:
        **kwargs (Any): Additional parameters passed to storage handler
    """
    pass 

get_agent_name

get_agent_name(agent: Union[str, dict, Agent]) -> str

Extract agent name from different agent representations.

Handles different ways to specify an agent (string name, dictionary, or Agent instance) and extracts the agent name.

Parameters:

Name Type Description Default
agent Union[str, dict, Agent]

Agent specified as a string name, dictionary with 'name' key, or Agent instance

required

Returns:

Type Description
str

The extracted agent name as a string

Source code in evoagentx/agents/agent_manager.py
def get_agent_name(self, agent: Union[str, dict, Agent]) -> str:
    """Extract agent name from different agent representations.

    Handles different ways to specify an agent (string name, dictionary, or
    Agent instance) and extracts the agent name.

    Args:
        agent: Agent specified as a string name, dictionary with 'name' key,
              or Agent instance

    Returns:
        The extracted agent name as a string
    """
    if isinstance(agent, str):
        agent_name = agent
    elif isinstance(agent, dict):
        agent_name = agent["name"]
    elif isinstance(agent, Agent):
        agent_name = agent.name
    else:
        raise ValueError(f"{type(agent)} is not a supported type for ``get_agent_name``. Supported types: [str, dict, Agent].")
    return agent_name

add_agent

add_agent(agent: Union[str, dict, Agent], llm_config: Optional[LLMConfig] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs)

add a single agent, ignore if the agent already exists (judged by the name of an agent).

Parameters:

Name Type Description Default
agent Union[str, dict, Agent]

The agent to be added, specified as: - String: Agent name to load from storage - Dictionary: Agent specification to create a CustomizeAgent - Agent: Existing Agent instance to add directly

required
llm_config Optional[LLMConfig]

The LLM configuration to be used for the agent. Only used when the agent is a dictionary, used to create a CustomizeAgent.

None
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools available for the agent, used to resolve tool_names in agent config.

None
**kwargs Any

Additional parameters for agent creation

{}
Source code in evoagentx/agents/agent_manager.py
@atomic_method
def add_agent(self, agent: Union[str, dict, Agent], llm_config: Optional[LLMConfig]=None, tools: Optional[List[Union[Tool, Toolkit]]]=None, **kwargs):
    """
    add a single agent, ignore if the agent already exists (judged by the name of an agent).

    Args:
        agent: The agent to be added, specified as:
            - String: Agent name to load from storage
            - Dictionary: Agent specification to create a CustomizeAgent
            - Agent: Existing Agent instance to add directly
        llm_config (Optional[LLMConfig]): The LLM configuration to be used for the agent. Only used when the `agent` is a dictionary, used to create a CustomizeAgent.
        tools: Optional list of tools available for the agent, used to resolve tool_names in agent config.
        **kwargs (Any): Additional parameters for agent creation
    """
    agent_name = self.get_agent_name(agent=agent)
    if self.has_agent(agent_name=agent_name):
        return
    agent_instance = self.create_agent(agent=agent, llm_config=llm_config, tools=tools, **kwargs)
    self.agents.append(agent_instance)
    self.agent_states[agent_instance.name] = AgentState.AVAILABLE
    if agent_instance.name not in self._state_conditions:
        self._state_conditions[agent_instance.name] = threading.Condition()
    self.check_agents()

add_agents

add_agents(agents: List[Union[str, dict, Agent]], llm_config: Optional[LLMConfig] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs)

add several agents by using self.add_agent().

Source code in evoagentx/agents/agent_manager.py
def add_agents(self, agents: List[Union[str, dict, Agent]], llm_config: Optional[LLMConfig]=None, tools: Optional[List[Union[Tool, Toolkit]]]=None, **kwargs):
    """
    add several agents by using self.add_agent().
    """
    for agent in agents:
        self.add_agent(agent=agent, llm_config=llm_config, tools=tools, **kwargs)

add_agents_from_workflow

add_agents_from_workflow(workflow_graph, llm_config: Optional[LLMConfig] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs)

Initialize agents from the nodes of a given WorkFlowGraph and add these agents to self.agents.

Parameters:

Name Type Description Default
workflow_graph WorkFlowGraph

The workflow graph containing nodes with agents information.

required
llm_config Optional[LLMConfig]

The LLM configuration to be used for the agents.

None
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools available for the agents, used to resolve tool_names in agent config.

None
**kwargs Any

Additional parameters passed to add_agent

{}
Source code in evoagentx/agents/agent_manager.py
def add_agents_from_workflow(self, workflow_graph, llm_config: Optional[LLMConfig]=None, tools: Optional[List[Union[Tool, Toolkit]]]=None, **kwargs):
    """
    Initialize agents from the nodes of a given WorkFlowGraph and add these agents to self.agents.

    Args:
        workflow_graph (WorkFlowGraph): The workflow graph containing nodes with agents information.
        llm_config (Optional[LLMConfig]): The LLM configuration to be used for the agents.
        tools: Optional list of tools available for the agents, used to resolve tool_names in agent config.
        **kwargs (Any): Additional parameters passed to add_agent
    """
    from ..workflow.workflow_graph import WorkFlowGraph
    if not isinstance(workflow_graph, WorkFlowGraph):
        raise TypeError("workflow_graph must be an instance of WorkFlowGraph")
    for node in workflow_graph.nodes:
        if node.agents:
            for agent in node.agents:
                self.add_agent(agent=agent, llm_config=llm_config, tools=tools, **kwargs)

update_agents_from_workflow

update_agents_from_workflow(workflow_graph, llm_config: Optional[LLMConfig] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs)

Update agents from a given WorkFlowGraph.

Parameters:

Name Type Description Default
workflow_graph WorkFlowGraph

The workflow graph containing nodes with agents information.

required
llm_config Optional[LLMConfig]

The LLM configuration to be used for the agents.

None
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools available for the agents, used to resolve tool_names in agent config.

None
**kwargs

Additional parameters passed to update_agent

{}
Source code in evoagentx/agents/agent_manager.py
def update_agents_from_workflow(self, workflow_graph, llm_config: Optional[LLMConfig]=None, tools: Optional[List[Union[Tool, Toolkit]]]=None, **kwargs):
    """
    Update agents from a given WorkFlowGraph.

    Args:
        workflow_graph (WorkFlowGraph): The workflow graph containing nodes with agents information.
        llm_config (Optional[LLMConfig]): The LLM configuration to be used for the agents.
        tools: Optional list of tools available for the agents, used to resolve tool_names in agent config.
        **kwargs: Additional parameters passed to update_agent
    """
    from ..workflow.workflow_graph import WorkFlowGraph
    if not isinstance(workflow_graph, WorkFlowGraph):
        raise TypeError("workflow_graph must be an instance of WorkFlowGraph")
    for node in workflow_graph.nodes:
        if node.agents:
            for agent in node.agents:
                agent_name = self.get_agent_name(agent=agent)
                if self.has_agent(agent_name=agent_name):
                    # use the llm_config of the existing agent
                    agent_llm_config = self.get_agent(agent_name).llm_config
                    self.update_agent(agent=agent, llm_config=agent_llm_config, tools=tools, **kwargs)
                else:
                    self.add_agent(agent=agent, llm_config=llm_config, tools=tools, **kwargs)

get_agent

get_agent(agent_name: str, **kwargs) -> Agent

Retrieve an agent by its name from managed agents.

Searches the list of managed agents for an agent with the specified name.

Parameters:

Name Type Description Default
agent_name str

The name of the agent to retrieve

required
**kwargs Any

Additional parameters (unused)

{}

Returns:

Type Description
Agent

The Agent instance with the specified name

Source code in evoagentx/agents/agent_manager.py
def get_agent(self, agent_name: str, **kwargs) -> Agent:
    """Retrieve an agent by its name from managed agents.

    Searches the list of managed agents for an agent with the specified name.

    Args:
        agent_name: The name of the agent to retrieve
        **kwargs (Any): Additional parameters (unused)

    Returns:
        The Agent instance with the specified name
    """
    for agent in self.agents:
        if agent.name == agent_name:
            return agent
    raise ValueError(f"Agent ``{agent_name}`` does not exists!")

update_agent

update_agent(agent: Union[dict, Agent], llm_config: Optional[LLMConfig] = None, tools: Optional[List[Union[Tool, Toolkit]]] = None, **kwargs)

Update an agent in the manager.

Parameters:

Name Type Description Default
agent Union[dict, Agent]

The agent to be updated, specified as: - Dictionary: Agent specification to update a CustomizeAgent - Agent: Existing Agent instance to update

required
llm_config Optional[LLMConfig]

The LLM configuration to be used for the agent.

None
tools Optional[List[Union[Tool, Toolkit]]]

Optional list of tools available for the agent, used to resolve tool_names in agent config.

None
Source code in evoagentx/agents/agent_manager.py
def update_agent(self, agent: Union[dict, Agent], llm_config: Optional[LLMConfig]=None, tools: Optional[List[Union[Tool, Toolkit]]]=None, **kwargs):
    """
    Update an agent in the manager.

    Args:
        agent: The agent to be updated, specified as:
            - Dictionary: Agent specification to update a CustomizeAgent
            - Agent: Existing Agent instance to update
        llm_config (Optional[LLMConfig]): The LLM configuration to be used for the agent.
        tools: Optional list of tools available for the agent, used to resolve tool_names in agent config.
    """
    agent_name = self.get_agent_name(agent=agent)
    self.remove_agent(agent_name=agent_name)
    self.add_agent(agent=agent, llm_config=llm_config, tools=tools, **kwargs)

remove_agent

remove_agent(agent_name: str, remove_from_storage: bool = False, **kwargs)

Remove an agent from the manager and optionally from storage.

Parameters:

Name Type Description Default
agent_name str

The name of the agent to remove

required
remove_from_storage bool

If True, also remove the agent from storage

False
**kwargs Any

Additional parameters passed to storage_handler.remove_agent

{}
Source code in evoagentx/agents/agent_manager.py
@atomic_method
def remove_agent(self, agent_name: str, remove_from_storage: bool=False, **kwargs):
    """
    Remove an agent from the manager and optionally from storage.

    Args:
        agent_name: The name of the agent to remove
        remove_from_storage: If True, also remove the agent from storage
        **kwargs (Any): Additional parameters passed to storage_handler.remove_agent
    """
    self.agents = [agent for agent in self.agents if agent.name != agent_name]
    self.agent_states.pop(agent_name, None)
    self._state_conditions.pop(agent_name, None) 
    if remove_from_storage:
        self.storage_handler.remove_agent(agent_name=agent_name, **kwargs)
    self.check_agents()

get_agent_state

get_agent_state(agent_name: str) -> AgentState

Get the state of a specific agent by its name.

Parameters:

Name Type Description Default
agent_name str

The name of the agent.

required

Returns:

Name Type Description
AgentState AgentState

The current state of the agent.

Source code in evoagentx/agents/agent_manager.py
def get_agent_state(self, agent_name: str) -> AgentState:
    """
    Get the state of a specific agent by its name.

    Args:
        agent_name: The name of the agent.

    Returns:
        AgentState: The current state of the agent.
    """
    return self.agent_states[agent_name]

set_agent_state

set_agent_state(agent_name: str, new_state: AgentState) -> bool

Changes an agent's state and notifies any threads waiting on that agent's state. Thread-safe operation for coordinating multi-threaded agent execution.

Parameters:

Name Type Description Default
agent_name str

The name of the agent

required
new_state AgentState

The new state to set

required

Returns:

Type Description
bool

True if the state was updated successfully, False otherwise

Source code in evoagentx/agents/agent_manager.py
@atomic_method
def set_agent_state(self, agent_name: str, new_state: AgentState) -> bool:
    """
    Changes an agent's state and notifies any threads waiting on that agent's state.
    Thread-safe operation for coordinating multi-threaded agent execution.

    Args:
        agent_name: The name of the agent
        new_state: The new state to set

    Returns:
        True if the state was updated successfully, False otherwise
    """

    # if agent_name in self.agent_states and isinstance(new_state, AgentState):
    #     # self.agent_states[agent_name] = new_state
    #     with self._state_conditions[agent_name]:
    #         self.agent_states[agent_name] = new_state
    #         self._state_conditions[agent_name].notify_all()
    #     self.check_agents()
    #     return True
    # else:
    #     return False
    if agent_name in self.agent_states and isinstance(new_state, AgentState):
        if agent_name not in self._state_conditions:
            self._state_conditions[agent_name] = threading.Condition()
        with self._state_conditions[agent_name]:
            self.agent_states[agent_name] = new_state
            self._state_conditions[agent_name].notify_all()
        return True
    return False

get_all_agent_states

get_all_agent_states() -> Dict[str, AgentState]

Get the states of all managed agents.

Returns:

Type Description
Dict[str, AgentState]

Dict[str, AgentState]: A dictionary mapping agent names to their states.

Source code in evoagentx/agents/agent_manager.py
def get_all_agent_states(self) -> Dict[str, AgentState]:
    """Get the states of all managed agents.

    Returns:
        Dict[str, AgentState]: A dictionary mapping agent names to their states.
    """
    return self.agent_states

save_all_agents

save_all_agents(**kwargs)

Save all managed agents to persistent storage.

Parameters:

Name Type Description Default
**kwargs Any

Additional parameters passed to the storage handler

{}
Source code in evoagentx/agents/agent_manager.py
@atomic_method
def save_all_agents(self, **kwargs):
    """Save all managed agents to persistent storage.

    Args:
        **kwargs (Any): Additional parameters passed to the storage handler
    """
    pass 

clear_agents

clear_agents()

Remove all agents from the manager.

Source code in evoagentx/agents/agent_manager.py
@atomic_method
def clear_agents(self):
    """
    Remove all agents from the manager.
    """
    self.agents = [] 
    self.agent_states = {}
    self._state_conditions = {}
    self.check_agents()

wait_for_agent_available

wait_for_agent_available(agent_name: str, timeout: Optional[float] = None) -> bool

Wait for an agent to be available.

Parameters:

Name Type Description Default
agent_name str

The name of the agent to wait for

required
timeout Optional[float]

Maximum time to wait in seconds, or None to wait indefinitely

None

Returns:

Type Description
bool

True if the agent became available, False if timed out

Source code in evoagentx/agents/agent_manager.py
def wait_for_agent_available(self, agent_name: str, timeout: Optional[float] = None) -> bool:
    """Wait for an agent to be available.

    Args:
        agent_name: The name of the agent to wait for
        timeout: Maximum time to wait in seconds, or None to wait indefinitely

    Returns:
        True if the agent became available, False if timed out
    """
    if agent_name not in self._state_conditions:
        self._state_conditions[agent_name] = threading.Condition()
    condition = self._state_conditions[agent_name]

    with condition:
        return condition.wait_for(
            lambda: self.agent_states.get(agent_name) == AgentState.AVAILABLE,
            timeout=timeout
        )

copy

copy() -> AgentManager

Create a shallow copy of the AgentManager.

Source code in evoagentx/agents/agent_manager.py
def copy(self) -> "AgentManager":
    """
    Create a shallow copy of the AgentManager.
    """
    return AgentManager(agents=self.agents, storage_handler=self.storage_handler)