🤖 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}".
`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, 
    custom_output_format: Optional[str] = None, 
    **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 
    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 
    )
    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)
    self.output_parser = output_parser 
    self.parse_mode = parse_mode 
    self.parse_func = parse_func 
    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

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) -> 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

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`

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
) -> 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

    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

    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"]] = (Union[str, dict, list], Field(description=field["description"]))
            else:
                action_output_fields[field["name"]] = (Optional[Union[str, dict, list]], 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")
    )
    customize_action_cls = create_model(
        action_cls_name,
        parse_mode=(Optional[str], Field(default="title", description="the parse mode of the action, must be one of: ['title', 'str', 'json', 'xml', 'custom']")),
        parse_func=(Optional[Callable], Field(default=None, exclude=True, description="the function to parse the LLM output. It receives the LLM output and returns a dict.")),
        title_format=(Optional[str], Field(default="## {title}", exclude=True, description="the format of the title. It is used when the `parse_mode` is 'title'.")),
        custom_output_format=(Optional[str], Field(default=None, exclude=True, description="the format of the output. It is used when the `prompt_template` is provided.")),
        __base__=Action, 
        execute=customize_action_execute,
        async_execute=customize_action_async_execute,
        prepare_action_prompt=prepare_action_prompt,
        prepare_extraction_prompt=prepare_extraction_prompt
    )

    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
    )
    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 
    }
    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 

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 

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()
        else:
            raise ValueError(f"llm_config must be a dictionary or an instance of LLMConfig. Got {type(agent_llm_config)}.") 

    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
    """
    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)

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.

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	`{}`

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	`{}`

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`

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	`{}`

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