Skip to content

Agent

Base agent class to be inherited by all agents.

WaldiezAgent

Bases: WaldiezBase

Waldiez Agent to be inherited by all other agents.

Attributes:

NameTypeDescription
idstr

The ID of the agent.

typeLiteral['agent']

The type of the "node" in a graph: "agent"

agent_typeWaldiezAgentType

The type of the agent

namestr

The name of the agent.

descriptionstr

The description of the agent.

tagslist[str]

Tags for this agent.

requirementslist[str]

Python requirements for the agent.

created_atstr

The date and time when the agent was created.

updated_atstr

The date and time when the agent was last updated.

dataWaldiezAgentData

The data (properties) of this agent. See waldiez.models.agents.WaldiezAgentData for more info.

Methods:

NameDescription
validate_linked_tools

Validate the tools linked to the agent.

validate_linked_models

Validate the models linked to the agent.

ag2_class property

ag2_class: str

Return the AG2 class of the agent.

ag2_imports property

ag2_imports: set[str]

Return the AG2 imports of the agent.

gather_handoff_ids

gather_handoff_ids(
    group_chats: list[WaldiezChat], nested_chat_id: str
) -> None

Gather all the handoff IDs for this agent.

This method will gather all the handoff IDs from the agent's data, including those that might not be passed in data.handoffs.

Parameters:

NameTypeDescriptionDefault
group_chatslist[WaldiezChat]

The list of group chats that this agent is part of.

required
nested_chat_idstr

The ID of the nested chat to include in handoffs if it exists.

required
Source code in waldiez/models/agents/agent/agent.py
def gather_handoff_ids(
    self,
    group_chats: list["WaldiezChat"],
    nested_chat_id: str,
) -> None:
    """Gather all the handoff IDs for this agent.

    This method will gather all the handoff IDs from the agent's data,
    including those that might not be passed in data.handoffs.

    Parameters
    ----------
    group_chats : list["WaldiezChat"]
        The list of group chats that this agent is part of.
    nested_chat_id : str
        The ID of the nested chat to include in handoffs if it exists.

    """
    existing_handoffs = set(self.data.handoffs)
    has_nested_chat = len(self.data.nested_chats) > 0 and any(
        chat.messages for chat in self.data.nested_chats
    )
    # Step 1: Add missing group chat handoffs
    # These are chats between group members (equivalent to groupEdges)
    for chat in group_chats:
        if chat.id not in existing_handoffs:
            self.data.handoffs.append(chat.id)
            existing_handoffs.add(chat.id)

    # Step 2: Add nested chat if it exists and is not already in handoffs
    if (
        has_nested_chat and nested_chat_id not in existing_handoffs
    ):  # pragma: no branch
        self.data.handoffs.append(nested_chat_id)
        existing_handoffs.add(nested_chat_id)

    # Step 3: Validate all handoffs still exist
    # Remove any handoffs that reference non-existent chats
    valid_chat_ids = {chat.id for chat in group_chats}
    if has_nested_chat:  # pragma: no branch
        valid_chat_ids.add(nested_chat_id)

    # Filter out invalid handoffs
    self.data.handoffs = [
        handoff
        for handoff in self.data.handoffs
        if handoff in valid_chat_ids
    ]

gather_handoffs

gather_handoffs(
    all_agents: list[WaldiezAgent],
    all_chats: list[WaldiezChat],
) -> None

Gather all the handoffs including.

Including ones that might not be passed in data.handoffs.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

The list of all agents in the flow.

required
all_chatslist[WaldiezChat]

The list of all chats in the flow.

required
Source code in waldiez/models/agents/agent/agent.py
def gather_handoffs(
    self,
    all_agents: list["WaldiezAgent"],
    all_chats: list["WaldiezChat"],
) -> None:
    """Gather all the handoffs including.

    Including ones that might not be passed in data.handoffs.

    Parameters
    ----------
    all_agents : list["WaldiezAgent"]
        The list of all agents in the flow.
    all_chats : list["WaldiezChat"]
        The list of all chats in the flow.

    """
    self.gather_nested_chats(all_agents, all_chats)
    if not self.is_group_member or self._checked_handoffs:
        return
    nested_chat_id = "nested-chat"
    self._checked_handoffs = True
    group_chats, group_nested_chats = self._get_agent_chats(
        all_agents, all_chats
    )
    if group_nested_chats:  # pragma: no branch
        self._ensure_one_nested_chat(group_nested_chats)
    self.gather_handoff_ids(
        group_chats=group_chats, nested_chat_id=nested_chat_id
    )
    # generate the actual handoff instances
    for handoff_id in self.data.handoffs:
        if handoff_id == nested_chat_id:
            nested_chat_handoff = self._generate_handoff_from_nested(
                group_nested_chats
            )
            if nested_chat_handoff:  # pragma: no branch
                self._handoffs.append(nested_chat_handoff)
        else:
            chat = next(
                (chat for chat in group_chats if chat.id == handoff_id),
                None,
            )
            if chat:  # pragma: no branch
                self._handoffs.append(chat.as_handoff())

gather_nested_chats

gather_nested_chats(
    all_agents: list[WaldiezAgent],
    all_chats: list[WaldiezChat],
) -> None

Gather the nested chats for the agent.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

All the agents in the flow.

required
all_chatslist[WaldiezChat]

All the chats in the flow.

required
Source code in waldiez/models/agents/agent/agent.py
def gather_nested_chats(
    self,
    all_agents: list["WaldiezAgent"],
    all_chats: list["WaldiezChat"],
) -> None:
    """Gather the nested chats for the agent.

    Parameters
    ----------
    all_agents : list["WaldiezAgent"]
        All the agents in the flow.
    all_chats : list["WaldiezChat"]
        All the chats in the flow.
    """
    if self._checked_nested_chats:
        return
    self._checked_nested_chats = True
    all_chat_ids = {chat.id for chat in all_chats}
    all_agent_ids = {agent.id for agent in all_agents}
    # only use chats that do have messages and "triggered_by"
    nested_chats: list[WaldiezAgentNestedChat] = []
    for chat in self.data.nested_chats:
        if not chat.messages or not chat.triggered_by:  # pragma: no cover
            continue
        # make sure the ids exist
        chat.messages = [
            message
            for message in chat.messages
            if message.id in all_chat_ids
        ]
        chat.triggered_by = [
            agent_id
            for agent_id in chat.triggered_by
            if agent_id in all_agent_ids
        ]
        if chat.messages and chat.triggered_by:  # pragma: no branch
            nested_chats.append(chat)
    self.data.nested_chats = nested_chats

handoffs property

handoffs: list[WaldiezHandoff]

Get the handoffs for this agent.

Returns:

TypeDescription
list[WaldiezHandoff]

The list of handoffs for this agent.

Raises:

TypeDescription
RuntimeError

If handoffs have not been gathered yet.

is_captain property

is_captain: bool

Check if the agent is a captain.

Returns:

TypeDescription
bool

True if the agent is a captain, False otherwise.

is_group_manager property

is_group_manager: bool

Check if the agent is a group manager.

Returns:

TypeDescription
bool

True if the agent is a group manager, False otherwise.

is_group_member property

is_group_member: bool

Check if the agent is a group member.

Returns:

TypeDescription
bool

True if the agent is a group member, False otherwise.

is_rag_user property

is_rag_user: bool

Check if the agent is a RAG user.

Returns:

TypeDescription
bool

True if the agent is a RAG user, False otherwise.

is_reasoning property

is_reasoning: bool

Check if the agent is a reasoning agent.

Returns:

TypeDescription
bool

True if the agent is a reasoning agent, False otherwise.

is_user property

is_user: bool

Check if the agent is a user.

Returns:

TypeDescription
bool

True if the agent is a user, False otherwise.

validate_agent_type classmethod

validate_agent_type(
    v: WaldiezAgentType,
) -> WaldiezAgentType

Validate the agent type.

Parameters:

NameTypeDescriptionDefault
vWaldiezAgentType

The agent type.

required

Returns:

TypeDescription
WaldiezAgentType

The validated agent type.

Raises:

TypeDescription
ValueError

If the agent type is not valid.

Source code in waldiez/models/agents/agent/agent.py
@field_validator("agent_type")
@classmethod
def validate_agent_type(cls, v: WaldiezAgentType) -> WaldiezAgentType:
    """Validate the agent type.

    Parameters
    ----------
    v : WaldiezAgentType
        The agent type.

    Returns
    -------
    WaldiezAgentType
        The validated agent type.

    Raises
    ------
    ValueError
        If the agent type is not valid.
    """

    def _get_warning_message(old_type: str, new_type: str) -> str:
        return (
            f"The agent type '{old_type}' is deprecated. "
            f"Use '{new_type}' instead."
        )

    if v == "user":
        warnings.warn(
            _get_warning_message("user", "user_proxy"),
            DeprecationWarning,
            stacklevel=2,
        )
        return "user_proxy"
    if v == "rag_user":
        warnings.warn(
            _get_warning_message("rag_user", "rag_user_proxy"),
            DeprecationWarning,
            stacklevel=2,
        )
        return "rag_user_proxy"
    return v

validate_code_execution

validate_code_execution(tool_ids: list[str]) -> None

Validate the code execution config.

Parameters:

NameTypeDescriptionDefault
tool_idslist[str]

The list of tool IDs.

required

Raises:

TypeDescription
ValueError

If a function is not found

Source code in waldiez/models/agents/agent/agent.py
def validate_code_execution(self, tool_ids: list[str]) -> None:
    """Validate the code execution config.

    Parameters
    ----------
    tool_ids : list[str]
        The list of tool IDs.

    Raises
    ------
    ValueError
        If a function is not found
    """
    # if the config dict has functions, make sure they can be found
    if isinstance(
        self.data.code_execution_config, WaldiezAgentCodeExecutionConfig
    ):
        for function in self.data.code_execution_config.functions:
            if function not in tool_ids:
                raise ValueError(
                    f"Function '{function}' not found in tools"
                )

validate_linked_models

validate_linked_models(model_ids: list[str]) -> None

Validate the models.

Parameters:

NameTypeDescriptionDefault
model_idsList[str]

The list of model IDs.

required

Raises:

TypeDescription
ValueError

If a model is not found

Source code in waldiez/models/agents/agent/agent.py
def validate_linked_models(self, model_ids: list[str]) -> None:
    """Validate the models.

    Parameters
    ----------
    model_ids : List[str]
        The list of model IDs.

    Raises
    ------
    ValueError
        If a model is not found
    """
    # if the config dict has models, make sure they can be found
    for model in self.data.model_ids:
        if model not in model_ids:
            raise ValueError(
                f"Model '{model}' not found in agent's {self.id} models"
            )

validate_linked_tools

validate_linked_tools(
    tool_ids: list[str], agent_ids: list[str]
) -> None

Validate the tools.

Parameters:

NameTypeDescriptionDefault
tool_idslist[str]

The list of tool IDs.

required
agent_idslist[str]

The list of agent IDs.

required

Raises:

TypeDescription
ValueError

If a tool or agent is not found

Source code in waldiez/models/agents/agent/agent.py
def validate_linked_tools(
    self, tool_ids: list[str], agent_ids: list[str]
) -> None:
    """Validate the tools.

    Parameters
    ----------
    tool_ids : list[str]
        The list of tool IDs.
    agent_ids : list[str]
        The list of agent IDs.

    Raises
    ------
    ValueError
        If a tool or agent is not found
    """
    # if the config dict has tools, make sure they can be found
    for tool in self.data.tools:
        if tool.id not in tool_ids:
            raise ValueError(
                f"Tool '{tool.id}' not found in agent's {self.id} tools"
            )
        if tool.executor_id not in agent_ids:
            raise ValueError(
                f"Agent '{tool.executor_id}' not found in agents"
            )

Common data structures for agents.

WaldiezAgentData

Bases: WaldiezBase

Waldiez Agent Data.

Attributes:

NameTypeDescription
system_messageOptional[str]

The agent's system message. Default: None (depends on the agent's type)

human_input_modeLiteral['ALWAYS', 'NEVER', 'TERMINATE']

The human input mode to use for the agent.

code_execution_configUnion[WaldiezAgentCodeExecutionConfig, False]

The code execution config. Either False (no execution) or a dict.

agent_default_auto_replyOptional[str]

The agent's default auto reply when no input is received.

max_consecutive_auto_replyOptional[int]

The maximum number or consecutive auto replies to use before ending the chat. Default: None (no limit).

terminationWaldiezAgentTerminationMessage

The message termination check to use (keyword, method, none)

model_idsList[str]

The ids of the models to link with the agent.

toolslist[WaldiezAgentLinkedTool]

A list of tools (id and executor) to register.

nested_chatslist[WaldiezAgentNestedChat]

A list of nested chats (triggered_by, messages), to register.

context_variablesOptional[dict[str, Any]]

Context variables that provide a persistent context for the agent. Note: This will be a reference to a shared context for multi-agent chats. Behaves like a dictionary with keys and values (akin to dict[str, Any]).

update_agent_state_before_replylist[str | WaldiezAgentUpdateSystemMessage]

A list of functions, including UpdateSystemMessage, called to update the agent's state before it replies. Each function is called when the agent is selected and before it speaks.

handoffslist[WaldiezAgentHandoff]

A list of handoffs (conditions, targets) to register.

update_context_variables

update_context_variables() -> Self

Update the context variables.

Returns:

TypeDescription
Self

The updated instance of the class.

Source code in waldiez/models/agents/agent/agent_data.py
@model_validator(mode="after")
def update_context_variables(self) -> Self:
    """Update the context variables.

    Returns
    -------
    Self
        The updated instance of the class.
    """
    context_vars = update_dict(self.context_variables)
    self.context_variables = context_vars
    return self

Waldiez Agent types.

WaldiezAgentType module-attribute

WaldiezAgentType = Literal[
    "user_proxy",
    "assistant",
    "group_manager",
    "manager",
    "rag_user",
    "swarm",
    "reasoning",
    "captain",
    "user",
    "rag_user_proxy",
]

Possible types of a Waldiez Agent: - user_proxy, - assistant, - group_manager, - rag_user_proxy, - reasoning, - captain, - swarm (deprecated: do not use it), - user (deprecated: use user_proxy) - rag_user (deprecated: user rag_user_proxy) - manager (deprecated: use group_manager)

Waldiez Agent Code Execution Configuration.

WaldiezAgentCodeExecutionConfig

Bases: WaldiezBase

Waldiez Agent Code Execution Configuration.

Attributes:

NameTypeDescription
work_dirOptional[str]

The working directory for the code execution.

use_dockerOptional[bool]

Run the code in a docker container.

timeoutOptional[float]

The timeout for the code execution. By default None (no timeout).

last_n_messagesOptional[int]

The chat's last n messages to consider for the code execution.

functionsOptional[list[str]]

If not using docker, a list of function ids to use.

Waldiez Agent Tool Model.

WaldiezAgentLinkedTool

Bases: WaldiezBase

Waldiez Agent Linked Tool.

Attributes:

NameTypeDescription
idstr

The id of the tool to use.

executor_idstr

The id of the agent to use that tool.

Waldiez Agent Nested Chat.

WaldiezAgentNestedChat

Bases: WaldiezBase

Waldiez Agent Nested Chat.

Attributes:

NameTypeDescription
triggered_bylist[str]

A list of agent ids that trigger the nested chat.

messageslist[WaldiezAgentNestedChatMessage]

The list of messages (chat ids and 'is_reply'z) to include the in the nested chat registration.

orderint

The order of the nested chat (if used as a handoff). Defaults to 0.

WaldiezAgentNestedChatMessage

Bases: WaldiezBase

Waldiez Agent nested chat message.

A reference to a chat's message or reply in a nested chat

Attributes:

NameTypeDescription
idstr

The id of the chat.

is_replybool

Whether to use the reply in the chat or not.

Waldiez Agent Termination Message Check.

WaldiezAgentTerminationMessage

Bases: WaldiezBase

Waldiez Agent Termination Message Check.

Attributes:

NameTypeDescription
typeLiteral['none', 'keyword', 'method']

The type of the termination check to use: "none", "keyword", "method"

keywordslist[str]

If the type is "keyword", the keywords to search in the message.

criterionOptional[Literal["found", "ending", "exact"]] = None

If the type is "keyword", the criterion to use (e.g.: in, endswith, ==)

method_contentOptional[str]

If the type is "method", the code of the method to use. The method must be called is_termination_message, have one argument (message) which is a dict, and return a bool (whether the message is a termination message or not.)

stringstr

The value of the termination message.

Methods:

NameDescription
validate_termination_message

Validate the termination message configuration.

get_termination_function

get_termination_function(
    name_prefix: Optional[str] = None,
    name_suffix: Optional[str] = None,
) -> tuple[str, str]

Get the termination function.

Parameters:

NameTypeDescriptionDefault
name_prefixstr

The function name prefix.

None
name_suffixstr

The function name suffix.

None

Returns:

TypeDescription
tuple[str, str]

The termination function and the function name.

Source code in waldiez/models/agents/agent/termination_message.py
def get_termination_function(
    self,
    name_prefix: Optional[str] = None,
    name_suffix: Optional[str] = None,
) -> tuple[str, str]:
    """Get the termination function.

    Parameters
    ----------
    name_prefix : str
        The function name prefix.
    name_suffix : str
        The function name suffix.

    Returns
    -------
    tuple[str, str]
        The termination function and the function name.
    """
    function_name = "is_termination_message"
    if name_prefix:
        function_name = f"{name_prefix}_{function_name}"
    if name_suffix:
        function_name = f"{function_name}_{name_suffix}"
    if self.type in ("none", "keyword"):
        return self.string, function_name
    return (
        generate_function(
            function_name=function_name,
            function_args=IS_TERMINATION_MESSAGE_ARGS,
            function_types=IS_TERMINATION_MESSAGE_TYPES,
            function_body=self.string,
        ),
        function_name,
    )

string property

string: str

Get the value of the termination message.

  • If the type is "none", the value is "None".
  • If the type is "keyword", the value is a lambda function that checks if any of the keywords comply with the criterion.
  • If the type is "method", the value is the method content.

Returns:

TypeDescription
str

The value of the termination message.

validate_termination_message

validate_termination_message() -> Self

Validate the termination message configuration.

Raises:

TypeDescription
ValueError

If the configuration is invalid.

Returns:

TypeDescription
WaldiezAgentTerminationMessage

The validated termination message configuration.

Source code in waldiez/models/agents/agent/termination_message.py
@model_validator(mode="after")
def validate_termination_message(self) -> Self:
    """Validate the termination message configuration.

    Raises
    ------
    ValueError
        If the configuration is invalid.

    Returns
    -------
    WaldiezAgentTerminationMessage
        The validated termination message configuration.
    """
    if self.type == "method":
        self._validate_method_content()
    if self.type == "keyword":
        self._validate_keyword()
    if self.type == "none":
        self._string = "None"
    return self