Skip to content

Chats

Chats exporter.

`ChatsExporter(get_swarm_members: Callable[[WaldiezAgent], Tuple[List[WaldiezAgent], Optional[WaldiezAgent]]], all_agents: List[WaldiezAgent], agent_names: Dict[str, str], all_chats: List[WaldiezChat], chat_names: Dict[str, str], main_chats: List[Tuple[WaldiezChat, WaldiezAgent, WaldiezAgent]], for_notebook: bool, is_async: bool)` ¶

Bases: BaseExporter, ExporterMixin

Chats exporter.

Parameters:

Name	Type	Description	Default
`get_swarm_members`	`Callable[`	`[WaldiezAgent], Tuple[List[WaldiezAgent], Optional[WaldiezAgent]]` ] The function to use to resolve the swarm members.	required
`all_agents`	`List[WaldiezAgent]`	All the agents in the flow.	required
`agent_names`	`Dict[str, str]`	A mapping of agent id to agent name.	required
`all_chats`	`List[WaldiezChat]`	All the chats in the flow.	required
`chat_names`	`Dict[str, str]`	A mapping of chat id to chat name.	required
`main_chats`	`List[Tuple[WaldiezChat, WaldiezAgent, WaldiezAgent]]`	The main chats in the flow.	required
`for_notebook`	`bool`	Whether the export is for a notebook.	required
`is_async`	`bool`	Whether the chat is asynchronous.	required

Source code in waldiez/exporting/chats/chats_exporter.py

def __init__(
    self,
    get_swarm_members: Callable[
        [WaldiezAgent], Tuple[List[WaldiezAgent], Optional[WaldiezAgent]]
    ],
    all_agents: List[WaldiezAgent],
    agent_names: Dict[str, str],
    all_chats: List[WaldiezChat],
    chat_names: Dict[str, str],
    main_chats: List[Tuple[WaldiezChat, WaldiezAgent, WaldiezAgent]],
    for_notebook: bool,
    is_async: bool,
):
    """Initialize the chats exporter.

    Parameters
    ----------
    get_swarm_members : Callable[
            [WaldiezAgent],
            Tuple[List[WaldiezAgent], Optional[WaldiezAgent]]
        ]
        The function to use to resolve the swarm members.
    all_agents : List[WaldiezAgent]
        All the agents in the flow.
    agent_names : Dict[str, str]
        A mapping of agent id to agent name.
    all_chats : List[WaldiezChat]
        All the chats in the flow.
    chat_names : Dict[str, str]
        A mapping of chat id to chat name.
    main_chats : List[Tuple[WaldiezChat, WaldiezAgent, WaldiezAgent]]
        The main chats in the flow.
    for_notebook : bool
        Whether the export is for a notebook.
    is_async : bool
        Whether the chat is asynchronous.
    """
    self.all_agents = all_agents
    self.agent_names = agent_names
    self.main_chats = main_chats
    self.all_chats = all_chats
    self.chat_names = chat_names
    self.get_swarm_members = get_swarm_members
    self.for_notebook = for_notebook
    self.is_async = is_async
    self._chat_string = None
    self._before_chat = None
    self._generated = False

`export() -> ExporterReturnType` ¶

Export the chats.

Returns:

Type	Description
`ExporterReturnType`	The exported chats, the imports, the before export strings, the after export strings, and the environment variables.

Source code in waldiez/exporting/chats/chats_exporter.py

def export(self) -> ExporterReturnType:
    """Export the chats.

    Returns
    -------
    ExporterReturnType
        The exported chats, the imports, the before export strings,
        the after export strings, and the environment variables.
    """
    exported_string = self.generate()
    imports = self.get_imports()
    before_export = self.get_before_export()
    after_export = self.get_after_export()
    return {
        "content": exported_string,
        "imports": imports,
        "before_export": before_export,
        "after_export": after_export,
        "environment_variables": None,
    }

`generate() -> str` ¶

Generate the chats content.

Returns:

Type	Description
`str`	The chats content.

Source code in waldiez/exporting/chats/chats_exporter.py

def generate(self) -> str:
    """Generate the chats content.

    Returns
    -------
    str
        The chats content.
    """
    if self._generated is False:
        self._export_chats()
        self._generated = True
    return self._chat_string or ""

`get_after_export() -> Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]` ¶

Generate the content after the main export.

Returns:

Type	Description
`Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]`	The exported content after the main export and its position.

Source code in waldiez/exporting/chats/chats_exporter.py

def get_after_export(
    self,
) -> Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]:
    """Generate the content after the main export.

    Returns
    -------
    Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]
        The exported content after the main export and its position.
    """
    after: List[Tuple[str, Union[ExportPosition, AgentPosition]]] = []
    # not per agent, we might have references to agents not yet defined.
    # let's use one string for all nested chat registrations
    nested_chat_registrations = ""
    for agent in self.all_agents:
        if agent.agent_type != "swarm":
            registration_string = export_nested_chat_registration(
                agent=agent,
                all_chats=self.all_chats,
                chat_names=self.chat_names,
                agent_names=self.agent_names,
                string_escape=self.string_escape,
                serializer=self.serializer,
                is_async=self.is_async,
            )
            if registration_string:
                nested_chat_registrations += "\n" + registration_string
    if nested_chat_registrations:
        # let's place it before the chats (after all agents are defined)
        after.append(
            (
                nested_chat_registrations,
                AgentPosition(None, AgentPositions.AFTER_ALL, 2),
            )
        )
    return after

`get_before_export() -> Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]` ¶

Generate the content before the main export.

Returns:

Type	Description
`Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]`	The exported content before the main export and its position.

Source code in waldiez/exporting/chats/chats_exporter.py

def get_before_export(
    self,
) -> Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]:
    """Generate the content before the main export.

    Returns
    -------
    Optional[List[Tuple[str, Union[ExportPosition, AgentPosition]]]]
        The exported content before the main export and its position.
    """
    before: List[Tuple[str, Union[ExportPosition, AgentPosition]]] = []
    if self._generated is False:
        self._export_chats()
        self._generated = True
    if self._before_chat:
        before.append(
            (self._before_chat, ExportPosition(ExportPositions.CHATS))
        )
    return before

`get_imports() -> Optional[List[Tuple[str, ImportPosition]]]` ¶

Get the imports string.

Returns:

Type	Description
`str`	The imports string.

Source code in waldiez/exporting/chats/chats_exporter.py

def get_imports(self) -> Optional[List[Tuple[str, ImportPosition]]]:
    """Get the imports string.

    Returns
    -------
    str
        The imports string.
    """
    if len(self.main_chats) == 1:
        _, sender, recipient = self.main_chats[0]
        if sender.agent_type == "swarm" or recipient.agent_type == "swarm":
            import_string = "from autogen import initiate_swarm_chat"
            if self.is_async:
                import_string = "from autogen import a_initiate_swarm_chat"
            return [(import_string, ImportPosition.THIRD_PARTY)]
        # no additional imports, it is `sender.initiate_chat(....)`
        return None
    if self.is_async:
        import_string = (
            "from autogen.agentchat.chat import a_initiate_chats"
        )
    else:
        import_string = "from autogen.agentchat.chat import initiate_chats"
    return [(import_string, ImportPosition.THIRD_PARTY)]