Skip to content

Chats

Chats exporter.

ChatsExporter

ChatsExporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    context: Optional[ExporterContext] = None,
    **kwargs: Any
)

Bases: Exporter[ChatExtras]

Chats exporter with structured extras.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

All agents involved in the chats.

required
agent_namesdict[str, str]

Mapping of agent IDs to their names.

required
all_chatslist[WaldiezChat]

All chats to be exported.

required
chat_namesdict[str, str]

Mapping of chat IDs to their names.

required
main_chatslist[WaldiezAgentConnection]

Main chats that are connections between agents.

required
root_group_managerOptional[WaldiezGroupManager]

The root group manager for managing chat groups, if any.

required
contextOptional[ExporterContext]

Exporter context with dependencies, by default None

None
**kwargsAny

Additional keyword arguments for the exporter.

{}
Source code in waldiez/exporting/chats/exporter.py
def __init__(
    self,
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> None:
    """Initialize the chats exporter.

    Parameters
    ----------
    all_agents : list[WaldiezAgent]
        All agents involved in the chats.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    all_chats : list[WaldiezChat]
        All chats to be exported.
    chat_names : dict[str, str]
        Mapping of chat IDs to their names.
    main_chats : list[WaldiezAgentConnection]
        Main chats that are connections between agents.
    root_group_manager : Optional[WaldiezGroupManager]
        The root group manager for managing chat groups, if any.
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments for the exporter.
    """
    super().__init__(context, **kwargs)

    self.all_agents = all_agents
    self.agent_names = agent_names
    self.all_chats = all_chats
    self.chat_names = chat_names
    self.main_chats = main_chats
    self.root_group_manager = root_group_manager
    config = self.context.get_config()
    self.for_notebook = config.for_notebook
    self.is_async = config.is_async
    self.cache_seed = config.cache_seed
    # Initialize extras with processed chat content
    self._extras = self._create_chat_extras()

extras property

extras: ChatExtras

Get the chat extras.

generate_main_content

generate_main_content() -> Optional[str]

Generate the main content of the export.

Source code in waldiez/exporting/chats/exporter.py
def generate_main_content(self) -> Optional[str]:
    """Generate the main content of the export."""
    return None

create_chats_exporter

create_chats_exporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[
        WaldiezGroupManager
    ] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
) -> ChatsExporter

Create a chats exporter.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

All agents involved in the chats.

required
agent_namesdict[str, str]

Mapping of agent IDs to their names.

required
all_chatslist[WaldiezChat]

All chats to be exported.

required
chat_namesdict[str, str]

Mapping of chat IDs to their names.

required
main_chatslist[WaldiezAgentConnection]

Main chats that are connections between agents.

required
root_group_managerOptional[WaldiezGroupManager]

The root group manager for managing chat groups, if any.

None
contextOptional[ExporterContext]

Exporter context with dependencies, by default None

None
**kwargsAny

Additional keyword arguments for the exporter.

{}

Returns:

TypeDescription
ChatsExporter

The created chats exporter.

Source code in waldiez/exporting/chats/factory.py
def create_chats_exporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> ChatsExporter:
    """Create a chats exporter.

    Parameters
    ----------
    all_agents : list[WaldiezAgent]
        All agents involved in the chats.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    all_chats : list[WaldiezChat]
        All chats to be exported.
    chat_names : dict[str, str]
        Mapping of chat IDs to their names.
    main_chats : list[WaldiezAgentConnection]
        Main chats that are connections between agents.
    root_group_manager : Optional[WaldiezGroupManager], optional
        The root group manager for managing chat groups, if any.
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments for the exporter.

    Returns
    -------
    ChatsExporter
        The created chats exporter.
    """
    if context is None:
        context = get_default_exporter_context()
    return ChatsExporter(
        all_agents=all_agents,
        agent_names=agent_names,
        all_chats=all_chats,
        chat_names=chat_names,
        main_chats=main_chats,
        root_group_manager=root_group_manager,
        context=context,
        **kwargs,
    )

exporter

Chats exporter.

ChatsExporter

ChatsExporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    context: Optional[ExporterContext] = None,
    **kwargs: Any
)

Bases: Exporter[ChatExtras]

Chats exporter with structured extras.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

All agents involved in the chats.

required
agent_namesdict[str, str]

Mapping of agent IDs to their names.

required
all_chatslist[WaldiezChat]

All chats to be exported.

required
chat_namesdict[str, str]

Mapping of chat IDs to their names.

required
main_chatslist[WaldiezAgentConnection]

Main chats that are connections between agents.

required
root_group_managerOptional[WaldiezGroupManager]

The root group manager for managing chat groups, if any.

required
contextOptional[ExporterContext]

Exporter context with dependencies, by default None

None
**kwargsAny

Additional keyword arguments for the exporter.

{}
Source code in waldiez/exporting/chats/exporter.py
def __init__(
    self,
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> None:
    """Initialize the chats exporter.

    Parameters
    ----------
    all_agents : list[WaldiezAgent]
        All agents involved in the chats.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    all_chats : list[WaldiezChat]
        All chats to be exported.
    chat_names : dict[str, str]
        Mapping of chat IDs to their names.
    main_chats : list[WaldiezAgentConnection]
        Main chats that are connections between agents.
    root_group_manager : Optional[WaldiezGroupManager]
        The root group manager for managing chat groups, if any.
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments for the exporter.
    """
    super().__init__(context, **kwargs)

    self.all_agents = all_agents
    self.agent_names = agent_names
    self.all_chats = all_chats
    self.chat_names = chat_names
    self.main_chats = main_chats
    self.root_group_manager = root_group_manager
    config = self.context.get_config()
    self.for_notebook = config.for_notebook
    self.is_async = config.is_async
    self.cache_seed = config.cache_seed
    # Initialize extras with processed chat content
    self._extras = self._create_chat_extras()

extras property

extras: ChatExtras

Get the chat extras.

generate_main_content

generate_main_content() -> Optional[str]

Generate the main content of the export.

Source code in waldiez/exporting/chats/exporter.py
def generate_main_content(self) -> Optional[str]:
    """Generate the main content of the export."""
    return None

factory

Factory function for creating a ChatsExporter instance.

create_chats_exporter

create_chats_exporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[
        WaldiezGroupManager
    ] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
) -> ChatsExporter

Create a chats exporter.

Parameters:

NameTypeDescriptionDefault
all_agentslist[WaldiezAgent]

All agents involved in the chats.

required
agent_namesdict[str, str]

Mapping of agent IDs to their names.

required
all_chatslist[WaldiezChat]

All chats to be exported.

required
chat_namesdict[str, str]

Mapping of chat IDs to their names.

required
main_chatslist[WaldiezAgentConnection]

Main chats that are connections between agents.

required
root_group_managerOptional[WaldiezGroupManager]

The root group manager for managing chat groups, if any.

None
contextOptional[ExporterContext]

Exporter context with dependencies, by default None

None
**kwargsAny

Additional keyword arguments for the exporter.

{}

Returns:

TypeDescription
ChatsExporter

The created chats exporter.

Source code in waldiez/exporting/chats/factory.py
def create_chats_exporter(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> ChatsExporter:
    """Create a chats exporter.

    Parameters
    ----------
    all_agents : list[WaldiezAgent]
        All agents involved in the chats.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    all_chats : list[WaldiezChat]
        All chats to be exported.
    chat_names : dict[str, str]
        Mapping of chat IDs to their names.
    main_chats : list[WaldiezAgentConnection]
        Main chats that are connections between agents.
    root_group_manager : Optional[WaldiezGroupManager], optional
        The root group manager for managing chat groups, if any.
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments for the exporter.

    Returns
    -------
    ChatsExporter
        The created chats exporter.
    """
    if context is None:
        context = get_default_exporter_context()
    return ChatsExporter(
        all_agents=all_agents,
        agent_names=agent_names,
        all_chats=all_chats,
        chat_names=chat_names,
        main_chats=main_chats,
        root_group_manager=root_group_manager,
        context=context,
        **kwargs,
    )

processor

Chats processor.

ChatParams dataclass

ChatParams(
    main: list[WaldiezAgentConnection],
    all: list[WaldiezChat],
    names: dict[str, str],
)

Parameters for the chat export processor.

all instance-attribute

All the chats in the flow.

main instance-attribute

main: list[WaldiezAgentConnection]

Main chats that are connections between agents.

names instance-attribute

names: dict[str, str]

Mapping of chat IDs to their names.

ChatsProcessor

ChatsProcessor(
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    for_notebook: bool,
    is_async: bool,
    cache_seed: Optional[int],
    serializer: Serializer,
    extras: ChatExtras,
)

Processor for chats export.

Parameters:

NameTypeDescriptionDefault
extrasChatExtras

The structured extras for the chats export.

required
Source code in waldiez/exporting/chats/processor.py
def __init__(
    self,
    all_agents: list[WaldiezAgent],
    agent_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    main_chats: list[WaldiezAgentConnection],
    root_group_manager: Optional[WaldiezGroupManager],
    for_notebook: bool,
    is_async: bool,
    cache_seed: Optional[int],
    serializer: Serializer,
    extras: ChatExtras,
) -> None:
    """Initialize the chats processor.

    Parameters
    ----------
    extras : ChatExtras
        The structured extras for the chats export.
    """
    self._all_agents = all_agents
    self._agent_names = agent_names
    self._chats = ChatParams(
        main=main_chats,
        all=all_chats,
        names=chat_names,
    )
    self._root_group_manager = root_group_manager
    self._is_async = is_async
    self._for_notebook = for_notebook
    self._cache_seed = cache_seed
    self._serializer = serializer
    self._extras = extras
    chat_tabs = 1
    if cache_seed is not None:
        chat_tabs += 1
    self._chat_tabs = chat_tabs

is_group_patterned

is_group_patterned() -> bool

Check if the chats are group patterned.

Returns:

TypeDescription
bool

True if the chats are group patterned, False otherwise.

Source code in waldiez/exporting/chats/processor.py
def is_group_patterned(self) -> bool:
    """Check if the chats are group patterned.

    Returns
    -------
    bool
        True if the chats are group patterned, False otherwise.
    """
    if len(self._chats.main) == 0 and self._root_group_manager is not None:
        return True
    if len(self._chats.main) == 1:
        main_chat = self._chats.main[0]
        sender = main_chat["source"]
        recipient = main_chat["target"]
        if recipient.is_group_manager or sender.is_group_member:
            return True
    return False

process

process() -> None

Process the chats export.

Source code in waldiez/exporting/chats/processor.py
def process(self) -> None:
    """Process the chats export."""
    self._gather_imports()
    self._handle_chat_registrations()
    chat_initiation = self._generate_chat_initation()
    self._extras.set_chat_initiation(chat_initiation)

utils

Chat utils for exporting chatsper chat type.

common

Common utilities for exporting chats.

get_chat_message_string

get_chat_message_string(
    sender: WaldiezAgent,
    chat: WaldiezChat,
    chat_names: dict[str, str],
) -> tuple[str, Optional[str]]

Get the agent's message as a string.

Parameters:

NameTypeDescriptionDefault
senderWaldiezAgent

The sender.

required
chatWaldiezChat

The chat.

required
chat_namesdict[str, str]

A mapping of chat id to chat name with all the chats in the flow.

required

Returns:

TypeDescription
tuple[str, Optional[str]]

If the message is a string, the message content and None. If the message is a method, the method name and the method content. If the message is None, 'None' and None.

Source code in waldiez/exporting/chats/utils/common.py
def get_chat_message_string(
    sender: WaldiezAgent,
    chat: WaldiezChat,
    chat_names: dict[str, str],
) -> tuple[str, Optional[str]]:
    """Get the agent's message as a string.

    Parameters
    ----------
    sender : WaldiezAgent
        The sender.
    chat : WaldiezChat
        The chat.
    chat_names : dict[str, str]
        A mapping of chat id to chat name with all the chats in the flow.

    Returns
    -------
    tuple[str, Optional[str]]
        If the message is a string, the message content and None.
        If the message is a method, the method name and the method content.
        If the message is None, 'None' and None.
    """
    if not chat.message or chat.message.type == "none":
        return "None", None
    if chat.message.type == "string":
        if chat.message.content is None:  # pragma: no cover
            # should be coverred previousliy on pydantic validation
            return "None", None
        return chat.message.content, None

    is_rag_with_carryover = sender.is_rag_user and chat.message.use_carryover
    chat_name = chat_names[chat.id]
    function_content, function_name = chat.get_message_function(
        name_suffix=chat_name,
        is_rag=is_rag_with_carryover,
    )
    return function_name, function_content

export_group_chats

export_group_chats(
    agent_names: dict[str, str],
    manager: WaldiezGroupManager,
    intial_chat: str | None,
    tabs: int,
    is_async: bool,
) -> str

Get the group chat string.

Parameters:

NameTypeDescriptionDefault
agent_namesdict[str, str]

The agent names.

required
managerWaldiezGroupManager

The group manager agent.

required
intial_chatstr | None

The initial chat to use if any.

required
tabsint

The number of tabs for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required

Returns:

TypeDescription
tuple[str, str]

The group chat string and the import string.

Source code in waldiez/exporting/chats/utils/group.py
def export_group_chats(
    agent_names: dict[str, str],
    manager: WaldiezGroupManager,
    intial_chat: str | None,
    tabs: int,
    is_async: bool,
) -> str:
    """Get the group chat string.

    Parameters
    ----------
    agent_names : dict[str, str]
        The agent names.
    manager : WaldiezGroupManager
        The group manager agent.
    intial_chat : str | None
        The initial chat to use if any.
    tabs : int
        The number of tabs for indentation.
    is_async : bool
        Whether the chat is asynchronous.

    Returns
    -------
    tuple[str, str]
        The group chat string and the import string.
    """
    tab = "    " * tabs
    initiate_group_chat = "initiate_group_chat"
    if is_async:
        initiate_group_chat = "a_initiate_group_chat"
    manager_name = agent_names[manager.id]
    pattern_name = f"{manager_name}_pattern"
    content = f"{tab}results, _, __ = {initiate_group_chat}(" + "\n"
    content += f"{tab}    pattern={pattern_name}," + "\n"
    if intial_chat:
        content += f"{tab}    messages={json.dumps(intial_chat)}," + "\n"
    else:
        content += f'{tab}    messages="",\n'
    content += f"{tab}    max_rounds={manager.data.max_round},\n"
    content += f"{tab})\n"
    return content

export_nested_chat_registration

export_nested_chat_registration(
    agent: WaldiezAgent,
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    is_async: bool,
) -> str

Get the nested chat string.

Parameters:

NameTypeDescriptionDefault
agentWaldiezAgent

The agent.

required
all_chatslist[WaldiezChat]

All the chats in the flow.

required
chat_namesdict[str, str]

The chat names.

required
agent_namesdict[str, str]

The agent names.

required
serializerCallable[..., str]

The serializer to use to escape quotes in a string.

required
is_asyncbool

Whether the chat is asynchronous.

required

Returns:

TypeDescription
str

The nested chat string.

Source code in waldiez/exporting/chats/utils/nested.py
def export_nested_chat_registration(
    agent: WaldiezAgent,
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    is_async: bool,
) -> str:
    """Get the nested chat string.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent.
    all_chats : list[WaldiezChat]
        All the chats in the flow.
    chat_names : dict[str, str]
        The chat names.
    agent_names : dict[str, str]
        The agent names.
    serializer : Callable[..., str]
        The serializer to use to escape quotes in a string.
    is_async : bool
        Whether the chat is asynchronous.

    Returns
    -------
    str
        The nested chat string.
    """
    if not agent.data.nested_chats:
        return ""
    content = ""
    extra_contents: list[str] = []
    agent_name = agent_names[agent.id]
    use_suffix = len(agent.data.nested_chats) > 1
    for index, entry in enumerate(agent.data.nested_chats):
        trigger_names = get_nested_chat_trigger_agent_names(
            nested_chat=entry, agent_names=agent_names
        )
        chat_queue, extra_methods = get_nested_chat_queue(
            nested_chat=entry,
            agent=agent,
            agent_names=agent_names,
            chat_names=chat_names,
            all_chats=all_chats,
            serializer=serializer,
        )
        if not chat_queue:  # pragma: no cover
            continue
        extra_contents.extend(extra_methods)
        var_name = (
            f"{agent_name}_chat_queue_{index}"
            if use_suffix
            else f"{agent_name}_chat_queue"
        )
        content += f"{var_name}: list[dict[str, Any]] = {chat_queue}" + "\n"
        content += f"""
{agent_name}.register_nested_chats(  # pyright: ignore
    trigger={trigger_names},
    chat_queue={var_name},
    use_async={is_async},
    ignore_async_in_sync_chat=True,
)
"""
    functions_string = "\n".join(sorted(extra_contents))
    if functions_string:
        functions_string = functions_string + "\n"
    content = f"{functions_string}{content}"
    return (
        content.replace('"None"', "None")
        .replace("'None'", "None")
        .replace('"False"', "False")
        .replace("'False'", "False")
        .replace("'True'", "True")
        .replace('"True"', "True")
    )

export_sequential_chat

export_sequential_chat(
    main_chats: list[WaldiezAgentConnection],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]

Get the chats content, when there are more than one chats in the flow.

Parameters:

NameTypeDescriptionDefault
main_chatslist[WaldiezAgentConnection]

The main chats.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
serializerCallable[..., str]

The serializer function to escape quotes in a string.

required
tabsint

The number of tabs to use for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required

Returns:

TypeDescription
tuple[str, str]

The main chats content and additional methods string if any.

Example
>>> from waldiez.models import (
...     WaldiezAgent,
...     WaldiezChat,
...     WaldiezChatData,
...     WaldiezChatMessage,
... )
>>> chat1 = WaldiezChat(
...     id="wc-1",
...     name="chat1",
...     description="A chat between two agents.",
...     tags=["chat", "chat1"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-1",
...         recipient="wa-2",
...         position=0,
...         message=WaldiezChatMessage(
...             type="string",
...             content="Hello, how are you?",
...         ),
...     ),
... )
>>> chat2 = WaldiezChat(
...     id="wc-2",
...     name="chat2",
...     description="A chat between two agents.",
...     tags=["chat", "chat2"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-2",
...         recipient="wa-1",
...         position=1,
...         message=WaldiezChatMessage(
...             type="string",
...             content="I am good, thank you. How about you?",
...         ),
...     ),
... )
>>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
>>> chat_names = {"wc-1": "chat1", "wc-2": "chat2"}
>>> serializer = lambda x: x.replace('"', "\"").replace("\n", "\\n")
>>>  export_sequential_chat(
...     main_chats=[(chat1, agent1, agent2), (chat2, agent2, agent1)],
...     chat_names=chat_names,
...     agent_names=agent_names,
...     serializer=serializer,
...     tabs=0,
...     is_async=False,
... )
results = initiate_chats([
    {
        "sender": agent1,
        "recipient": agent2,
        "message": "Hello, how are you?",
    },
    {
        "sender": agent2,
        "recipient": agent1,
        "message": "I am good, thank you. How about you?",
    },
])
Source code in waldiez/exporting/chats/utils/sequential.py
def export_sequential_chat(
    main_chats: list[WaldiezAgentConnection],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]:
    r"""Get the chats content, when there are more than one chats in the flow.

    Parameters
    ----------
    main_chats : list[WaldiezAgentConnection]
        The main chats.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    serializer : Callable[..., str]
        The serializer function to escape quotes in a string.
    tabs : int
        The number of tabs to use for indentation.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.

    Returns
    -------
    tuple[str, str]
        The main chats content and additional methods string if any.

    Example
    -------
    ```python
    >>> from waldiez.models import (
    ...     WaldiezAgent,
    ...     WaldiezChat,
    ...     WaldiezChatData,
    ...     WaldiezChatMessage,
    ... )
    >>> chat1 = WaldiezChat(
    ...     id="wc-1",
    ...     name="chat1",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat1"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-1",
    ...         recipient="wa-2",
    ...         position=0,
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="Hello, how are you?",
    ...         ),
    ...     ),
    ... )
    >>> chat2 = WaldiezChat(
    ...     id="wc-2",
    ...     name="chat2",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat2"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-2",
    ...         recipient="wa-1",
    ...         position=1,
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="I am good, thank you. How about you?",
    ...         ),
    ...     ),
    ... )
    >>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
    >>> chat_names = {"wc-1": "chat1", "wc-2": "chat2"}
    >>> serializer = lambda x: x.replace('"', "\"").replace("\n", "\\n")
    >>>  export_sequential_chat(
    ...     main_chats=[(chat1, agent1, agent2), (chat2, agent2, agent1)],
    ...     chat_names=chat_names,
    ...     agent_names=agent_names,
    ...     serializer=serializer,
    ...     tabs=0,
    ...     is_async=False,
    ... )
    results = initiate_chats([
        {
            "sender": agent1,
            "recipient": agent2,
            "message": "Hello, how are you?",
        },
        {
            "sender": agent2,
            "recipient": agent1,
            "message": "I am good, thank you. How about you?",
        },
    ])
    ```
    """
    tab = "    " * tabs if tabs > 0 else ""
    content = "\n"
    additional_methods_string = ""
    content += _get_initiate_chats_line(tab, is_async)
    for connection in main_chats:
        chat_string, additional_methods = _get_chat_dict_string(
            chat_names=chat_names,
            connection=connection,
            agent_names=agent_names,
            serializer=serializer,
            tabs=tabs + 1,
            skip_cache=skip_cache,
        )
        additional_methods_string += additional_methods
        content += "\n" + f"{tab}    {chat_string}"
    content += "\n" + "    " * tabs + "])\n"
    return content, additional_methods_string

export_single_chat

export_single_chat(
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    chat: WaldiezChat,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[[str], str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
    tab_leng: int = 4,
) -> tuple[str, str]

Get the chat string when there is only one chat in the flow.

Parameters:

NameTypeDescriptionDefault
senderWaldiezAgent

The sender.

required
recipientWaldiezAgent

The recipient.

required
chatWaldiezChat

The chat.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
serializerCallable[[str], str]

The serializer function to escape quotes in a string.

required
tabsint

The number of tabs to use for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required
tab_lengint

The length of the tab string, by default 4.

4

Returns:

TypeDescription
tuple[str, str]

The chat string and additional methods string if any

Example
>>> from waldiez.models import WaldiezAgent, WaldiezChat, WaldiezChatData, WaldiezChatMessage
>>> chat = WaldiezChat(
...     id="wc-1",
...     name="chat1",
...     description="A chat between two agents.",
...     tags=["chat", "chat1"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-1",
...         recipient="wa-2",
...         message=WaldiezChatMessage(
...             type="string",
...             content="Hello, how are you?",
...         ),
...     ),
... )
>>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
>>> chat_names = {"wc-1": "chat1"}
>>> export_single_chat_string(
...     sender=agent1,
...     recipient=agent2,
...     chat=chat,
...     agent_names=agent_names,
...     chat_names=chat_names,
...     tabs=0,
... )
agent1.initiate_chat(
    agent2,
    message="Hello, how are you?",
    cache=cache,
)
Source code in waldiez/exporting/chats/utils/single.py
def export_single_chat(
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    chat: WaldiezChat,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[[str], str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
    tab_leng: int = 4,
) -> tuple[str, str]:
    """Get the chat string when there is only one chat in the flow.

    Parameters
    ----------
    sender : WaldiezAgent
        The sender.
    recipient : WaldiezAgent
        The recipient.
    chat : WaldiezChat
        The chat.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    serializer : Callable[[str], str]
        The serializer function to escape quotes in a string.
    tabs : int
        The number of tabs to use for indentation.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.
    tab_leng : int, optional
        The length of the tab string, by default 4.

    Returns
    -------
    tuple[str, str]
        The chat string and additional methods string if any

    Example
    -------
    ```python
    >>> from waldiez.models import WaldiezAgent, WaldiezChat, WaldiezChatData, WaldiezChatMessage
    >>> chat = WaldiezChat(
    ...     id="wc-1",
    ...     name="chat1",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat1"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-1",
    ...         recipient="wa-2",
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="Hello, how are you?",
    ...         ),
    ...     ),
    ... )
    >>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
    >>> chat_names = {"wc-1": "chat1"}
    >>> export_single_chat_string(
    ...     sender=agent1,
    ...     recipient=agent2,
    ...     chat=chat,
    ...     agent_names=agent_names,
    ...     chat_names=chat_names,
    ...     tabs=0,
    ... )
    agent1.initiate_chat(
        agent2,
        message="Hello, how are you?",
        cache=cache,
    )
    ```
    """
    tab = " " * tab_leng * tabs if tabs > 0 else ""
    chat_args = chat.get_chat_args(for_queue=False, sender=sender)
    # chat_args = update_summary_chat_args(chat_args, skip_summary=True)
    if not chat_args:
        return get_empty_simple_chat_string(
            chat=chat,
            sender=sender,
            recipient=recipient,
            agent_names=agent_names,
            tab=tab,
            is_async=is_async,
            skip_cache=skip_cache,
        )
    return get_simple_chat_string(
        chat=chat,
        chat_args=chat_args,
        sender=sender,
        recipient=recipient,
        agent_names=agent_names,
        chat_names=chat_names,
        serializer=serializer,
        tabs=tabs,
        is_async=is_async,
        skip_cache=skip_cache,
    )

get_nested_chat_queue

get_nested_chat_queue(
    nested_chat: WaldiezAgentNestedChat,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> tuple[str, list[str]]

Get the nested chat queue.

Parameters:

NameTypeDescriptionDefault
nested_chatWaldiezAgentNestedChat

The nested chat.

required
agentWaldiezAgent

The agent.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
all_chatslist[WaldiezChat]

All the chats in the flow.

required
serializerCallable[..., str]

The serializer to use to escape quotes in a string.

required

Returns:

TypeDescription
tuple[str, list[str]]

The nested chat queue and the methods to include (methods: message string and method name if the message is a method).

Source code in waldiez/exporting/chats/utils/nested.py
def get_nested_chat_queue(
    nested_chat: WaldiezAgentNestedChat,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> tuple[str, list[str]]:
    """Get the nested chat queue.

    Parameters
    ----------
    nested_chat : WaldiezAgentNestedChat
        The nested chat.
    agent : WaldiezAgent
        The agent.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    all_chats : list[WaldiezChat]
        All the chats in the flow.
    serializer : Callable[..., str]
        The serializer to use to escape quotes in a string.

    Returns
    -------
    tuple[str, list[str]]
        The nested chat queue and the methods to include
        (methods: message string and method name if the message is a method).
    """
    message_methods_to_include: list[str] = []
    chat_messages_str = "[\n"
    for message in nested_chat.messages:
        waldiez_chat = next(chat for chat in all_chats if chat.id == message.id)
        message_str, message_source = get_nested_chat_message_string(
            waldiez_chat=waldiez_chat,
            message=message,
            agent=agent,
            agent_names=agent_names,
            chat_names=chat_names,
            serializer=serializer,
        )
        if message_source:
            message_methods_to_include.append(message_source)
        chat_messages_str += f"    {message_str}," + "\n"
    chat_messages_str += "]"
    if chat_messages_str == "[\n]":
        return "", message_methods_to_include
    return chat_messages_str, message_methods_to_include

group

Exporting group chat utils.

Using the group patterns (no group manager agent)

export_group_chats

export_group_chats(
    agent_names: dict[str, str],
    manager: WaldiezGroupManager,
    intial_chat: str | None,
    tabs: int,
    is_async: bool,
) -> str

Get the group chat string.

Parameters:

NameTypeDescriptionDefault
agent_namesdict[str, str]

The agent names.

required
managerWaldiezGroupManager

The group manager agent.

required
intial_chatstr | None

The initial chat to use if any.

required
tabsint

The number of tabs for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required

Returns:

TypeDescription
tuple[str, str]

The group chat string and the import string.

Source code in waldiez/exporting/chats/utils/group.py
def export_group_chats(
    agent_names: dict[str, str],
    manager: WaldiezGroupManager,
    intial_chat: str | None,
    tabs: int,
    is_async: bool,
) -> str:
    """Get the group chat string.

    Parameters
    ----------
    agent_names : dict[str, str]
        The agent names.
    manager : WaldiezGroupManager
        The group manager agent.
    intial_chat : str | None
        The initial chat to use if any.
    tabs : int
        The number of tabs for indentation.
    is_async : bool
        Whether the chat is asynchronous.

    Returns
    -------
    tuple[str, str]
        The group chat string and the import string.
    """
    tab = "    " * tabs
    initiate_group_chat = "initiate_group_chat"
    if is_async:
        initiate_group_chat = "a_initiate_group_chat"
    manager_name = agent_names[manager.id]
    pattern_name = f"{manager_name}_pattern"
    content = f"{tab}results, _, __ = {initiate_group_chat}(" + "\n"
    content += f"{tab}    pattern={pattern_name}," + "\n"
    if intial_chat:
        content += f"{tab}    messages={json.dumps(intial_chat)}," + "\n"
    else:
        content += f'{tab}    messages="",\n'
    content += f"{tab}    max_rounds={manager.data.max_round},\n"
    content += f"{tab})\n"
    return content

nested

Nested chats exporting.

export_nested_chat_registration

export_nested_chat_registration(
    agent: WaldiezAgent,
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    is_async: bool,
) -> str

Get the nested chat string.

Parameters:

NameTypeDescriptionDefault
agentWaldiezAgent

The agent.

required
all_chatslist[WaldiezChat]

All the chats in the flow.

required
chat_namesdict[str, str]

The chat names.

required
agent_namesdict[str, str]

The agent names.

required
serializerCallable[..., str]

The serializer to use to escape quotes in a string.

required
is_asyncbool

Whether the chat is asynchronous.

required

Returns:

TypeDescription
str

The nested chat string.

Source code in waldiez/exporting/chats/utils/nested.py
def export_nested_chat_registration(
    agent: WaldiezAgent,
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    is_async: bool,
) -> str:
    """Get the nested chat string.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent.
    all_chats : list[WaldiezChat]
        All the chats in the flow.
    chat_names : dict[str, str]
        The chat names.
    agent_names : dict[str, str]
        The agent names.
    serializer : Callable[..., str]
        The serializer to use to escape quotes in a string.
    is_async : bool
        Whether the chat is asynchronous.

    Returns
    -------
    str
        The nested chat string.
    """
    if not agent.data.nested_chats:
        return ""
    content = ""
    extra_contents: list[str] = []
    agent_name = agent_names[agent.id]
    use_suffix = len(agent.data.nested_chats) > 1
    for index, entry in enumerate(agent.data.nested_chats):
        trigger_names = get_nested_chat_trigger_agent_names(
            nested_chat=entry, agent_names=agent_names
        )
        chat_queue, extra_methods = get_nested_chat_queue(
            nested_chat=entry,
            agent=agent,
            agent_names=agent_names,
            chat_names=chat_names,
            all_chats=all_chats,
            serializer=serializer,
        )
        if not chat_queue:  # pragma: no cover
            continue
        extra_contents.extend(extra_methods)
        var_name = (
            f"{agent_name}_chat_queue_{index}"
            if use_suffix
            else f"{agent_name}_chat_queue"
        )
        content += f"{var_name}: list[dict[str, Any]] = {chat_queue}" + "\n"
        content += f"""
{agent_name}.register_nested_chats(  # pyright: ignore
    trigger={trigger_names},
    chat_queue={var_name},
    use_async={is_async},
    ignore_async_in_sync_chat=True,
)
"""
    functions_string = "\n".join(sorted(extra_contents))
    if functions_string:
        functions_string = functions_string + "\n"
    content = f"{functions_string}{content}"
    return (
        content.replace('"None"', "None")
        .replace("'None'", "None")
        .replace('"False"', "False")
        .replace("'False'", "False")
        .replace("'True'", "True")
        .replace('"True"', "True")
    )

get_chat_nested_string

get_chat_nested_string(
    chat: WaldiezChat,
    is_reply: bool,
    chat_names: dict[str, str],
) -> tuple[str, Optional[str]]

Get the nested chat message.

Parameters:

NameTypeDescriptionDefault
chatWaldiezChat

The chat.

required
is_replybool

Whether to use the nested chat's reply message or not.

required
chat_namesdict[str, str]

A mapping of chat id to chat name..

required

Returns:

TypeDescription
tuple[str, Optional[str]]

If the message is a string, the message content and None. If the message is a method, the method name and the method content. If the message is None, 'None' and None.

Source code in waldiez/exporting/chats/utils/nested.py
def get_chat_nested_string(
    chat: WaldiezChat,
    is_reply: bool,
    chat_names: dict[str, str],
) -> tuple[str, Optional[str]]:
    """Get the nested chat message.

    Parameters
    ----------
    chat : WaldiezChat
        The chat.
    is_reply : bool
        Whether to use the nested chat's reply message or not.
    chat_names : dict[str, str]
        A mapping of chat id to chat name..

    Returns
    -------
    tuple[str, Optional[str]]
        If the message is a string, the message content and None.
        If the message is a method, the method name and the method content.
        If the message is None, 'None' and None.
    """
    message = (
        chat.data.nested_chat.reply
        if is_reply
        else chat.data.nested_chat.message
    )
    if not message or message.type == "none" or message.content is None:
        return "None", None
    if message.type == "string":
        return message.content, None
    chat_name = chat_names[chat.id]
    if is_reply:
        function_content, function_name = chat.get_nested_chat_message_function(
            name_suffix=chat_name
        )
    else:
        function_content, function_name = chat.get_nested_chat_message_function(
            name_suffix=chat_name
        )
    return function_name, function_content

get_nested_chat_message_string

get_nested_chat_message_string(
    waldiez_chat: WaldiezChat,
    message: WaldiezAgentNestedChatMessage,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
) -> tuple[str, Optional[str]]

Get the nested chat message string.

Parameters:

NameTypeDescriptionDefault
waldiez_chatWaldiezChat

The chat.

required
messageWaldiezAgentNestedChatMessage

The message.

required
agentWaldiezAgent

The agent.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
serializerCallable[..., str]

The function to serialize the chat arguments.

required

Returns:

TypeDescription
tuple[str, Optional[str]]

The message string and the method name if the message is a method.

Source code in waldiez/exporting/chats/utils/nested.py
def get_nested_chat_message_string(
    waldiez_chat: WaldiezChat,
    message: WaldiezAgentNestedChatMessage,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
) -> tuple[str, Optional[str]]:
    """Get the nested chat message string.

    Parameters
    ----------
    waldiez_chat : WaldiezChat
        The chat.
    message : WaldiezAgentNestedChatMessage
        The message.
    agent : WaldiezAgent
        The agent.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    serializer : Callable[..., str]
        The function to serialize the chat arguments.

    Returns
    -------
    tuple[str, Optional[str]]
        The message string and the method name if the message is a method.
    """
    sender_name: Optional[str] = None
    sender_id = waldiez_chat.target if message.is_reply else waldiez_chat.source
    recipient_id = (
        waldiez_chat.source if message.is_reply else waldiez_chat.target
    )
    if sender_id != agent.id:
        sender_name = agent_names[sender_id]
    recipient_name = agent_names[recipient_id]
    chat_dict: dict[str, Any] = waldiez_chat.get_chat_args(for_queue=True)
    # chat_dict = update_summary_chat_args(chat_dict)
    chat_dict["recipient"] = recipient_name
    if sender_name:
        chat_dict["sender"] = sender_name
    message_value, message_source = get_chat_nested_string(
        chat=waldiez_chat,
        is_reply=message.is_reply,
        chat_names=chat_names,
    )
    chat_dict["message"] = message_value
    message_dict_str = serializer(chat_dict, tabs=1)
    if message_source:
        # it's not a string, its the name of the function
        message_dict_str = message_dict_str.replace(
            f': "{message_value}"', f": {message_value}"
        ).replace(f'"{message_value}"', f"{message_value}")
    if sender_name:
        message_dict_str = message_dict_str.replace(
            f': "{sender_name}"', f": {sender_name}"
        )
    if recipient_name:
        message_dict_str = message_dict_str.replace(
            f': "{recipient_name}"', f": {recipient_name}"
        )
    return message_dict_str, message_source

get_nested_chat_queue

get_nested_chat_queue(
    nested_chat: WaldiezAgentNestedChat,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> tuple[str, list[str]]

Get the nested chat queue.

Parameters:

NameTypeDescriptionDefault
nested_chatWaldiezAgentNestedChat

The nested chat.

required
agentWaldiezAgent

The agent.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
all_chatslist[WaldiezChat]

All the chats in the flow.

required
serializerCallable[..., str]

The serializer to use to escape quotes in a string.

required

Returns:

TypeDescription
tuple[str, list[str]]

The nested chat queue and the methods to include (methods: message string and method name if the message is a method).

Source code in waldiez/exporting/chats/utils/nested.py
def get_nested_chat_queue(
    nested_chat: WaldiezAgentNestedChat,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> tuple[str, list[str]]:
    """Get the nested chat queue.

    Parameters
    ----------
    nested_chat : WaldiezAgentNestedChat
        The nested chat.
    agent : WaldiezAgent
        The agent.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    all_chats : list[WaldiezChat]
        All the chats in the flow.
    serializer : Callable[..., str]
        The serializer to use to escape quotes in a string.

    Returns
    -------
    tuple[str, list[str]]
        The nested chat queue and the methods to include
        (methods: message string and method name if the message is a method).
    """
    message_methods_to_include: list[str] = []
    chat_messages_str = "[\n"
    for message in nested_chat.messages:
        waldiez_chat = next(chat for chat in all_chats if chat.id == message.id)
        message_str, message_source = get_nested_chat_message_string(
            waldiez_chat=waldiez_chat,
            message=message,
            agent=agent,
            agent_names=agent_names,
            chat_names=chat_names,
            serializer=serializer,
        )
        if message_source:
            message_methods_to_include.append(message_source)
        chat_messages_str += f"    {message_str}," + "\n"
    chat_messages_str += "]"
    if chat_messages_str == "[\n]":
        return "", message_methods_to_include
    return chat_messages_str, message_methods_to_include

get_nested_chat_trigger_agent_names

get_nested_chat_trigger_agent_names(
    nested_chat: WaldiezAgentNestedChat,
    agent_names: dict[str, str],
) -> str

Get the trigger agent names for the nested chat.

Parameters:

NameTypeDescriptionDefault
nested_chatWaldiezAgentNestedChat

The nested chat.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required

Returns:

TypeDescription
str

The trigger agent names.

Source code in waldiez/exporting/chats/utils/nested.py
def get_nested_chat_trigger_agent_names(
    nested_chat: WaldiezAgentNestedChat,
    agent_names: dict[str, str],
) -> str:
    """Get the trigger agent names for the nested chat.

    Parameters
    ----------
    nested_chat : WaldiezAgentNestedChat
        The nested chat.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.

    Returns
    -------
    str
        The trigger agent names.
    """
    agents = [agent_names[agent_id] for agent_id in nested_chat.triggered_by]
    agents_string = [", ".join(agents)]
    trigger_string = f"{agents_string}"
    return trigger_string.replace("'", '"')

sequential

Utility functions for exporting sequential chats.

export_sequential_chat

export_sequential_chat(
    main_chats: list[WaldiezAgentConnection],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]

Get the chats content, when there are more than one chats in the flow.

Parameters:

NameTypeDescriptionDefault
main_chatslist[WaldiezAgentConnection]

The main chats.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
serializerCallable[..., str]

The serializer function to escape quotes in a string.

required
tabsint

The number of tabs to use for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required

Returns:

TypeDescription
tuple[str, str]

The main chats content and additional methods string if any.

Example
>>> from waldiez.models import (
...     WaldiezAgent,
...     WaldiezChat,
...     WaldiezChatData,
...     WaldiezChatMessage,
... )
>>> chat1 = WaldiezChat(
...     id="wc-1",
...     name="chat1",
...     description="A chat between two agents.",
...     tags=["chat", "chat1"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-1",
...         recipient="wa-2",
...         position=0,
...         message=WaldiezChatMessage(
...             type="string",
...             content="Hello, how are you?",
...         ),
...     ),
... )
>>> chat2 = WaldiezChat(
...     id="wc-2",
...     name="chat2",
...     description="A chat between two agents.",
...     tags=["chat", "chat2"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-2",
...         recipient="wa-1",
...         position=1,
...         message=WaldiezChatMessage(
...             type="string",
...             content="I am good, thank you. How about you?",
...         ),
...     ),
... )
>>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
>>> chat_names = {"wc-1": "chat1", "wc-2": "chat2"}
>>> serializer = lambda x: x.replace('"', "\"").replace("\n", "\\n")
>>>  export_sequential_chat(
...     main_chats=[(chat1, agent1, agent2), (chat2, agent2, agent1)],
...     chat_names=chat_names,
...     agent_names=agent_names,
...     serializer=serializer,
...     tabs=0,
...     is_async=False,
... )
results = initiate_chats([
    {
        "sender": agent1,
        "recipient": agent2,
        "message": "Hello, how are you?",
    },
    {
        "sender": agent2,
        "recipient": agent1,
        "message": "I am good, thank you. How about you?",
    },
])
Source code in waldiez/exporting/chats/utils/sequential.py
def export_sequential_chat(
    main_chats: list[WaldiezAgentConnection],
    chat_names: dict[str, str],
    agent_names: dict[str, str],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]:
    r"""Get the chats content, when there are more than one chats in the flow.

    Parameters
    ----------
    main_chats : list[WaldiezAgentConnection]
        The main chats.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    serializer : Callable[..., str]
        The serializer function to escape quotes in a string.
    tabs : int
        The number of tabs to use for indentation.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.

    Returns
    -------
    tuple[str, str]
        The main chats content and additional methods string if any.

    Example
    -------
    ```python
    >>> from waldiez.models import (
    ...     WaldiezAgent,
    ...     WaldiezChat,
    ...     WaldiezChatData,
    ...     WaldiezChatMessage,
    ... )
    >>> chat1 = WaldiezChat(
    ...     id="wc-1",
    ...     name="chat1",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat1"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-1",
    ...         recipient="wa-2",
    ...         position=0,
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="Hello, how are you?",
    ...         ),
    ...     ),
    ... )
    >>> chat2 = WaldiezChat(
    ...     id="wc-2",
    ...     name="chat2",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat2"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-2",
    ...         recipient="wa-1",
    ...         position=1,
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="I am good, thank you. How about you?",
    ...         ),
    ...     ),
    ... )
    >>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
    >>> chat_names = {"wc-1": "chat1", "wc-2": "chat2"}
    >>> serializer = lambda x: x.replace('"', "\"").replace("\n", "\\n")
    >>>  export_sequential_chat(
    ...     main_chats=[(chat1, agent1, agent2), (chat2, agent2, agent1)],
    ...     chat_names=chat_names,
    ...     agent_names=agent_names,
    ...     serializer=serializer,
    ...     tabs=0,
    ...     is_async=False,
    ... )
    results = initiate_chats([
        {
            "sender": agent1,
            "recipient": agent2,
            "message": "Hello, how are you?",
        },
        {
            "sender": agent2,
            "recipient": agent1,
            "message": "I am good, thank you. How about you?",
        },
    ])
    ```
    """
    tab = "    " * tabs if tabs > 0 else ""
    content = "\n"
    additional_methods_string = ""
    content += _get_initiate_chats_line(tab, is_async)
    for connection in main_chats:
        chat_string, additional_methods = _get_chat_dict_string(
            chat_names=chat_names,
            connection=connection,
            agent_names=agent_names,
            serializer=serializer,
            tabs=tabs + 1,
            skip_cache=skip_cache,
        )
        additional_methods_string += additional_methods
        content += "\n" + f"{tab}    {chat_string}"
    content += "\n" + "    " * tabs + "])\n"
    return content, additional_methods_string

single

Utilities for exporting a single chat in a flow.

export_single_chat

export_single_chat(
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    chat: WaldiezChat,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[[str], str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
    tab_leng: int = 4,
) -> tuple[str, str]

Get the chat string when there is only one chat in the flow.

Parameters:

NameTypeDescriptionDefault
senderWaldiezAgent

The sender.

required
recipientWaldiezAgent

The recipient.

required
chatWaldiezChat

The chat.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
serializerCallable[[str], str]

The serializer function to escape quotes in a string.

required
tabsint

The number of tabs to use for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required
tab_lengint

The length of the tab string, by default 4.

4

Returns:

TypeDescription
tuple[str, str]

The chat string and additional methods string if any

Example
>>> from waldiez.models import WaldiezAgent, WaldiezChat, WaldiezChatData, WaldiezChatMessage
>>> chat = WaldiezChat(
...     id="wc-1",
...     name="chat1",
...     description="A chat between two agents.",
...     tags=["chat", "chat1"],
...     requirements=[],
...     data=WaldiezChatData(
...         sender="wa-1",
...         recipient="wa-2",
...         message=WaldiezChatMessage(
...             type="string",
...             content="Hello, how are you?",
...         ),
...     ),
... )
>>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
>>> chat_names = {"wc-1": "chat1"}
>>> export_single_chat_string(
...     sender=agent1,
...     recipient=agent2,
...     chat=chat,
...     agent_names=agent_names,
...     chat_names=chat_names,
...     tabs=0,
... )
agent1.initiate_chat(
    agent2,
    message="Hello, how are you?",
    cache=cache,
)
Source code in waldiez/exporting/chats/utils/single.py
def export_single_chat(
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    chat: WaldiezChat,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    serializer: Callable[[str], str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
    tab_leng: int = 4,
) -> tuple[str, str]:
    """Get the chat string when there is only one chat in the flow.

    Parameters
    ----------
    sender : WaldiezAgent
        The sender.
    recipient : WaldiezAgent
        The recipient.
    chat : WaldiezChat
        The chat.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    serializer : Callable[[str], str]
        The serializer function to escape quotes in a string.
    tabs : int
        The number of tabs to use for indentation.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.
    tab_leng : int, optional
        The length of the tab string, by default 4.

    Returns
    -------
    tuple[str, str]
        The chat string and additional methods string if any

    Example
    -------
    ```python
    >>> from waldiez.models import WaldiezAgent, WaldiezChat, WaldiezChatData, WaldiezChatMessage
    >>> chat = WaldiezChat(
    ...     id="wc-1",
    ...     name="chat1",
    ...     description="A chat between two agents.",
    ...     tags=["chat", "chat1"],
    ...     requirements=[],
    ...     data=WaldiezChatData(
    ...         sender="wa-1",
    ...         recipient="wa-2",
    ...         message=WaldiezChatMessage(
    ...             type="string",
    ...             content="Hello, how are you?",
    ...         ),
    ...     ),
    ... )
    >>> agent_names = {"wa-1": "agent1", "wa-2": "agent2"}
    >>> chat_names = {"wc-1": "chat1"}
    >>> export_single_chat_string(
    ...     sender=agent1,
    ...     recipient=agent2,
    ...     chat=chat,
    ...     agent_names=agent_names,
    ...     chat_names=chat_names,
    ...     tabs=0,
    ... )
    agent1.initiate_chat(
        agent2,
        message="Hello, how are you?",
        cache=cache,
    )
    ```
    """
    tab = " " * tab_leng * tabs if tabs > 0 else ""
    chat_args = chat.get_chat_args(for_queue=False, sender=sender)
    # chat_args = update_summary_chat_args(chat_args, skip_summary=True)
    if not chat_args:
        return get_empty_simple_chat_string(
            chat=chat,
            sender=sender,
            recipient=recipient,
            agent_names=agent_names,
            tab=tab,
            is_async=is_async,
            skip_cache=skip_cache,
        )
    return get_simple_chat_string(
        chat=chat,
        chat_args=chat_args,
        sender=sender,
        recipient=recipient,
        agent_names=agent_names,
        chat_names=chat_names,
        serializer=serializer,
        tabs=tabs,
        is_async=is_async,
        skip_cache=skip_cache,
    )

get_chat_message

get_chat_message(
    tab: str,
    chat: WaldiezChat,
    chat_names: dict[str, str],
    sender: WaldiezAgent,
    sender_name: str,
) -> tuple[str, str]

Get the chat message string.

Parameters:

NameTypeDescriptionDefault
tabstr

The tab string.

required
chatWaldiezChat

The chat.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
senderWaldiezAgent

The sender.

required
sender_namestr

The sender name.

required

Returns:

TypeDescription
tuple[str, str]

The message argument and additional methods string if any.

Source code in waldiez/exporting/chats/utils/single.py
def get_chat_message(
    tab: str,
    chat: WaldiezChat,
    chat_names: dict[str, str],
    sender: WaldiezAgent,
    sender_name: str,
) -> tuple[str, str]:
    """Get the chat message string.

    Parameters
    ----------
    tab : str
        The tab string.
    chat : WaldiezChat
        The chat.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    sender : WaldiezAgent
        The sender.
    sender_name : str
        The sender name.

    Returns
    -------
    tuple[str, str]
        The message argument and additional methods string if any.
    """
    additional_methods_string = ""
    method_content: Optional[str] = None
    if (
        sender.is_rag_user
        and isinstance(sender, WaldiezRagUserProxy)
        and chat.message.type == "rag_message_generator"
        and chat.message.use_carryover is False
    ):
        message = f"{sender_name}.message_generator"
        return "\n" + f"{tab}    message={message},", additional_methods_string
    message, method_content = get_chat_message_string(
        sender=sender,
        chat=chat,
        chat_names=chat_names,
    )
    if message and isinstance(chat.data.message, WaldiezChatMessage):
        if chat.data.message.type == "method":
            additional_methods_string += (
                method_content if method_content else ""
            )
            return (
                "\n" + f"{tab}    message={message},",
                additional_methods_string,
            )
        if chat.message.type == "string" and chat.data.message.content:
            return (
                "\n" + f'{tab}    message="{message}",',
                additional_methods_string,
            )
        if chat.message.type == "rag_message_generator":
            additional_methods_string += (
                method_content if method_content else ""
            )
            return (
                "\n" + f"{tab}    message={message},",
                additional_methods_string,
            )
    return "", additional_methods_string  # pragma: no cover

get_empty_simple_chat_string

get_empty_simple_chat_string(
    chat: WaldiezChat,
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    agent_names: dict[str, str],
    tab: str,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]

Get the chat string when there are no chat arguments.

Parameters:

NameTypeDescriptionDefault
chatWaldiezChat

The chat.

required
senderWaldiezAgent

The sender.

required
recipientWaldiezAgent

The recipient.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
tabstr

The tab string.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required

Returns:

TypeDescription
tuple[str, str]

The chat string and additional methods string if any

Source code in waldiez/exporting/chats/utils/single.py
def get_empty_simple_chat_string(
    chat: WaldiezChat,
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    agent_names: dict[str, str],
    tab: str,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]:
    """Get the chat string when there are no chat arguments.

    Parameters
    ----------
    chat : WaldiezChat
        The chat.
    sender : WaldiezAgent
        The sender.
    recipient : WaldiezAgent
        The recipient.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    tab : str
        The tab string.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.

    Returns
    -------
    tuple[str, str]
        The chat string and additional methods string if any
    """
    sender_name = agent_names[sender.id]
    if is_async:
        sender_name = f"await {sender_name}"
    recipient_name = agent_names[recipient.id]
    initiate = "a_initiate_chat" if is_async else "initiate_chat"
    content = "\n" + f"{tab}results = {sender_name}.{initiate}(" + "\n"
    content += f"{tab}    {recipient_name}," + "\n"
    if not skip_cache:
        content += f"{tab}    cache=cache," + "\n"
    message_arg, _ = get_chat_message(
        tab=tab,
        chat=chat,
        chat_names={},
        sender=sender,
        sender_name=sender_name,
    )
    content += message_arg
    content += f"{tab})" + "\n"
    return content, ""

get_simple_chat_string

get_simple_chat_string(
    chat: WaldiezChat,
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    chat_args: dict[str, Any],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]

Get the chat string when there are chat arguments.

Parameters:

NameTypeDescriptionDefault
chatWaldiezChat

The chat.

required
senderWaldiezAgent

The sender.

required
recipientWaldiezAgent

The recipient.

required
agent_namesdict[str, str]

A mapping of agent id to agent name.

required
chat_namesdict[str, str]

A mapping of chat id to chat name.

required
chat_argsdict[str, Any]

The chat arguments.

required
serializerCallable[[str], str]

The serializer function to escape quotes in a string.

required
tabsint

The number of tabs to use for indentation.

required
is_asyncbool

Whether the chat is asynchronous.

required
skip_cachebool

Whether to skip the cache argument.

required

Returns:

TypeDescription
tuple[str, str]

The chat string and additional methods string if any.

Source code in waldiez/exporting/chats/utils/single.py
def get_simple_chat_string(
    chat: WaldiezChat,
    sender: WaldiezAgent,
    recipient: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    chat_args: dict[str, Any],
    serializer: Callable[..., str],
    tabs: int,
    is_async: bool,
    skip_cache: bool,
) -> tuple[str, str]:
    """Get the chat string when there are chat arguments.

    Parameters
    ----------
    chat : WaldiezChat
        The chat.
    sender : WaldiezAgent
        The sender.
    recipient : WaldiezAgent
        The recipient.
    agent_names : dict[str, str]
        A mapping of agent id to agent name.
    chat_names : dict[str, str]
        A mapping of chat id to chat name.
    chat_args : dict[str, Any]
        The chat arguments.
    serializer : Callable[[str], str]
        The serializer function to escape quotes in a string.
    tabs : int
        The number of tabs to use for indentation.
    is_async : bool
        Whether the chat is asynchronous.
    skip_cache : bool
        Whether to skip the cache argument.

    Returns
    -------
    tuple[str, str]
        The chat string and additional methods string if any.
    """
    tab = "    " * tabs
    sender_name = agent_names[sender.id]
    initiate = "initiate_chat"
    if is_async:
        sender_name = f"await {sender_name}"
        initiate = "a_initiate_chat"
    recipient_name = agent_names[recipient.id]
    chat_string = "\n" + f"{tab}results = {sender_name}.{initiate}(" + "\n"
    chat_string += f"{tab}    {recipient_name},"
    if not skip_cache:
        chat_string += "\n" + f"{tab}    cache=cache,"
    for key, value in chat_args.items():
        if isinstance(value, str):
            chat_string += "\n" + f'{tab}    {key}="{value}",'
        elif isinstance(value, dict):
            chat_string += (
                "\n" + f"{tab}    {key}={serializer(value, tabs=tabs + 1)},"
            )
        else:
            chat_string += "\n" + f"{tab}    {key}={value},"
    message_arg, additional_methods_string = get_chat_message(
        tab=tab,
        chat=chat,
        chat_names=chat_names,
        sender=sender,
        sender_name=sender_name,
    )
    chat_string += message_arg
    chat_string += "\n" + f"{tab})" + "\n"
    return chat_string, additional_methods_string