🤖 Agent 接口

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
    )

    # execute action asynchronously
    async_execute_source = inspect.getsource(action.async_execute)
    if "NotImplementedError" in async_execute_source:
        # if the async_execute method is not implemented, use the execute method instead
        execution_results = action.execute(
            llm=self.llm, 
            inputs=action_input_data, 
            sys_msg=self.system_prompt,
            return_prompt=True,
            **kwargs
        )
    else:
        execution_results = await action.async_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

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

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.
    """
    assert self.llm_config or self.llm, "must provide either 'llm_config' or 'llm' when is_human=False"
    if self.llm_config and not self.llm:
        llm_cls = MODEL_REGISTRY.get_model(self.llm_config.llm_type)
        self.llm = llm_cls(config=self.llm_config)
    if self.llm:
        self.llm_config = self.llm.config

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)

load_module classmethod

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

load the agent from local storage. Must provide llm_config when loading the agent from local storage.

Parameters:

Name	Type	Description	Default
path	str	The path of the file	required
llm_config	LLMConfig	The LLMConfig instance	None

Returns:

Name	Type	Description
Agent	Agent	The loaded agent instance

Source code in evoagentx/agents/agent.py

@classmethod
def load_module(cls, path: str, llm_config: LLMConfig = None, **kwargs) -> "Agent":
    """
    load the agent from local storage. Must provide `llm_config` when loading the agent from local storage. 

    Args:
        path: The path of the file
        llm_config: The LLMConfig instance

    Returns:
        Agent: The loaded agent instance
    """
    assert llm_config is not None, "must provide `llm_config` when using `load_module` or `from_file` to load the agent from local storage"
    agent = super().load_module(path=path, **kwargs)
    agent["llm_config"] = llm_config.to_dict()
    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[dict]] = None, outputs: Optional[List[dict]] = 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, max_tool_calls: Optional[int] = 5, custom_output_format: Optional[str] = None, **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[dict]	List of input specifications, where each dict (e.g., {"name": str, "type": str, "description": str, ["required": bool]}) 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 (e.g., {"name": str, "type": str, "description": str, ["required": bool]}) 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)
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.
max_tool_calls	int	Maximum number of tool calls. Defaults to 5.
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[dict]] = None, 
    outputs: Optional[List[dict]] = 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,
    max_tool_calls: Optional[int] = 5,
    custom_output_format: Optional[str] = None, 
    **kwargs
):
    system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT
    inputs = inputs or [] 
    outputs = outputs or [] 
    if tools is not None:
        raw_tool_map = {tool.name: tool for tool in tools}
        tools = [tool if isinstance(tool, Toolkit) else Toolkit(name=tool.name, tools=[tool]) for tool in tools]
    else:
        raw_tool_map = None

    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 
    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 = self.create_customize_action(
        name=name, 
        desc=description, 
        prompt=prompt, 
        prompt_template=prompt_template, 
        inputs=inputs, 
        outputs=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_tool_calls=max_tool_calls
    )
    super().__init__(
        name=name, 
        description=description, 
        llm_config=llm_config, 
        system_prompt=system_prompt, 
        actions=[customize_action], 
        **kwargs
    )
    self._store_inputs_outputs_info(inputs, outputs, raw_tool_map)
    self.output_parser = output_parser 
    self.parse_mode = parse_mode 
    self.parse_func = parse_func 
    self.title_format = title_format
    self.tools = tools
    self.max_tool_calls = max_tool_calls
    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

create_customize_action

create_customize_action(name: str, desc: str, prompt: str, prompt_template: PromptTemplate, inputs: List[dict], outputs: List[dict], 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[Toolkit]] = None, max_tool_calls: Optional[int] = 5) -> 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[dict]	List of input field specifications	required
outputs	List[dict]	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[Toolkit]]	Optional list of tools	None

Returns:

Type	Description
Action	A newly created Action instance

Source code in evoagentx/agents/customize_agent.py

def create_customize_action(
    self, 
    name: str, 
    desc: str, 
    prompt: str, 
    prompt_template: PromptTemplate, 
    inputs: List[dict], 
    outputs: List[dict], 
    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[Toolkit]] = None,
    max_tool_calls: Optional[int] = 5
) -> 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

    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"

    # create the action input type
    action_input_fields = {}
    for field in inputs:
        required = field.get("required", True)
        if required:
            action_input_fields[field["name"]] = (str, Field(description=field["description"]))
        else:
            action_input_fields[field["name"]] = (Optional[str], Field(default=None, description=field["description"]))

    action_input_type = create_model(
        self._get_unique_class_name(
            generate_dynamic_class_name(name+" action_input")
        ),
        **action_input_fields, 
        __base__=ActionInput
    )

    # create the action output type
    if output_parser is None:
        action_output_fields = {}
        for field in outputs:
            required = field.get("required", True)
            if required:
                action_output_fields[field["name"]] = (Any, Field(description=field["description"]))
            else:
                action_output_fields[field["name"]] = (Optional[Any], Field(default=None, description=field["description"]))
        action_output_type = create_model(
            self._get_unique_class_name(
                generate_dynamic_class_name(name+" action_output")
            ),
            **action_output_fields, 
            __base__=ActionOutput,
            # get_content_data=customize_get_content_data,
            # to_str=customize_to_str
        )
    else:
        # self._check_output_parser(outputs, output_parser)
        action_output_type = output_parser

    action_cls_name = self._get_unique_class_name(
        generate_dynamic_class_name(name+" action")
    )

    # Create CustomizeAction-based action with parsing properties only
    customize_action_cls = create_model(
        action_cls_name,
        __base__=CustomizeAction
    )

    customize_action = customize_action_cls(
        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,
        max_tool_try=max_tool_calls,
        tools=tools
    )

    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)
    action_input_params = customize_action.inputs_format.get_attrs()
    action_output_params = customize_action.outputs_format.get_attrs()

    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": [
            {
                "name": field,
                "type": self._action_input_types[field],
                "description": field_info.description,
                "required": self._action_input_required[field]
            }
            for field, field_info in customize_action.inputs_format.model_fields.items() if field in action_input_params
        ],
        "outputs": [
            {
                "name": field,
                "type": self._action_output_types[field],
                "description": field_info.description,
                "required": self._action_output_required[field]
            }
            for field, field_info in customize_action.outputs_format.model_fields.items() if field in action_output_params
        ],
        "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 [],
        "max_tool_calls": self.max_tool_calls,
        "custom_output_format": self.custom_output_format
    }
    return config

load_module classmethod

load_module(path: str, llm_config: LLMConfig = None, tools: List[Union[Toolkit, Tool]] = None, **kwargs) -> CustomizeAgent

load the agent from local storage. Must provide llm_config when loading the agent from local storage. If tools is provided, tool_names must also be provided.

Parameters:

Name	Type	Description	Default
path	str	The path of the file	required
llm_config	LLMConfig	The LLMConfig instance	None
tool_names		List of tool names to be used by the agent. If provided,	required
tool_dict		Dictionary mapping tool names to Tool instances. Required when tool_names is provided.	required

Returns:

Name	Type	Description
CustomizeAgent	CustomizeAgent	The loaded agent instance

Source code in evoagentx/agents/customize_agent.py

@classmethod
def load_module(cls, path: str, llm_config: LLMConfig = None, tools: List[Union[Toolkit, Tool]] = None, **kwargs) -> "CustomizeAgent":
    """
    load the agent from local storage. Must provide `llm_config` when loading the agent from local storage. 
        If tools is provided, tool_names must also be provided. 

    Args:
        path: The path of the file
        llm_config: The LLMConfig instance
        tool_names: List of tool names to be used by the agent. If provided,
        tool_dict: Dictionary mapping tool names to Tool instances. Required when tool_names is provided.

    Returns:
        CustomizeAgent: The loaded agent instance
    """
    match_dict = {}
    agent = super().load_module(path=path, llm_config=llm_config, **kwargs)
    if tools:
        match_dict = {tool.name:tool for tool in tools}
    if agent.get("tool_names", None):
        assert tools is not None, "must provide `tools: List[Union[Toolkit, Tool]]` when using `load_module` or `from_file` to load the agent from local storage and `tool_names` is not None or empty"
        added_tools = [match_dict[tool_name] for tool_name in agent["tool_names"]]
        agent["tools"] = [tool if isinstance(tool, Toolkit) else Toolkit(name=tool.name, tools=[tool]) for tool in added_tools]
    return agent 

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()
    tool_names = config.pop("tool_names", None)
    if tool_names:
        config["tools"] = [self._raw_tool_map[name] for name in tool_names]
    return config

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, **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
**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, **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
        **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 = self.create_customize_agent(agent_data=agent_data)
    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 

update_tools

update_tools(agent_data: dict) -> None

Update agent_data with tools based on tool_names.

Handles four scenarios: 1. Neither tool_names nor tools exist: return directly 2. Only tool_names exists: resolve tool_names to tools and set tools field 3. Only tools exists: return directly (no action needed) 4. Both exist: merge tool_names into existing tools (skip duplicates)

Parameters:

Name	Type	Description	Default
agent_data	dict	Agent configuration dictionary that may contain 'tool_names' and/or 'tools'	required

Raises:

Type	Description
ValueError	If tool_names exist but self.tools is None, or if requested tools are not found

Source code in evoagentx/agents/agent_manager.py

def update_tools(self, agent_data: dict) -> None:
    """
    Update agent_data with tools based on tool_names.

    Handles four scenarios:
    1. Neither tool_names nor tools exist: return directly
    2. Only tool_names exists: resolve tool_names to tools and set tools field
    3. Only tools exists: return directly (no action needed)
    4. Both exist: merge tool_names into existing tools (skip duplicates)

    Args:
        agent_data (dict): Agent configuration dictionary that may contain 'tool_names' and/or 'tools'

    Raises:
        ValueError: If tool_names exist but self.tools is None, or if requested tools are not found
    """
    tool_names = agent_data.get("tool_names", None)
    existing_tools = agent_data.get("tools", None)

    # Case 1: Neither tool_names nor tools exist
    if not tool_names and not existing_tools:
        return

    # Case 3: Only tools exist (no tool_names)
    if not tool_names and existing_tools:
        return

    # For cases 2 and 4: tool_names exists, need to resolve
    if self.tools is None:
        raise ValueError(
            f"Agent requires tools {tool_names}, but no tools are available in AgentManager. "
            f"Please set self.tools before creating agents with tool_names."
        )

    # Create tool mapping from available tools
    tool_mapping = {}
    for tool in self.tools:
        tool_mapping[tool.name] = tool

    # Case 2: Only tool_names exists - initialize empty tools list
    if tool_names and not existing_tools:
        existing_tools = []

    # Case 2 & 4: Process tool_names (either with empty or existing tools list)
    if tool_names:
        # Create a set of existing tool names for quick lookup
        existing_tool_names = {tool.name for tool in existing_tools}

        tools_to_add = []
        missing_tools = []

        for tool_name in tool_names:
            # Skip if tool already exists in tools
            if tool_name in existing_tool_names:
                continue

            # Try to resolve new tool
            if tool_name in tool_mapping:
                tools_to_add.append(tool_mapping[tool_name])
            else:
                missing_tools.append(tool_name)

        if missing_tools:
            available_tools = list(tool_mapping.keys())
            raise ValueError(
                f"The following tools are not available: {missing_tools}. "
                f"Available tools: {available_tools}"
            )

        # Merge new tools with existing ones
        if tools_to_add:
            agent_data["tools"] = list(existing_tools) + tools_to_add

create_customize_agent

create_customize_agent(agent_data: dict, llm_config: Optional[Union[LLMConfig, dict]] = None, **kwargs) -> CustomizeAgent

create a customized agent from the provided agent_data.

Parameters:

Name	Type	Description	Default
agent_data	dict	The data used to create an Agent instance, must contain 'name', 'description' and 'prompt' keys.	required
llm_config	Optional[LLMConfig]	The LLM configuration to be used for the agent. It will be used as the default LLM for agents without a llm_config key. If not provided, the agent_data should contain a llm_config key. If provided and agent_data contains a llm_config key, the llm_config in agent_data will be used.	None
**kwargs	Any	Additional parameters for agent creation	{}

Returns:

Name	Type	Description
Agent	CustomizeAgent	the instantiated agent instance.

Source code in evoagentx/agents/agent_manager.py

def create_customize_agent(self, agent_data: dict, llm_config: Optional[Union[LLMConfig, dict]]=None, **kwargs) -> CustomizeAgent:
    """
    create a customized agent from the provided `agent_data`. 

    Args:
        agent_data: The data used to create an Agent instance, must contain 'name', 'description' and 'prompt' keys.
        llm_config (Optional[LLMConfig]): The LLM configuration to be used for the agent. 
            It will be used as the default LLM for agents without a `llm_config` key. 
            If not provided, the `agent_data` should contain a `llm_config` key. 
            If provided and `agent_data` contains a `llm_config` key, the `llm_config` in `agent_data` will be used.  
        **kwargs (Any): Additional parameters for agent creation

    Returns:
        Agent: the instantiated agent instance.
    """

    agent_data = deepcopy(agent_data)
    agent_llm_config = agent_data.get("llm_config", llm_config)
    if not agent_data.get("is_human", False) and not agent_llm_config:
        raise ValueError("`agent_data` should contain a `llm_config` key or `llm_config` should be provided.")

    if agent_llm_config:
        if isinstance(agent_llm_config, dict):
            agent_data["llm_config"] = agent_llm_config
        elif isinstance(agent_llm_config, LLMConfig):
            agent_data["llm_config"] = agent_llm_config.to_dict()

    # tool_mapping = {}
    # if self.tools is not None:
    #     for tool in self.tools:
    #         tool_mapping[tool.name] = tool
    # if agent_data.get("tool_names", None):
    #     agent_data["tools"] = [tool_mapping[tool_name] for tool_name in agent_data["tool_names"]]
    self.update_tools(agent_data=agent_data) # add `tools` field if needed 
    return CustomizeAgent.from_dict(data=agent_data)

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, **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
**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, **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. 
        **kwargs (Any): Additional parameters for agent creation
    """
    # Check for 'tool' key and convert it to 'tools' if needed
    # if isinstance(agent, dict) and "tool_names" in agent:
    #     tools_mapping = {}
    #     if self.tools is not None:
    #         for tool in self.tools:
    #             tools_mapping[tool.name] = tool
    #     agent["tools"] = [tools_mapping[tool_name] for tool_name in agent["tool_names"]]
    #     agent["tools"] = [tool if isinstance(tool, Toolkit) else Toolkit(name=tool.name, tools=[tool]) for tool in agent["tools"]]

    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, **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, **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, **kwargs):
    """
    add several agents by using self.add_agent().
    """
    for agent in agents:
        self.add_agent(agent=agent, llm_config=llm_config, **kwargs)

add_agents_from_workflow

add_agents_from_workflow(workflow_graph, llm_config: Optional[LLMConfig] = 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
**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, **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.
        **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, **kwargs)

update_agents_from_workflow

update_agents_from_workflow(workflow_graph, llm_config: Optional[LLMConfig] = 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
**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, **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.
        **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, **kwargs)
                else:
                    self.add_agent(agent=agent, llm_config=llm_config, **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, **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

Source code in evoagentx/agents/agent_manager.py

def update_agent(self, agent: Union[dict, Agent], llm_config: Optional[LLMConfig]=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.
    """
    agent_name = self.get_agent_name(agent=agent)
    self.remove_agent(agent_name=agent_name)
    self.add_agent(agent=agent, llm_config=llm_config, **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)