Skip to content

Flow

Waldiez flow model.

WaldiezFlow

Bases: WaldiezBase

Flow data class.

Attributes:

NameTypeDescription
idstr

The ID of the flow.

type(Literal['flow'], optional)

The type of the "node" in a graph: "flow".

namestr

The name of the flow.

descriptionstr

The description of the flow.

tagslist[str]

The tags of the flow.

requirementslist[str]

The requirements of the flow.

storage_idstr

The storage ID of the flow (ignored, UI related).

created_atstr

The date and time when the flow was created.

updated_atstr

The date and time when the flow was last updated.

dataWaldiezFlowData

The data of the flow. See WaldiezFlowData.

cache_seed property

cache_seed: Optional[int]

Check if the flow has caching disabled.

Returns:

TypeDescription
bool

True if the flow has caching disabled, False otherwise.

default classmethod

default() -> WaldiezFlow

Get the default flow.

Returns:

TypeDescription
WaldiezFlow

The default flow.

Source code in waldiez/models/flow/flow.py
@classmethod
def default(cls) -> "WaldiezFlow":
    """Get the default flow.

    Returns
    -------
    WaldiezFlow
        The default flow.
    """
    an_id = get_id()
    return cls(
        id=an_id,
        storage_id=an_id,
        created_at=now(),
        updated_at=now(),
        type="flow",
        name="Default Flow",
        description="Default Flow",
        tags=[],
        requirements=[],
        data=WaldiezFlowData.default(),
    )

get_agent_by_id

get_agent_by_id(agent_id: str) -> WaldiezAgent

Get the agent by ID.

Parameters:

NameTypeDescriptionDefault
agent_idstr

The ID of the agent.

required

Returns:

TypeDescription
WaldiezAgent

The agent.

Raises:

TypeDescription
ValueError

If the agent with the given ID is not found.

Source code in waldiez/models/flow/flow.py
def get_agent_by_id(self, agent_id: str) -> WaldiezAgent:
    """Get the agent by ID.

    Parameters
    ----------
    agent_id : str
        The ID of the agent.

    Returns
    -------
    WaldiezAgent
        The agent.

    Raises
    ------
    ValueError
        If the agent with the given ID is not found.
    """
    for agent in self.data.agents.members:
        if agent.id == agent_id:
            return agent
    raise ValueError(f"Agent with ID {agent_id} not found.")

get_agent_connections

get_agent_connections(
    agent_id: str, all_chats: bool = True
) -> list[str]

Get the agent connections.

Parameters:

NameTypeDescriptionDefault
agent_idstr

The ID of the agent.

required
all_chatsbool

If True, get the connections from all the chats, otherwise get the connections from the ordered flow (main chat flow).

True

Returns:

TypeDescription
list[str]

The list of agent ids that the agent with the given ID connects to.

Source code in waldiez/models/flow/flow.py
def get_agent_connections(
    self, agent_id: str, all_chats: bool = True
) -> list[str]:
    """Get the agent connections.

    Parameters
    ----------
    agent_id : str
        The ID of the agent.
    all_chats : bool, optional
        If True, get the connections from all the chats, otherwise
        get the connections from the ordered flow (main chat flow).

    Returns
    -------
    list[str]
        The list of agent ids that the agent with the given ID connects to.
    """
    connections: list[str] = []
    if all_chats:
        for chat in self.data.chats:
            if chat.source == agent_id:
                connections.append(chat.target)
            if chat.target == agent_id:
                connections.append(chat.source)
    else:
        for entry in self.ordered_flow:
            source = entry["source"]
            target = entry["target"]
            if source.id == agent_id:
                connections.append(target.id)
            if target.id == agent_id:
                connections.append(source.id)
    return connections

get_group_chat_members

get_group_chat_members(
    group_manager_id: str,
) -> list[WaldiezAgent]

Get the group chat members.

Parameters:

NameTypeDescriptionDefault
group_manager_idstr

The ID of the group manager.

required

Returns:

TypeDescription
list[WaldiezAgent]

The list of group chat members.

Source code in waldiez/models/flow/flow.py
def get_group_chat_members(
    self, group_manager_id: str
) -> list[WaldiezAgent]:
    """Get the group chat members.

    Parameters
    ----------
    group_manager_id : str
        The ID of the group manager.

    Returns
    -------
    list[WaldiezAgent]
        The list of group chat members.
    """
    agent = self.get_agent_by_id(group_manager_id)
    if not agent.is_group_manager or not isinstance(
        agent, WaldiezGroupManager
    ):
        return []
    members = [
        agent
        for agent in self.data.agents.members
        if agent.data.parent_id == group_manager_id
    ]
    if agent.data.speakers.selection_method != "round_robin":
        return members
    ordered_ids = agent.get_speakers_order()
    if not ordered_ids:
        return members
    members_dict = {member.id: member for member in members}
    ordered_members = [
        members_dict[member_id]
        for member_id in ordered_ids
        if member_id in members_dict
    ]
    return ordered_members

get_group_members

get_group_members(group_id: str) -> list[WaldiezAgent]

Get the group members.

Parameters:

NameTypeDescriptionDefault
group_idstr

The ID of the group.

required

Returns:

TypeDescription
list[WaldiezAgent]

The list of group members.

Source code in waldiez/models/flow/flow.py
def get_group_members(self, group_id: str) -> list[WaldiezAgent]:
    """Get the group members.

    Parameters
    ----------
    group_id : str
        The ID of the group.

    Returns
    -------
    list[WaldiezAgent]
        The list of group members.
    """
    return [
        agent
        for agent in self.data.agents.members
        if agent.data.parent_id == group_id
    ]

get_root_group_manager

get_root_group_manager() -> WaldiezGroupManager

Get the root group manager.

Returns:

TypeDescription
WaldiezGroupManager

The root group manager.

Raises:

TypeDescription
ValueError

If no group manager is found.

Source code in waldiez/models/flow/flow.py
def get_root_group_manager(self) -> WaldiezGroupManager:
    """Get the root group manager.

    Returns
    -------
    WaldiezGroupManager
        The root group manager.

    Raises
    ------
    ValueError
        If no group manager is found.
    """
    for agent in self.data.agents.groupManagerAgents:
        if agent.data.parent_id is None:
            return agent
    raise ValueError("No group manager found.")

is_async property

is_async: bool

Check if the flow is asynchronous.

Returns:

TypeDescription
bool

True if the flow is asynchronous, False otherwise.

is_single_agent_mode property

is_single_agent_mode: bool

Check if the flow is in single agent mode.

Returns:

TypeDescription
bool

True if the flow is in single agent mode, False otherwise.

ordered_flow property

ordered_flow: list[WaldiezAgentConnection]

Get the ordered flow.

unique_names cached property

unique_names: WaldiezUniqueNames

Get the unique names for the flow.

Returns:

TypeDescription
WaldiezUniqueNames

The unique names for the flow.

validate_flow

validate_flow() -> Self

Flow validation.

  • unique node ids
  • there are at least two agents - (or a single agent but not a group manager)
  • all the agents connect to at least one other agent
  • all the linked agent tools are found in the flow
  • all the linked agent models are found in the flow
  • all the managers have at least one member in the chat group
  • the ordered flow (chats with position >=0) is not empty
  • all agents' code execution config functions exist in the flow tools
  • if group chat flow, there is at least one group manager agent
  • if group chat flow, there is an initial group member agent

Returns:

TypeDescription
WaldiezFlow

The validated flow.

Raises:

TypeDescription
ValueError

If the ordered flow is empty. If the model IDs are not unique. If the tool IDs are not unique. If the agents do not connect to any other node. If the manager's group chat has no members.

Source code in waldiez/models/flow/flow.py
@model_validator(mode="after")
def validate_flow(self) -> Self:
    """Flow validation.

    - unique node ids
    - there are at least two agents
        - (or a single agent but not a group manager)
    - all the agents connect to at least one other agent
    - all the linked agent tools are found in the flow
    - all the linked agent models are found in the flow
    - all the managers have at least one member in the chat group
    - the ordered flow (chats with position >=0) is not empty
    - all agents' code execution config functions exist in the flow tools
    - if group chat flow, there is at least one group manager agent
    - if group chat flow, there is an initial group member agent

    Returns
    -------
    WaldiezFlow
        The validated flow.

    Raises
    ------
    ValueError
        If the ordered flow is empty.
        If the model IDs are not unique.
        If the tool IDs are not unique.
        If the agents do not connect to any other node.
        If the manager's group chat has no members.
    """
    all_members = list(self.data.agents.members)
    all_chats = list(self.data.chats)
    for agent in all_members:
        agent.gather_nested_chats(
            all_agents=all_members, all_chats=all_chats
        )
        agent.gather_handoffs(all_agents=all_members, all_chats=all_chats)
    self._validate_group_chat(all_members)
    if len(all_members) == 1:
        return self._validate_single_agent_mode(all_members[0])
    ordered_flow = self.ordered_flow  # could be empty (if group chat)
    if not ordered_flow and self._ordered_flow is None:
        raise ValueError("The ordered flow is empty.")
    model_ids = self._validate_flow_models()
    tools_ids = self._validate_flow_tools()
    self.data.agents.validate_flow(model_ids, tools_ids)
    self._validate_agent_connections()
    return self

Waldiez flow data.

WaldiezFlowData

Bases: WaldiezBase

Flow data class.

Attributes:

NameTypeDescription
nodeslist[dict[str, Any]]

The nodes of the flow. We ignore this (UI-related)

edgeslist[dict[str, Any]]

The edges of the flow. We ignore this (UI-related)

viewportdict[str, Any]

The viewport of the flow. We ignore this (UI-related)

agentsWaldiezAgents

The agents of the flow: users: list[WaldiezUserProxy] assistants: list[WaldiezAssistant] managers: list[WaldiezGroupManager] rag_users : list[WaldiezRagUserProxy] See WaldiezAgents for more info.

modelslist[WaldiezModel]

The models of the flow. See WaldiezModel.

toolslist[WaldiezTool]

The tools of the flow. See WaldiezTool.

chatslist[WaldiezChat]

The chats of the flow. See WaldiezChat.

is_asyncbool

Whether the flow is asynchronous or not.

cache_seedOptional[int]

The seed for the cache. If None, the seed is not set. Default is 41.

default classmethod

default() -> WaldiezFlowData

Create a default flow data.

Returns:

TypeDescription
WaldiezFlowData

The default flow data.

Source code in waldiez/models/flow/flow_data.py
@classmethod
def default(cls) -> "WaldiezFlowData":
    """Create a default flow data.

    Returns
    -------
    WaldiezFlowData
        The default flow data.
    """
    return cls(
        nodes=[],
        edges=[],
        viewport={},
        agents=WaldiezAgents(
            userProxyAgents=[],
            assistantAgents=[
                WaldiezAssistant(
                    id="assistant1",
                    name="Assistant 1",
                    created_at=now(),
                    updated_at=now(),
                    data=WaldiezAssistantData(
                        termination=WaldiezAgentTerminationMessage()
                    ),
                ),
                WaldiezAssistant(
                    id="assistant2",
                    name="Assistant 2",
                    created_at=now(),
                    updated_at=now(),
                    data=WaldiezAssistantData(
                        # is_multimodal=True,  # we need an api key for this
                        termination=WaldiezAgentTerminationMessage(),
                    ),
                ),
            ],
            ragUserProxyAgents=[],
            reasoningAgents=[],
            captainAgents=[],
        ),
        models=[],
        tools=[],
        chats=[
            WaldiezChat(
                id="chat1",
                type="chat",
                source="assistant1",
                target="assistant2",
                data=WaldiezChatData(
                    name="Chat 1",
                    order=0,
                    position=0,
                    source_type="assistant",
                    target_type="assistant",
                    summary=WaldiezChatSummary(),
                    message=WaldiezChatMessage(
                        type="string",
                        content="Hello, how can I help you?",
                    ),
                    condition=WaldiezDefaultCondition.create(),
                    available=WaldiezTransitionAvailability(),
                    nested_chat=WaldiezChatNested(),
                ),
            ),
            WaldiezChat(
                id="chat2",
                type="chat",
                source="assistant2",
                target="assistant1",
                data=WaldiezChatData(
                    name="Chat 2",
                    order=1,
                    position=1,
                    source_type="assistant",
                    target_type="assistant",
                    summary=WaldiezChatSummary(),
                    message=WaldiezChatMessage(
                        type="string",
                        content="Hello, I need some help.",
                    ),
                    condition=WaldiezDefaultCondition.create(),
                    available=WaldiezTransitionAvailability(),
                    nested_chat=WaldiezChatNested(),
                    prerequisites=["chat1"],
                ),
            ),
        ],
        is_async=False,
        cache_seed=42,
    )

validate_flow_chats

validate_flow_chats() -> Self

Validate the flow chats.

Returns:

TypeDescription
WaldiezFlowData

The flow data.

Raises:

TypeDescription
ValueError

If there is a chat with a prerequisite that does not exist.

Source code in waldiez/models/flow/flow_data.py
@model_validator(mode="after")
def validate_flow_chats(self) -> Self:
    """Validate the flow chats.

    Returns
    -------
    WaldiezFlowData
        The flow data.

    Raises
    ------
    ValueError
        If there is a chat with a prerequisite that does not exist.
    """
    self.chats.sort(key=lambda x: x.order)
    # in async, ag2 uses the "chat_id" field (and it must be an int):
    # ```
    #    prerequisites = []
    #    for chat_info in chat_queue:
    #        if "chat_id" not in chat_info:
    #            raise ValueError(
    #               "Each chat must have a unique id for "
    #               "async multi-chat execution."
    #            )
    #     chat_id = chat_info["chat_id"]
    #     pre_chats = chat_info.get("prerequisites", [])
    #     for pre_chat_id in pre_chats:
    #         if not isinstance(pre_chat_id, int):
    #             raise ValueError("Prerequisite chat id is not int.")
    #         prerequisites.append((chat_id, pre_chat_id))
    #    return prerequisites
    # ```
    id_to_chat_id: dict[str, int] = {}
    for index, chat in enumerate(self.chats):
        id_to_chat_id[chat.id] = index
        chat.set_chat_id(index)
    if not self.is_async:
        return self
    # also update the chat prerequisites (if async)
    #  we have ids(str), not chat_ids(int)
    for chat in self.chats:
        chat_prerequisites: list[int] = []
        for chat_id in chat.data.prerequisites:
            if chat_id not in id_to_chat_id:  # pragma: no cover
                raise ValueError(
                    f"Chat with id {chat_id} not found in the flow."
                )
            chat_prerequisites.append(id_to_chat_id[chat_id])
        chat.set_prerequisites(chat_prerequisites)
    return self

get_flow_data

get_flow_data(
    data: dict[str, Any],
    flow_id: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    tags: Optional[list[str]] = None,
    requirements: Optional[list[str]] = None,
) -> dict[str, Any]

Get the flow from the passed data dict.

Parameters:

NameTypeDescriptionDefault
datadict[str, Any]

The data dict.

required
flow_idOptional[str]

The flow ID, by default None.

None
nameOptional[str]

The flow name, by default None.

None
descriptionOptional[str]

The flow description, by default None.

None
tagsOptional[list[str]]

The flow tags, by default None.

None
requirementsOptional[list[str]]

The flow requirements, by default None.

None

Returns:

TypeDescription
dict[str, Any]

The flow data.

Raises:

TypeDescription
ValueError

If the flow type is not "flow".

Source code in waldiez/models/flow/flow_data.py
def get_flow_data(
    data: dict[str, Any],
    flow_id: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    tags: Optional[list[str]] = None,
    requirements: Optional[list[str]] = None,
) -> dict[str, Any]:
    """Get the flow from the passed data dict.

    Parameters
    ----------
    data : dict[str, Any]
        The data dict.
    flow_id : Optional[str], optional
        The flow ID, by default None.
    name : Optional[str], optional
        The flow name, by default None.
    description : Optional[str], optional
        The flow description, by default None.
    tags : Optional[list[str]], optional
        The flow tags, by default None.
    requirements : Optional[list[str]], optional
        The flow requirements, by default None.

    Returns
    -------
    dict[str, Any]
        The flow data.

    Raises
    ------
    ValueError
        If the flow type is not "flow".
    """
    item_type = data.get("type", "flow")
    if item_type != "flow":
        # empty flow (from exported model/tool ?)
        raise ValueError(f"Invalid flow type: {item_type}")
    from_args: dict[str, Any] = {
        "id": flow_id,
        "name": name,
        "description": description,
        "tags": tags,
        "requirements": requirements,
    }
    for key, value in from_args.items():
        if value:
            data[key] = value
    if "name" not in data:
        data["name"] = "Waldiez Flow"
    if "description" not in data:
        data["description"] = "Waldiez Flow description"
    if "tags" not in data:
        data["tags"] = []
    if "requirements" not in data:
        data["requirements"] = []
    return data