Agents

Agent exporter.

AgentExporter ¶

AgentExporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    initial_chats: Optional[
        list[WaldiezAgentConnection]
    ] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
)

Bases: Exporter[StandardExtras]

Agent exporter with standard extras.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The agent to export.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to names.	required
`models`	`tuple[list[WaldiezModel], dict[str, str]]`	All models and model names mapping.	required
`chats`	`tuple[list[WaldiezChat], dict[str, str]]`	All chats and chat names mapping.	required
`tool_names`	`dict[str, str]`	Mapping of tool IDs to names.	required
`is_async`	`bool`	Whether the flow is async, by default False	`False`
`for_notebook`	`bool`	Whether exporting for notebook, by default False	`False`
`cache_seed`	`Optional[int]`	Cache seed if any, by default None	`None`
`initial_chats`	`Optional[list[WaldiezAgentConnection]]`	Initial chats for group managers, by default None	`None`
`group_chat_members`	`Optional[list[WaldiezAgent]]`	Group chat members if group manager, by default None	`None`
`arguments_resolver`	`Optional[Callable]`	Function to resolve additional arguments, by default None	`None`
`output_dir`	`Optional[Union[str, Path]]`	Output directory for generated files, by default None	`None`
`context`	`Optional[ExporterContext]`	Exporter context with dependencies, by default None	`None`
`**kwargs`	`Any`	Additional keyword arguments.	`{}`

Source code in waldiez/exporting/agent/exporter.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    initial_chats: Optional[list[WaldiezAgentConnection]] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
):
    """Initialize the agent exporter.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent to export.
    agent_names : dict[str, str]
        Mapping of agent IDs to names.
    models : tuple[list[WaldiezModel], dict[str, str]]
        All models and model names mapping.
    chats : tuple[list[WaldiezChat], dict[str, str]]
        All chats and chat names mapping.
    tool_names : dict[str, str]
        Mapping of tool IDs to names.
    is_async : bool, optional
        Whether the flow is async, by default False
    for_notebook : bool, optional
        Whether exporting for notebook, by default False
    cache_seed : Optional[int], optional
        Cache seed if any, by default None
    initial_chats : Optional[list[WaldiezAgentConnection]], optional
        Initial chats for group managers, by default None
    group_chat_members : Optional[list[WaldiezAgent]], optional
        Group chat members if group manager, by default None
    arguments_resolver : Optional[Callable], optional
        Function to resolve additional arguments, by default None
    output_dir : Optional[Union[str, Path]], optional
        Output directory for generated files, by default None
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(context, **kwargs)

    self.agent = agent
    self.agent_names = agent_names
    self.models = models[0]
    self.model_names = models[1]
    self.all_chats = chats[0]
    self.chat_names = chats[1]
    self.tool_names = tool_names
    self.is_async = is_async
    self.for_notebook = for_notebook
    self.cache_seed = cache_seed
    self.initial_chats = initial_chats or []
    self.group_chat_members = group_chat_members or []
    self.arguments_resolver = arguments_resolver or fallback_args_resolver
    self.output_dir = Path(output_dir) if output_dir else None

    # Initialize extras based on agent type
    self._extras = self._create_agent_extras()

create_reasoning_extras ¶

create_reasoning_extras() -> StandardExtras

Create reasoning-specific extras for the agent.

Returns:

Type	Description
`StandardExtras`	The reasoning-specific extras.

Source code in waldiez/exporting/agent/exporter.py

def create_reasoning_extras(self) -> StandardExtras:
    """Create reasoning-specific extras for the agent.

    Returns
    -------
    StandardExtras
        The reasoning-specific extras.
    """
    # Start with standard extras
    extras = self._create_standard_extras()

    reasoning_processor = ReasoningAgentProcessor(
        agent=self.agent,
        serializer=(
            self.context.get_serializer()
            if self.context.serializer
            else None
        ),
    )
    # Process reasoning-specific content
    reason_extras = reasoning_processor.process(
        code_execution_config=extras.code_execution_config,
        termination_config=extras.termination_config,
        system_message_config=extras.system_message_config,
    )
    # Add reasoning arguments
    for arg in reason_extras.extra_args:
        extras.add_arg(arg)

    return reason_extras

extras `property` ¶

extras: StandardExtras

Get the agent extras.

generate_main_content ¶

generate_main_content() -> Optional[str]

Generate the main agent definition.

Returns:

Type	Description
`Optional[str]`	The main agent definition content, or None if not applicable.

Source code in waldiez/exporting/agent/exporter.py

def generate_main_content(self) -> Optional[str]:
    """Generate the main agent definition.

    Returns
    -------
    Optional[str]
        The main agent definition content, or None if not applicable.
    """
    # Use AgentProcessor to generate the main definition
    agent_processor = AgentProcessor(
        agent=self.agent,
        agent_names=self.agent_names,
        arguments_resolver=self.arguments_resolver,
        extras=self.extras,
    )
    result = agent_processor.process().content
    if result:
        # Add the main agent definition content
        self.add_content(
            result,
            ExportPosition.AGENTS,
            order=(
                ContentOrder.MAIN_CONTENT
                if not self.agent.is_group_manager
                else ContentOrder.LATE_CLEANUP
            ),
        )
        self.extras.append_after_all_agents(result)
    return result

code_execution ¶

Code execution processor for Waldiez agents.

CodeExecutionProcessor ¶

CodeExecutionProcessor(
    agent: WaldiezAgent,
    agent_name: str,
    tool_names: dict[str, str],
)

Processor for code execution configuration.

Source code in waldiez/exporting/agent/code_execution.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_name: str,
    tool_names: dict[str, str],
):
    self.agent = agent
    self.agent_name = agent_name
    self.tool_names = tool_names

process ¶

process() -> CodeExecutionConfig

Process code execution configuration.

Returns:

Type	Description
`CodeExecutionConfig`	The processed code execution configuration.

Source code in waldiez/exporting/agent/code_execution.py

def process(self) -> CodeExecutionConfig:
    """Process code execution configuration.

    Returns
    -------
    CodeExecutionConfig
        The processed code execution configuration.
    """
    if self.agent.data.code_execution_config is False:
        return CodeExecutionConfig(
            executor_content="",
            executor_argument="False",
            executor_import=None,
        )

    config = self.agent.data.code_execution_config
    use_docker = (
        config.use_docker if config.use_docker is not None else False
    )

    executor_class = self._get_executor_class_name(use_docker)
    executor_content = self._build_executor_content(config, use_docker)
    executor_arg = f'{{"executor": {self.agent_name}_executor}}'
    executor_import = f"from autogen.coding import {executor_class}"

    return CodeExecutionConfig(
        executor_content=executor_content,
        executor_argument=executor_arg,
        executor_import=ImportStatement(
            statement=executor_import,
            position=ImportPosition.THIRD_PARTY,
        ),
    )

create_agent_exporter ¶

create_agent_exporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    initial_chats: list[WaldiezAgentConnection],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
) -> AgentExporter

Create an agent exporter.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The agent to export.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to names.	required
`models`	`tuple[list[WaldiezModel], dict[str, str]]`	All models and model names mapping.	required
`chats`	`tuple[list[WaldiezChat], dict[str, str]]`	All chats and chat names mapping.	required
`tool_names`	`dict[str, str]`	Mapping of tool IDs to names.	required
`is_async`	`bool`	Whether the flow is async, by default False	`False`
`for_notebook`	`bool`	Whether exporting for notebook, by default False	`False`
`cache_seed`	`Optional[int]`	Cache seed if any, by default None	`None`
`initial_chats`	`Optional[list[WaldiezAgentConnection]]`	Initial chats for group managers, by default None	required
`group_chat_members`	`Optional[list[WaldiezAgent]]`	Group chat members if group manager, by default None	`None`
`arguments_resolver`	`Optional[Callable]`	Function to resolve additional arguments, by default None	`None`
`output_dir`	`Optional[Union[str, Path]]`	Output directory for generated files, by default None	`None`
`context`	`Optional[ExporterContext]`	Exporter context with dependencies, by default None	`None`
`**kwargs`	`Any`	Additional keyword arguments.	`{}`

Returns:

Type	Description
`AgentExporter`	The created agent exporter.

Source code in waldiez/exporting/agent/factory.py

def create_agent_exporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    initial_chats: list[WaldiezAgentConnection],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[Callable[[WaldiezAgent], list[str]]] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> AgentExporter:
    """Create an agent exporter.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent to export.
    agent_names : dict[str, str]
        Mapping of agent IDs to names.
    models : tuple[list[WaldiezModel], dict[str, str]]
        All models and model names mapping.
    chats : tuple[list[WaldiezChat], dict[str, str]]
        All chats and chat names mapping.
    tool_names : dict[str, str]
        Mapping of tool IDs to names.
    is_async : bool, optional
        Whether the flow is async, by default False
    for_notebook : bool, optional
        Whether exporting for notebook, by default False
    cache_seed : Optional[int], optional
        Cache seed if any, by default None
    initial_chats : Optional[list[WaldiezAgentConnection]], optional
        Initial chats for group managers, by default None
    group_chat_members : Optional[list[WaldiezAgent]], optional
        Group chat members if group manager, by default None
    arguments_resolver : Optional[Callable], optional
        Function to resolve additional arguments, by default None
    output_dir : Optional[Union[str, Path]], optional
        Output directory for generated files, by default None
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    AgentExporter
        The created agent exporter.
    """
    if context is None:
        context = get_default_exporter_context()
    return AgentExporter(
        agent=agent,
        agent_names=agent_names,
        models=models,
        chats=chats,
        tool_names=tool_names,
        is_async=is_async,
        for_notebook=for_notebook,
        cache_seed=cache_seed,
        initial_chats=initial_chats,
        group_chat_members=group_chat_members,
        arguments_resolver=arguments_resolver,
        output_dir=output_dir,
        context=context,
        **kwargs,
    )

exporter ¶

Export agents.

AgentExporter ¶

AgentExporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    initial_chats: Optional[
        list[WaldiezAgentConnection]
    ] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
)

Bases: Exporter[StandardExtras]

Agent exporter with standard extras.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The agent to export.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to names.	required
`models`	`tuple[list[WaldiezModel], dict[str, str]]`	All models and model names mapping.	required
`chats`	`tuple[list[WaldiezChat], dict[str, str]]`	All chats and chat names mapping.	required
`tool_names`	`dict[str, str]`	Mapping of tool IDs to names.	required
`is_async`	`bool`	Whether the flow is async, by default False	`False`
`for_notebook`	`bool`	Whether exporting for notebook, by default False	`False`
`cache_seed`	`Optional[int]`	Cache seed if any, by default None	`None`
`initial_chats`	`Optional[list[WaldiezAgentConnection]]`	Initial chats for group managers, by default None	`None`
`group_chat_members`	`Optional[list[WaldiezAgent]]`	Group chat members if group manager, by default None	`None`
`arguments_resolver`	`Optional[Callable]`	Function to resolve additional arguments, by default None	`None`
`output_dir`	`Optional[Union[str, Path]]`	Output directory for generated files, by default None	`None`
`context`	`Optional[ExporterContext]`	Exporter context with dependencies, by default None	`None`
`**kwargs`	`Any`	Additional keyword arguments.	`{}`

Source code in waldiez/exporting/agent/exporter.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    initial_chats: Optional[list[WaldiezAgentConnection]] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
):
    """Initialize the agent exporter.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent to export.
    agent_names : dict[str, str]
        Mapping of agent IDs to names.
    models : tuple[list[WaldiezModel], dict[str, str]]
        All models and model names mapping.
    chats : tuple[list[WaldiezChat], dict[str, str]]
        All chats and chat names mapping.
    tool_names : dict[str, str]
        Mapping of tool IDs to names.
    is_async : bool, optional
        Whether the flow is async, by default False
    for_notebook : bool, optional
        Whether exporting for notebook, by default False
    cache_seed : Optional[int], optional
        Cache seed if any, by default None
    initial_chats : Optional[list[WaldiezAgentConnection]], optional
        Initial chats for group managers, by default None
    group_chat_members : Optional[list[WaldiezAgent]], optional
        Group chat members if group manager, by default None
    arguments_resolver : Optional[Callable], optional
        Function to resolve additional arguments, by default None
    output_dir : Optional[Union[str, Path]], optional
        Output directory for generated files, by default None
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(context, **kwargs)

    self.agent = agent
    self.agent_names = agent_names
    self.models = models[0]
    self.model_names = models[1]
    self.all_chats = chats[0]
    self.chat_names = chats[1]
    self.tool_names = tool_names
    self.is_async = is_async
    self.for_notebook = for_notebook
    self.cache_seed = cache_seed
    self.initial_chats = initial_chats or []
    self.group_chat_members = group_chat_members or []
    self.arguments_resolver = arguments_resolver or fallback_args_resolver
    self.output_dir = Path(output_dir) if output_dir else None

    # Initialize extras based on agent type
    self._extras = self._create_agent_extras()

create_reasoning_extras ¶

create_reasoning_extras() -> StandardExtras

Create reasoning-specific extras for the agent.

Returns:

Type	Description
`StandardExtras`	The reasoning-specific extras.

Source code in waldiez/exporting/agent/exporter.py

def create_reasoning_extras(self) -> StandardExtras:
    """Create reasoning-specific extras for the agent.

    Returns
    -------
    StandardExtras
        The reasoning-specific extras.
    """
    # Start with standard extras
    extras = self._create_standard_extras()

    reasoning_processor = ReasoningAgentProcessor(
        agent=self.agent,
        serializer=(
            self.context.get_serializer()
            if self.context.serializer
            else None
        ),
    )
    # Process reasoning-specific content
    reason_extras = reasoning_processor.process(
        code_execution_config=extras.code_execution_config,
        termination_config=extras.termination_config,
        system_message_config=extras.system_message_config,
    )
    # Add reasoning arguments
    for arg in reason_extras.extra_args:
        extras.add_arg(arg)

    return reason_extras

extras `property` ¶

extras: StandardExtras

Get the agent extras.

generate_main_content ¶

generate_main_content() -> Optional[str]

Generate the main agent definition.

Returns:

Type	Description
`Optional[str]`	The main agent definition content, or None if not applicable.

Source code in waldiez/exporting/agent/exporter.py

def generate_main_content(self) -> Optional[str]:
    """Generate the main agent definition.

    Returns
    -------
    Optional[str]
        The main agent definition content, or None if not applicable.
    """
    # Use AgentProcessor to generate the main definition
    agent_processor = AgentProcessor(
        agent=self.agent,
        agent_names=self.agent_names,
        arguments_resolver=self.arguments_resolver,
        extras=self.extras,
    )
    result = agent_processor.process().content
    if result:
        # Add the main agent definition content
        self.add_content(
            result,
            ExportPosition.AGENTS,
            order=(
                ContentOrder.MAIN_CONTENT
                if not self.agent.is_group_manager
                else ContentOrder.LATE_CLEANUP
            ),
        )
        self.extras.append_after_all_agents(result)
    return result

fallback_args_resolver ¶

fallback_args_resolver(agent: WaldiezAgent) -> list[str]

Fallback resolver for agent arguments.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The agent for which to resolve arguments.	required

Returns:

Type	Description
`list[str]`	An empty list (no extra args).

Source code in waldiez/exporting/agent/exporter.py

def fallback_args_resolver(agent: WaldiezAgent) -> list[str]:
    """Fallback resolver for agent arguments.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent for which to resolve arguments.

    Returns
    -------
    list[str]
        An empty list (no extra args).
    """
    return []

extras ¶

Agents exporter extras.

CaptainAgentProcessor ¶

CaptainAgentProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Optional[Serializer] = None,
    output_dir: Optional[Path] = None,
)

Processor for captain agent configuration.

Source code in waldiez/exporting/agent/extras/captain_agent_extras.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Optional[Serializer] = None,
    output_dir: Optional[Path] = None,
):
    self.agent = agent
    self.agent_names = agent_names
    self.all_models = all_models
    self.serializer = serializer or DefaultSerializer()
    self.output_dir = output_dir

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> CaptainExtras

Process group manager and return extras.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, by default None.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, by default None.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, by default None.	`None`

Returns:

Type	Description
`CaptainExtras`	The processed result containing extra arguments, before content, imports, and environment variables.

Source code in waldiez/exporting/agent/extras/captain_agent_extras.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> CaptainExtras:
    """Process group manager and return extras.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, by default None.
    termination_config : TerminationConfig, optional
        Configuration for termination, by default None.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, by default None.

    Returns
    -------
    CaptainExtras
        The processed result containing extra arguments, before content,
        imports, and environment variables.
    """
    result = CaptainExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )
    if not self.agent.is_captain or not isinstance(
        self.agent, WaldiezCaptainAgent
    ):
        return result
    agent_name = self.agent_names[self.agent.id]
    save_path = str(self.output_dir) if self.output_dir else "."
    if save_path != ".":
        os.makedirs(save_path, exist_ok=True)
    result.add_arg("agent_config_save_path=os.getcwd()", tabs=1)
    if self.agent.data.agent_lib:
        lib_dict = [
            lib.model_dump(by_alias=False)
            for lib in self.agent.data.agent_lib
        ]
        lib_json_name = f"{agent_name}_agent_lib.json"
        agent_lib_path = os.path.join(save_path, lib_json_name)
        with open(agent_lib_path, "w", encoding="utf-8", newline="\n") as f:
            json.dump(lib_dict, f, ensure_ascii=False, indent=4)
        result.add_arg(f'agent_lib="{lib_json_name}"', tabs=1)
    if self.agent.data.tool_lib:
        result.add_arg(f'tool_lib="{self.agent.data.tool_lib}"', tabs=1)
    nested_config = self._generate_nested_config(
        self.agent,
        agent_name,
        save_path,
    )
    result.set_nested_config(nested_config)
    serialized_nested_config = self.serializer.serialize(nested_config)
    result.add_arg(f"nested_config={serialized_nested_config}", tabs=1)
    return result

GroupManagerProcessor ¶

GroupManagerProcessor(
    agent: WaldiezAgent,
    initial_chats: list[WaldiezAgentConnection],
    group_chat_members: list[WaldiezAgent],
    agent_names: dict[str, str],
    model_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Callable[..., str],
    cache_seed: Optional[int] = None,
)

Processes group manager configurations.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The group manager agent to process.	required
`initial_chats`	`list[WaldiezAgentConnection]`	The initial chats to process for the group manager.	required
`group_chat_members`	`list[WaldiezAgent]`	The members of the group chat.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to their names.	required
`model_names`	`dict[str, str]`	Mapping of model IDs to their names.	required
`all_models`	`list[WaldiezModel]`	List of all models available for the agent.	required
`serializer`	`Callable[..., str]`	Function to serialize data into a string format.	required
`cache_seed`	`Optional[int]`	Seed for caching purposes, by default None.	`None`

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

def __init__(
    self,
    agent: WaldiezAgent,
    initial_chats: list[WaldiezAgentConnection],
    group_chat_members: list[WaldiezAgent],
    agent_names: dict[str, str],
    model_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Callable[..., str],
    cache_seed: Optional[int] = None,
):
    """Initialize the group manager processor.

    Parameters
    ----------
    agent : WaldiezAgent
        The group manager agent to process.
    initial_chats : list[WaldiezAgentConnection]
        The initial chats to process for the group manager.
    group_chat_members : list[WaldiezAgent]
        The members of the group chat.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    model_names : dict[str, str]
        Mapping of model IDs to their names.
    all_models : list[WaldiezModel]
        List of all models available for the agent.
    serializer : Callable[..., str]
        Function to serialize data into a string format.
    cache_seed : Optional[int], optional
        Seed for caching purposes, by default None.
    """
    self.agent = agent
    self.initial_chats = initial_chats
    self.group_chat_members = group_chat_members
    self.agent_names = agent_names
    self.model_names = model_names
    self.all_models = all_models
    self.serializer = serializer
    self.cache_seed = cache_seed

is_pattern_strategy `staticmethod` ¶

is_pattern_strategy(
    initial_chats: list[WaldiezAgentConnection],
) -> bool

Check if the group manager should use pattern strategy.

Parameters:

Name	Type	Description	Default
`initial_chats`	`list[WaldiezAgentConnection]`	The initial chats to determine the strategy.	required

Returns:

Type	Description
`bool`	True if pattern strategy should be used, False otherwise.

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

@staticmethod
def is_pattern_strategy(
    initial_chats: list[WaldiezAgentConnection],
) -> bool:
    """Check if the group manager should use pattern strategy.

    Parameters
    ----------
    initial_chats : list[WaldiezAgentConnection]
        The initial chats to determine the strategy.

    Returns
    -------
    bool
        True if pattern strategy should be used, False otherwise.
    """
    if not initial_chats:
        return True

    first_chat = initial_chats[0]["chat"]
    return (
        isinstance(first_chat.data.message, str)
        or not first_chat.data.message.is_method()
    )

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> GroupManagerExtras

Process group manager and return extras.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, by default None.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, by default None.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, by default None.	`None`

Returns:

Type	Description
`GroupManagerExtras`	The processed group manager extras containing configuration.

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> GroupManagerExtras:
    """Process group manager and return extras.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, by default None.
    termination_config : TerminationConfig, optional
        Configuration for termination, by default None.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, by default None.

    Returns
    -------
    GroupManagerExtras
        The processed group manager extras containing configuration.
    """
    extras = GroupManagerExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )

    # Determine strategy based on initial chats and message types
    extras.strategy = self._determine_strategy()

    if extras.strategy == GroupManagerStrategy.PATTERN:
        self._process_pattern_strategy(extras)
    else:
        self._process_traditional_strategy(extras)

    return extras

GroupMemberAgentProcessor ¶

GroupMemberAgentProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    tool_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
)

Processor for group member agent configuration.

Source code in waldiez/exporting/agent/extras/group_member_extras.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    tool_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
) -> None:
    self.agent = agent
    self.agent_names = agent_names
    self.tool_names = tool_names
    self.all_chats = all_chats
    self.chat_names = chat_names
    self.serializer = serializer

process ¶

process() -> GroupMemberProcessorResult

Process group member agent configuration.

Returns:

Type	Description
`GroupMemberProcessorResult`	The processed result containing before_agent code, after_agent code, and extra imports.

Source code in waldiez/exporting/agent/extras/group_member_extras.py

def process(
    self,
) -> GroupMemberProcessorResult:
    """Process group member agent configuration.

    Returns
    -------
    GroupMemberProcessorResult
        The processed result containing before_agent code,
        after_agent code, and extra imports.
    """
    self.result = GroupMemberProcessorResult()
    if not self.agent.is_group_member:
        return self.result
    self._process_agent_functions()
    self._process_agent_update_state_before_reply()
    self._process_handoff_registrations()
    return self.result

ReasoningAgentProcessor ¶

ReasoningAgentProcessor(
    agent: WaldiezAgent, serializer: Serializer | None
)

Processor for reasoning agent configuration.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The Waldiez agent to process.	required
`serializer`	`Serializer \| None`	Optional serializer for the reasoning configuration. Defaults to DefaultSerializer if not provided.	required

Source code in waldiez/exporting/agent/extras/reasoning_agent_extras.py

def __init__(self, agent: WaldiezAgent, serializer: Serializer | None):
    """Initialize the processor with the agent and serializer.

    Parameters
    ----------
    agent : WaldiezAgent
        The Waldiez agent to process.
    serializer : Serializer | None
        Optional serializer for the reasoning configuration.
        Defaults to DefaultSerializer if not provided.
    """
    self.agent = agent
    self.serializer = serializer or DefaultSerializer()

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> ReasoningExtras

Process reasoning agent configuration.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, if applicable.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, if applicable.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, if applicable.	`None`

Returns:

Type	Description
`ReasoningExtras`	The processed result containing extra arguments, before content, imports, and environment variables.

Source code in waldiez/exporting/agent/extras/reasoning_agent_extras.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> ReasoningExtras:
    """Process reasoning agent configuration.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, if applicable.
    termination_config : TerminationConfig, optional
        Configuration for termination, if applicable.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, if applicable.

    Returns
    -------
    ReasoningExtras
        The processed result containing extra arguments, before content,
        imports, and environment variables.
    """
    result = ReasoningExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )
    if not self.agent.is_reasoning or not isinstance(
        self.agent, WaldiezReasoningAgent
    ):  # pragma: no cover
        return result

    reasoning_config = self.agent.get_reasoning_config()
    serialized = self.serializer.serialize(reasoning_config)
    reason_arg = InstanceArgument(
        instance_id=self.agent.id,
        name="reason_config",
        value=serialized,
        tabs=1,
    )
    result.add_arg(reason_arg)
    verbose_arg = InstanceArgument(
        instance_id=self.agent.id,
        name="verbose",
        value=self.agent.data.verbose,
        tabs=1,
    )
    result.add_arg(verbose_arg)
    return result

captain_agent_extras ¶

Captain agent configuration processor.

CaptainAgentProcessor ¶

CaptainAgentProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Optional[Serializer] = None,
    output_dir: Optional[Path] = None,
)

Processor for captain agent configuration.

Source code in waldiez/exporting/agent/extras/captain_agent_extras.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Optional[Serializer] = None,
    output_dir: Optional[Path] = None,
):
    self.agent = agent
    self.agent_names = agent_names
    self.all_models = all_models
    self.serializer = serializer or DefaultSerializer()
    self.output_dir = output_dir

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> CaptainExtras

Process group manager and return extras.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, by default None.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, by default None.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, by default None.	`None`

Returns:

Type	Description
`CaptainExtras`	The processed result containing extra arguments, before content, imports, and environment variables.

Source code in waldiez/exporting/agent/extras/captain_agent_extras.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> CaptainExtras:
    """Process group manager and return extras.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, by default None.
    termination_config : TerminationConfig, optional
        Configuration for termination, by default None.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, by default None.

    Returns
    -------
    CaptainExtras
        The processed result containing extra arguments, before content,
        imports, and environment variables.
    """
    result = CaptainExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )
    if not self.agent.is_captain or not isinstance(
        self.agent, WaldiezCaptainAgent
    ):
        return result
    agent_name = self.agent_names[self.agent.id]
    save_path = str(self.output_dir) if self.output_dir else "."
    if save_path != ".":
        os.makedirs(save_path, exist_ok=True)
    result.add_arg("agent_config_save_path=os.getcwd()", tabs=1)
    if self.agent.data.agent_lib:
        lib_dict = [
            lib.model_dump(by_alias=False)
            for lib in self.agent.data.agent_lib
        ]
        lib_json_name = f"{agent_name}_agent_lib.json"
        agent_lib_path = os.path.join(save_path, lib_json_name)
        with open(agent_lib_path, "w", encoding="utf-8", newline="\n") as f:
            json.dump(lib_dict, f, ensure_ascii=False, indent=4)
        result.add_arg(f'agent_lib="{lib_json_name}"', tabs=1)
    if self.agent.data.tool_lib:
        result.add_arg(f'tool_lib="{self.agent.data.tool_lib}"', tabs=1)
    nested_config = self._generate_nested_config(
        self.agent,
        agent_name,
        save_path,
    )
    result.set_nested_config(nested_config)
    serialized_nested_config = self.serializer.serialize(nested_config)
    result.add_arg(f"nested_config={serialized_nested_config}", tabs=1)
    return result

group_manager_agent_extas ¶

Group manager agent configuration processor.

GroupManagerProcessor ¶

GroupManagerProcessor(
    agent: WaldiezAgent,
    initial_chats: list[WaldiezAgentConnection],
    group_chat_members: list[WaldiezAgent],
    agent_names: dict[str, str],
    model_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Callable[..., str],
    cache_seed: Optional[int] = None,
)

Processes group manager configurations.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The group manager agent to process.	required
`initial_chats`	`list[WaldiezAgentConnection]`	The initial chats to process for the group manager.	required
`group_chat_members`	`list[WaldiezAgent]`	The members of the group chat.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to their names.	required
`model_names`	`dict[str, str]`	Mapping of model IDs to their names.	required
`all_models`	`list[WaldiezModel]`	List of all models available for the agent.	required
`serializer`	`Callable[..., str]`	Function to serialize data into a string format.	required
`cache_seed`	`Optional[int]`	Seed for caching purposes, by default None.	`None`

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

def __init__(
    self,
    agent: WaldiezAgent,
    initial_chats: list[WaldiezAgentConnection],
    group_chat_members: list[WaldiezAgent],
    agent_names: dict[str, str],
    model_names: dict[str, str],
    all_models: list[WaldiezModel],
    serializer: Callable[..., str],
    cache_seed: Optional[int] = None,
):
    """Initialize the group manager processor.

    Parameters
    ----------
    agent : WaldiezAgent
        The group manager agent to process.
    initial_chats : list[WaldiezAgentConnection]
        The initial chats to process for the group manager.
    group_chat_members : list[WaldiezAgent]
        The members of the group chat.
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    model_names : dict[str, str]
        Mapping of model IDs to their names.
    all_models : list[WaldiezModel]
        List of all models available for the agent.
    serializer : Callable[..., str]
        Function to serialize data into a string format.
    cache_seed : Optional[int], optional
        Seed for caching purposes, by default None.
    """
    self.agent = agent
    self.initial_chats = initial_chats
    self.group_chat_members = group_chat_members
    self.agent_names = agent_names
    self.model_names = model_names
    self.all_models = all_models
    self.serializer = serializer
    self.cache_seed = cache_seed

is_pattern_strategy `staticmethod` ¶

is_pattern_strategy(
    initial_chats: list[WaldiezAgentConnection],
) -> bool

Check if the group manager should use pattern strategy.

Parameters:

Name	Type	Description	Default
`initial_chats`	`list[WaldiezAgentConnection]`	The initial chats to determine the strategy.	required

Returns:

Type	Description
`bool`	True if pattern strategy should be used, False otherwise.

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

@staticmethod
def is_pattern_strategy(
    initial_chats: list[WaldiezAgentConnection],
) -> bool:
    """Check if the group manager should use pattern strategy.

    Parameters
    ----------
    initial_chats : list[WaldiezAgentConnection]
        The initial chats to determine the strategy.

    Returns
    -------
    bool
        True if pattern strategy should be used, False otherwise.
    """
    if not initial_chats:
        return True

    first_chat = initial_chats[0]["chat"]
    return (
        isinstance(first_chat.data.message, str)
        or not first_chat.data.message.is_method()
    )

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> GroupManagerExtras

Process group manager and return extras.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, by default None.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, by default None.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, by default None.	`None`

Returns:

Type	Description
`GroupManagerExtras`	The processed group manager extras containing configuration.

Source code in waldiez/exporting/agent/extras/group_manager_agent_extas.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> GroupManagerExtras:
    """Process group manager and return extras.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, by default None.
    termination_config : TerminationConfig, optional
        Configuration for termination, by default None.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, by default None.

    Returns
    -------
    GroupManagerExtras
        The processed group manager extras containing configuration.
    """
    extras = GroupManagerExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )

    # Determine strategy based on initial chats and message types
    extras.strategy = self._determine_strategy()

    if extras.strategy == GroupManagerStrategy.PATTERN:
        self._process_pattern_strategy(extras)
    else:
        self._process_traditional_strategy(extras)

    return extras

group_member_extras ¶

Group member agent configuration processor.

GroupMemberAgentProcessor ¶

GroupMemberAgentProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    tool_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
)

Processor for group member agent configuration.

Source code in waldiez/exporting/agent/extras/group_member_extras.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    tool_names: dict[str, str],
    all_chats: list[WaldiezChat],
    chat_names: dict[str, str],
    serializer: Callable[..., str],
) -> None:
    self.agent = agent
    self.agent_names = agent_names
    self.tool_names = tool_names
    self.all_chats = all_chats
    self.chat_names = chat_names
    self.serializer = serializer

process ¶

process() -> GroupMemberProcessorResult

Process group member agent configuration.

Returns:

Type	Description
`GroupMemberProcessorResult`	The processed result containing before_agent code, after_agent code, and extra imports.

Source code in waldiez/exporting/agent/extras/group_member_extras.py

def process(
    self,
) -> GroupMemberProcessorResult:
    """Process group member agent configuration.

    Returns
    -------
    GroupMemberProcessorResult
        The processed result containing before_agent code,
        after_agent code, and extra imports.
    """
    self.result = GroupMemberProcessorResult()
    if not self.agent.is_group_member:
        return self.result
    self._process_agent_functions()
    self._process_agent_update_state_before_reply()
    self._process_handoff_registrations()
    return self.result

GroupMemberProcessorResult `dataclass` ¶

GroupMemberProcessorResult(
    before_agent: str = "",
    after_agent: str = "",
    extra_arguments: list[InstanceArgument] = list[
        InstanceArgument
    ](),
    extra_imports: set[ImportStatement] = set[
        ImportStatement
    ](),
)

Result from processing group member agent configuration.

handoffs ¶

Utility functions for generating agent related strings.

AfterWorkProcessor ¶

AfterWorkProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for after-work configurations.

Source code in waldiez/exporting/agent/extras/handoffs/after_work.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
):
    """Initialize the processor."""
    self.agent = agent
    self.agent_names = agent_names
    self.agent_name = agent_names[agent.id]
    self.chat_names = chat_names
    self.all_chats = all_chats
    self.serializer = serializer

process ¶

process(
    after_work: WaldiezTransitionTarget,
) -> AfterWorkResult

Process after-work configuration.

Parameters:

Name	Type	Description	Default
`after_work`	`WaldiezTransitionTarget`	The after-work transition target to process.	required

Returns:

Type	Description
`AfterWorkResult`	The processed result containing the registration string, before_content code, and extra imports.

Source code in waldiez/exporting/agent/extras/handoffs/after_work.py

def process(self, after_work: WaldiezTransitionTarget) -> AfterWorkResult:
    """Process after-work configuration.

    Parameters
    ----------
    after_work : WaldiezTransitionTarget
        The after-work transition target to process.

    Returns
    -------
    AfterWorkResult
        The processed result containing the registration string,
        before_content code, and extra imports.
    """
    target_processor = TransitionTargetProcessor(
        agent=self.agent,
        agent_names=self.agent_names,
        chat_names=self.chat_names,
        all_chats=self.all_chats,
        serializer=self.serializer,
    )
    target_result = target_processor.process(after_work)
    return AfterWorkResult(
        content=target_result.content,
        before_content=target_result.before_content,
        extra_imports=target_result.extra_imports,
    )

AfterWorkResult `dataclass` ¶

AfterWorkResult(
    content: str = "",
    before_content: str = "",
    extra_imports: set[str] = set[str](),
)

Result from processing after-work configuration.

Attributes:

Name	Type	Description
`registration`	`str`	Code to register the after-work transition.
`after_agent`	`str`	Code to be placed after the agent definition.
`extra_imports`	`set[str]`	Additional imports required for the after-work transition.

HandoffProcessor ¶

HandoffProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for agent handoffs.

Source code in waldiez/exporting/agent/extras/handoffs/handoff.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> None:
    self.agent = agent
    self.agent_names = agent_names
    self.agent_name = agent_names[agent.id]
    self.chat_names = chat_names
    self.all_chats = all_chats
    self.serializer = serializer
    self.result = HandoffResult()

process ¶

process() -> HandoffResult

Process all handoff configurations.

Returns:

Type	Description
`HandoffResult`	The processed handoff result containing after_agent code, and extra imports.

Source code in waldiez/exporting/agent/extras/handoffs/handoff.py

def process(self) -> HandoffResult:
    """Process all handoff configurations.

    Returns
    -------
    HandoffResult
        The processed handoff result containing after_agent code,
        and extra imports.
    """
    for handoff in self.agent.handoffs:
        if handoff.is_empty():
            continue
        target = TransitionTargetProcessor(
            agent=self.agent,
            agent_names=self.agent_names,
            chat_names=self.chat_names,
            all_chats=self.all_chats,
            serializer=self.serializer,
        ).process(target=handoff.target)
        self.result.extra_imports.update(target.extra_imports)
        if target.before_content:
            self.result.after_agent += f"{target.before_content}\n"
        condition = ConditionProcessor(
            agent=self.agent,
            agent_names=self.agent_names,
            chat_names=self.chat_names,
            all_chats=self.all_chats,
            serializer=self.serializer,
        ).process(condition=handoff.condition)
        self.result.extra_imports.update(condition.extra_imports)
        available = TransitionAvailableProcessor(
            serializer=self.serializer,
        ).process(
            available=handoff.available,
        )
        registration = self._create_handoff_registration(
            handoff=handoff,
            target=target,
            condition=condition,
            available=available,
        )
        self.result.extra_imports.update(available.extra_imports)
        if registration:
            self.result.after_agent += f"{registration}\n"

    if self.agent.data.after_work:
        after_work_result = self._process_after_work(
            self.agent.data.after_work
        )
        if after_work_result.before_content:
            self.result.after_agent += (
                f"{after_work_result.before_content}\n"
            )
        self.result.after_agent += after_work_result.content
        self.result.extra_imports.update(after_work_result.extra_imports)
    return self.result

HandoffResult `dataclass` ¶

HandoffResult(
    after_agent: str = "",
    extra_imports: set[str] = set[str](),
)

Result from processing handoffs.

Attributes:

Name	Type	Description
`Content to be placed before the agent definition.`
`after_agent`	`str`	Content to be placed after the agent definition.
`extra_imports`	`set[str]`	Additional imports required for the handoff.

TargetResult `dataclass` ¶

TargetResult(
    content: str = "",
    before_content: str = "",
    extra_imports: Set[str] = set(),
)

Result from processing a transition target.

Attributes:

Name	Type	Description
`content`	`str`	Content to be used for the target processing.
`before_content`	`str`	Content to be placed before the target processing.
`extra_imports`	`Set[str]`	Additional imports required for the target processing.

TransitionTargetProcessor ¶

TransitionTargetProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for transition targets.

Parameters:

Name	Type	Description	Default
`agent_names`	`dict[str, str]`	Mapping of agent IDs to their names.	required

Source code in waldiez/exporting/agent/extras/handoffs/target.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> None:
    """Initialize the processor with agent names and context.

    Parameters
    ----------
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    """
    self.agent_names = agent_names
    self.chat_names = chat_names
    self.agent = agent
    self.agent_name = agent_names[agent.id]
    self.all_chats = all_chats
    self.serializer = serializer

process ¶

process(target: WaldiezTransitionTarget) -> TargetResult

Process transition target based on its type.

Parameters:

Name	Type	Description	Default
`target`	`WaldiezTransitionTarget`	The transition target to process.	required

Returns:

Type	Description
`TargetResult`	The processed result containing the target string, before_content code, and extra imports.

Raises:

Type	Description
`ValueError`	If the target type is unknown.

Source code in waldiez/exporting/agent/extras/handoffs/target.py

def process(self, target: WaldiezTransitionTarget) -> TargetResult:
    """Process transition target based on its type.

    Parameters
    ----------
    target : WaldiezTransitionTarget
        The transition target to process.

    Returns
    -------
    TargetResult
        The processed result containing the target string,
        before_content code, and extra imports.

    Raises
    ------
    ValueError
        If the target type is unknown.
    """
    result = TargetResult()
    where = "autogen.agentchat.group"
    if target.target_type == "GroupManagerTarget":
        where += ".targets.group_manager_target"
    what = target.target_type
    result.extra_imports.add(f"from {where} import {what}")

    processors: dict[str, Callable[[WaldiezTransitionTarget], str]] = {
        "AgentTarget": self._process_agent_target,
        "RandomAgentTarget": self._process_random_agent_target,
        "GroupChatTarget": self._process_group_chat_target,
        "NestedChatTarget": self._process_nested_chat_target,
        "AskUserTarget": self._process_simple_target,
        "GroupManagerTarget": self._process_simple_target,
        "RevertToUserTarget": self._process_simple_target,
        "StayTarget": self._process_simple_target,
        "TerminateTarget": self._process_simple_target,
    }

    processor = processors.get(target.target_type)
    if not processor:
        raise ValueError(f"Unknown target type: {target.target_type}")

    result.content = processor(target)

    # Special handling for nested chat targets
    if target.target_type == "NestedChatTarget":
        nested_result = self._process_nested_chat_target_full(target)
        result.before_content = nested_result.before_content

    return result

after_work ¶

After-work processor for Waldiez agents.

AfterWorkProcessor ¶

AfterWorkProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for after-work configurations.

Source code in waldiez/exporting/agent/extras/handoffs/after_work.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
):
    """Initialize the processor."""
    self.agent = agent
    self.agent_names = agent_names
    self.agent_name = agent_names[agent.id]
    self.chat_names = chat_names
    self.all_chats = all_chats
    self.serializer = serializer

process ¶

process(
    after_work: WaldiezTransitionTarget,
) -> AfterWorkResult

Process after-work configuration.

Parameters:

Name	Type	Description	Default
`after_work`	`WaldiezTransitionTarget`	The after-work transition target to process.	required

Returns:

Type	Description
`AfterWorkResult`	The processed result containing the registration string, before_content code, and extra imports.

Source code in waldiez/exporting/agent/extras/handoffs/after_work.py

def process(self, after_work: WaldiezTransitionTarget) -> AfterWorkResult:
    """Process after-work configuration.

    Parameters
    ----------
    after_work : WaldiezTransitionTarget
        The after-work transition target to process.

    Returns
    -------
    AfterWorkResult
        The processed result containing the registration string,
        before_content code, and extra imports.
    """
    target_processor = TransitionTargetProcessor(
        agent=self.agent,
        agent_names=self.agent_names,
        chat_names=self.chat_names,
        all_chats=self.all_chats,
        serializer=self.serializer,
    )
    target_result = target_processor.process(after_work)
    return AfterWorkResult(
        content=target_result.content,
        before_content=target_result.before_content,
        extra_imports=target_result.extra_imports,
    )

AfterWorkResult `dataclass` ¶

AfterWorkResult(
    content: str = "",
    before_content: str = "",
    extra_imports: set[str] = set[str](),
)

Result from processing after-work configuration.

Attributes:

Name	Type	Description
`registration`	`str`	Code to register the after-work transition.
`after_agent`	`str`	Code to be placed after the agent definition.
`extra_imports`	`set[str]`	Additional imports required for the after-work transition.

available ¶

Transition availability processor for Waldiez agents.

TransitionAvailableProcessor ¶

TransitionAvailableProcessor(
    serializer: Callable[..., str],
)

Processor for transition availability.

Source code in waldiez/exporting/agent/extras/handoffs/available.py

def __init__(
    self,
    serializer: Callable[..., str],
) -> None:
    """Initialize the processor."""
    self.serializer = serializer

process ¶

process(
    available: WaldiezTransitionAvailability,
) -> TransitionAvailableResult

Process the transition availability.

Parameters:

Name	Type	Description	Default
`available`	`WaldiezTransitionAvailability`	The transition availability to process.	required

Returns:

Type	Description
`TransitionAvailableResult`	The result of processing the transition availability.

Raises:

Type	Description
`ValueError`	If the transition type is unsupported.

Source code in waldiez/exporting/agent/extras/handoffs/available.py

def process(
    self,
    available: WaldiezTransitionAvailability,
) -> TransitionAvailableResult:
    """Process the transition availability.

    Parameters
    ----------
    available : WaldiezTransitionAvailability
        The transition availability to process.

    Returns
    -------
    TransitionAvailableResult
        The result of processing the transition availability.

    Raises
    ------
    ValueError
        If the transition type is unsupported.
    """
    result = TransitionAvailableResult()
    if available.type == "none":
        return result
    import_prefix = "from autogen.agentchat.group import "
    if available.type == "string":
        content = self.serializer(available.value)
        result.content = f"StringAvailableCondition({content})"
        result.extra_imports.add(f"{import_prefix}StringAvailableCondition")
        return result
    if available.type == "expression":
        content = self.serializer(available.value)
        result.content = (
            "ExpressionAvailableCondition(\n"
            f"            expression=ContextExpression({content})\n"
            "        )"
        )
        result.extra_imports.add(
            f"{import_prefix}ExpressionAvailableCondition"
        )
        result.extra_imports.add(f"{import_prefix}ContextExpression")
        return result
    raise ValueError(
        f"Unsupported transition availability type: {available.type}"
    )

TransitionAvailableResult `dataclass` ¶

TransitionAvailableResult(
    content: str = "", extra_imports: set[str] = set[str]()
)

Result from processing transition availability.

condition ¶

Handoff condition processing for Waldiez agents.

ConditionProcessor ¶

ConditionProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for handoff conditions.

Source code in waldiez/exporting/agent/extras/handoffs/condition.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> None:
    """Initialize the processor with agent context."""
    self.agent = agent
    self.agent_names = agent_names
    self.agent_name = agent_names[agent.id]
    self.chat_names = chat_names
    self.all_chats = all_chats
    self.serializer = serializer
    self.result = ConditionResult()

process ¶

process(
    condition: WaldiezHandoffCondition,
) -> ConditionResult

Process handoff conditions.

Parameters:

Name	Type	Description	Default
`condition`	`WaldiezHandoffCondition`	The handoff condition to process.	required

Returns:

Type	Description
`ConditionResult`	The processed result containing the relevant code and extra imports.

Raises:

Type	Description
`ValueError`	If the condition type is unsupported.

Source code in waldiez/exporting/agent/extras/handoffs/condition.py

def process(self, condition: WaldiezHandoffCondition) -> ConditionResult:
    """Process handoff conditions.

    Parameters
    ----------
    condition : WaldiezHandoffCondition
        The handoff condition to process.

    Returns
    -------
    ConditionResult
        The processed result containing the relevant code and extra imports.

    Raises
    ------
    ValueError
        If the condition type is unsupported.
    """
    if isinstance(condition, WaldiezStringLLMCondition):
        self._process_string_llm_condition(condition)
    elif isinstance(condition, WaldiezContextStrLLMCondition):
        self._process_context_str_llm_condition(condition)
    elif isinstance(condition, WaldiezStringContextCondition):
        self._process_string_context_condition(condition)
    elif isinstance(
        condition,
        WaldiezExpressionContextCondition,
    ):  # pyright: ignore
        self._process_expression_context_condition(condition)
    else:
        raise ValueError(f"Unsupported condition type: {type(condition)}")
    return self.result

ConditionResult `dataclass` ¶

ConditionResult(
    content: str = "", extra_imports: set[str] = set[str]()
)

Result from processing handoff conditions.

Attributes:

Name	Type	Description
`content`	`str`	Content to be used for the handoff condition.
`extra_imports`	`set[str]`	Additional imports required for the handoff.

handoff ¶

Handoff processor for Waldiez agents.

HandoffProcessor ¶

HandoffProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for agent handoffs.

Source code in waldiez/exporting/agent/extras/handoffs/handoff.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> None:
    self.agent = agent
    self.agent_names = agent_names
    self.agent_name = agent_names[agent.id]
    self.chat_names = chat_names
    self.all_chats = all_chats
    self.serializer = serializer
    self.result = HandoffResult()

process ¶

process() -> HandoffResult

Process all handoff configurations.

Returns:

Type	Description
`HandoffResult`	The processed handoff result containing after_agent code, and extra imports.

Source code in waldiez/exporting/agent/extras/handoffs/handoff.py

def process(self) -> HandoffResult:
    """Process all handoff configurations.

    Returns
    -------
    HandoffResult
        The processed handoff result containing after_agent code,
        and extra imports.
    """
    for handoff in self.agent.handoffs:
        if handoff.is_empty():
            continue
        target = TransitionTargetProcessor(
            agent=self.agent,
            agent_names=self.agent_names,
            chat_names=self.chat_names,
            all_chats=self.all_chats,
            serializer=self.serializer,
        ).process(target=handoff.target)
        self.result.extra_imports.update(target.extra_imports)
        if target.before_content:
            self.result.after_agent += f"{target.before_content}\n"
        condition = ConditionProcessor(
            agent=self.agent,
            agent_names=self.agent_names,
            chat_names=self.chat_names,
            all_chats=self.all_chats,
            serializer=self.serializer,
        ).process(condition=handoff.condition)
        self.result.extra_imports.update(condition.extra_imports)
        available = TransitionAvailableProcessor(
            serializer=self.serializer,
        ).process(
            available=handoff.available,
        )
        registration = self._create_handoff_registration(
            handoff=handoff,
            target=target,
            condition=condition,
            available=available,
        )
        self.result.extra_imports.update(available.extra_imports)
        if registration:
            self.result.after_agent += f"{registration}\n"

    if self.agent.data.after_work:
        after_work_result = self._process_after_work(
            self.agent.data.after_work
        )
        if after_work_result.before_content:
            self.result.after_agent += (
                f"{after_work_result.before_content}\n"
            )
        self.result.after_agent += after_work_result.content
        self.result.extra_imports.update(after_work_result.extra_imports)
    return self.result

HandoffResult `dataclass` ¶

HandoffResult(
    after_agent: str = "",
    extra_imports: set[str] = set[str](),
)

Result from processing handoffs.

Attributes:

Name	Type	Description
`Content to be placed before the agent definition.`
`after_agent`	`str`	Content to be placed after the agent definition.
`extra_imports`	`set[str]`	Additional imports required for the handoff.

target ¶

Transition target processor for Waldiez agents.

TargetResult `dataclass` ¶

TargetResult(
    content: str = "",
    before_content: str = "",
    extra_imports: Set[str] = set(),
)

Result from processing a transition target.

Attributes:

Name	Type	Description
`content`	`str`	Content to be used for the target processing.
`before_content`	`str`	Content to be placed before the target processing.
`extra_imports`	`Set[str]`	Additional imports required for the target processing.

TransitionTargetProcessor ¶

TransitionTargetProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
)

Processor for transition targets.

Parameters:

Name	Type	Description	Default
`agent_names`	`dict[str, str]`	Mapping of agent IDs to their names.	required

Source code in waldiez/exporting/agent/extras/handoffs/target.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    chat_names: dict[str, str],
    all_chats: list[WaldiezChat],
    serializer: Callable[..., str],
) -> None:
    """Initialize the processor with agent names and context.

    Parameters
    ----------
    agent_names : dict[str, str]
        Mapping of agent IDs to their names.
    """
    self.agent_names = agent_names
    self.chat_names = chat_names
    self.agent = agent
    self.agent_name = agent_names[agent.id]
    self.all_chats = all_chats
    self.serializer = serializer

process ¶

process(target: WaldiezTransitionTarget) -> TargetResult

Process transition target based on its type.

Parameters:

Name	Type	Description	Default
`target`	`WaldiezTransitionTarget`	The transition target to process.	required

Returns:

Type	Description
`TargetResult`	The processed result containing the target string, before_content code, and extra imports.

Raises:

Type	Description
`ValueError`	If the target type is unknown.

Source code in waldiez/exporting/agent/extras/handoffs/target.py

def process(self, target: WaldiezTransitionTarget) -> TargetResult:
    """Process transition target based on its type.

    Parameters
    ----------
    target : WaldiezTransitionTarget
        The transition target to process.

    Returns
    -------
    TargetResult
        The processed result containing the target string,
        before_content code, and extra imports.

    Raises
    ------
    ValueError
        If the target type is unknown.
    """
    result = TargetResult()
    where = "autogen.agentchat.group"
    if target.target_type == "GroupManagerTarget":
        where += ".targets.group_manager_target"
    what = target.target_type
    result.extra_imports.add(f"from {where} import {what}")

    processors: dict[str, Callable[[WaldiezTransitionTarget], str]] = {
        "AgentTarget": self._process_agent_target,
        "RandomAgentTarget": self._process_random_agent_target,
        "GroupChatTarget": self._process_group_chat_target,
        "NestedChatTarget": self._process_nested_chat_target,
        "AskUserTarget": self._process_simple_target,
        "GroupManagerTarget": self._process_simple_target,
        "RevertToUserTarget": self._process_simple_target,
        "StayTarget": self._process_simple_target,
        "TerminateTarget": self._process_simple_target,
    }

    processor = processors.get(target.target_type)
    if not processor:
        raise ValueError(f"Unknown target type: {target.target_type}")

    result.content = processor(target)

    # Special handling for nested chat targets
    if target.target_type == "NestedChatTarget":
        nested_result = self._process_nested_chat_target_full(target)
        result.before_content = nested_result.before_content

    return result

rag ¶

Agent exporter rag extras.

VectorDBExtras `dataclass` ¶

VectorDBExtras(
    before_arg: str,
    vector_db_arg: str,
    imports: set[str] = set[str](),
)

Vector DB exporting extras for RAG user agents.

chroma_extras ¶

Get chroma db related imports and content.

get_chroma_db_args ¶

get_chroma_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str, str]

Get the 'kwargs to use for ChromaVectorDB.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[str, Set[str], str]`	The 'kwargs' string. What to import. The custom embedding function. Any additional content to be used before the `kwargs` string.

Source code in waldiez/exporting/agent/extras/rag/chroma_extras.py

def get_chroma_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str, str]:
    """Get the 'kwargs to use for ChromaVectorDB.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[str, Set[str], str]

        - The 'kwargs' string.
        - What to import.
        - The custom embedding function.
        - Any additional content to be used before the `kwargs` string.
    """
    client_str, client_to_import = _get_chroma_client_string(agent)
    embedding_function_arg, to_import_embedding, embedding_function_body = (
        _get_chroma_embedding_function_string(agent, agent_name)
    )
    to_import = {client_to_import, "from chromadb.config import Settings"}
    if to_import_embedding:
        to_import.add(to_import_embedding)
    kwarg_string = (
        f"            client={agent_name}_client," + "\n"
        f"            embedding_function={embedding_function_arg}," + "\n"
    )
    # The RAG example:
    # https://ag2ai.github.io/ag2/docs/notebooks/agentchat_groupchat_RAG/
    # raises `InvalidCollectionException`: Collection groupchat does not exist.
    # https://github.com/chroma-core/chroma/issues/861
    # https://github.com/microsoft/autogen/issues/3551#issuecomment-2366930994
    # manually initializing the collection before running the flow,
    # might be a workaround.
    content_before = f"{agent_name}_client = {client_str}" + "\n"
    vector_db_model = agent.retrieve_config.db_config.model
    if not embedding_function_body:
        content_before += (
            f"{agent_name}_embedding_function = "
            "SentenceTransformerEmbeddingFunction(\n"
            f'    model_name="{vector_db_model}",' + "\n"
            ")\n"
        )
    collection_name = agent.retrieve_config.collection_name
    get_or_create = agent.retrieve_config.get_or_create
    if collection_name:
        if get_or_create:
            content_before += (
                f"{agent_name}_client.get_or_create_collection" + "(\n"
                f'    "{collection_name}",' + "\n"
                f"    embedding_function={embedding_function_arg}," + "\n"
                ")" + "\n"
            )
        else:
            content_before += (
                "try:\n"
                f"    {agent_name}_client.get_collection(" + "\n"
                f'        "{collection_name}",' + "\n"
                f"        embedding_function={embedding_function_arg}," + "\n"
                "    )\n"
                "except ValueError:\n"
                f"    {agent_name}_client.create_collection(" + "\n"
                f'        "{collection_name}",' + "\n"
                f"        embedding_function={embedding_function_arg}," + "\n"
                "    )\n"
            )
    return kwarg_string, to_import, embedding_function_body, content_before

get_vector_db_extras ¶

get_vector_db_extras(
    agent: WaldiezRagUserProxy, agent_name: str
) -> VectorDBExtras

Get the RAG user vector db string.

The vector db can be one of the following: "vector_db": ChromaVectorDB(....) "vector_db": QdrantVectorDB(....) "vector_db": MongoDBAtlasVectorDB(....) "vector_db": PGVectorDB(....)

If a custom embedding function is to be used, it's name will be in the arg and its definition will be before the arg.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[str, str, Set[str]]`	The content before the arg if any, the arg and the related imports.

Source code in waldiez/exporting/agent/extras/rag/vector_db_extras.py

def get_vector_db_extras(
    agent: WaldiezRagUserProxy,
    agent_name: str,
) -> VectorDBExtras:
    """Get the RAG user vector db string.

    The vector db can be one of the following:
    "vector_db": ChromaVectorDB(....)
    "vector_db": QdrantVectorDB(....)
    "vector_db": MongoDBAtlasVectorDB(....)
    "vector_db": PGVectorDB(....)

    If a custom embedding function is to be used,
    it's name will be in the arg and its definition will be before the arg.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[str, str, Set[str]]
        The content before the arg if any, the arg and the related imports.
    """
    before = ""
    imports: set[str] = set()
    ef_body: str = ""
    db_imports: set[str] = set()
    kwarg_string = ""
    content_before = ""
    vdb_class = "ChromaVectorDB"
    if agent.retrieve_config.vector_db == "chroma":
        imports.add(
            "from autogen.agentchat.contrib.vectordb.chromadb import ChromaVectorDB"
        )
        kwarg_string, db_imports, ef_body, content_before = get_chroma_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "qdrant":
        vdb_class = "QdrantVectorDB"
        imports.add(
            "from autogen.agentchat.contrib.vectordb.qdrant import QdrantVectorDB"
        )
        kwarg_string, db_imports, ef_body = get_qdrant_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "mongodb":
        vdb_class = "MongoDBAtlasVectorDB"
        imports.add(
            "from autogen.agentchat.contrib.vectordb.mongodb import MongoDBAtlasVectorDB"
        )
        kwarg_string, db_imports, ef_body = get_mongodb_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "pgvector":
        imports.add(
            "from autogen.agentchat.contrib.vectordb.pgvectordb import PGVectorDB"
        )
        vdb_class = "PGVectorDB"
        kwarg_string, db_imports, ef_body = get_pgvector_db_args(
            agent, agent_name
        )
    if content_before:
        before += "\n" + f"{content_before}"
    if ef_body:
        before += "\n" + f"{ef_body}" + "\n"
    imports.update(db_imports)
    kwarg_string += _get_metadata_arg(agent)
    vdb_arg = f"{vdb_class}(" + "\n"
    vdb_arg += kwarg_string + "        )"
    return VectorDBExtras(
        before_arg=before,
        vector_db_arg=vdb_arg,
        imports=imports,
    )

mongo_extras ¶

Get mongodb related content and imports.

get_mongodb_db_args ¶

get_mongodb_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]

Get the kwargs to use for MongoDBAtlasVectorDB.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[str, Set[str], str]`	The kwargs to use, what to import and the custom_embedding_function.

Source code in waldiez/exporting/agent/extras/rag/mongo_extras.py

def get_mongodb_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]:
    """Get the kwargs to use for MongoDBAtlasVectorDB.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[str, Set[str], str]
        The kwargs to use, what to import and the custom_embedding_function.
    """
    embedding_function_arg, to_import_embedding, embedding_function_body = (
        _get_mongodb_embedding_function_string(agent, agent_name)
    )
    to_import: Set[str] = (
        set() if not to_import_embedding else {to_import_embedding}
    )
    tab = " " * 12
    db_config = agent.retrieve_config.db_config
    kwarg_string = (
        f'{tab}connection_string="{db_config.connection_url}",' + "\n"
        f"{tab}embedding_function={embedding_function_arg}," + "\n"
    )
    wait_until_document_ready = db_config.wait_until_document_ready
    wait_until_index_ready = db_config.wait_until_index_ready
    if wait_until_document_ready is not None:
        kwarg_string += (
            f"{tab}wait_until_document_ready={wait_until_document_ready},"
            + "\n"
        )
    if wait_until_index_ready is not None:
        kwarg_string += (
            f"{tab}wait_until_index_ready={wait_until_index_ready}," + "\n"
        )
    return kwarg_string, to_import, embedding_function_body

pgvector_extras ¶

Get pgvector related content and imports.

get_pgvector_db_args ¶

get_pgvector_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]

Get the kwargs to use for PGVectorDB.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[tuple[str, str], Set[str], str]`	The kwargs to use, what to import and the custom_embedding_function.

Source code in waldiez/exporting/agent/extras/rag/pgvector_extras.py

def get_pgvector_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]:
    """Get the kwargs to use for PGVectorDB.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[tuple[str,str], Set[str], str]
        The kwargs to use, what to import and the custom_embedding_function.
    """
    client_str, to_import_client = _get_pgvector_client_string(agent)
    embedding_function_arg, to_import_embedding, embedding_function_body = (
        _get_pgvector_embedding_function_string(agent, agent_name)
    )
    to_import = (
        {to_import_client, to_import_embedding}
        if to_import_embedding
        else {to_import_client}
    )
    kwarg_str = (
        f"            client={client_str}," + "\n"
        f"            embedding_function={embedding_function_arg}," + "\n"
    )
    return kwarg_str, to_import, embedding_function_body

qdrant_extras ¶

Get qdrant db related imports and content.

get_qdrant_db_args ¶

get_qdrant_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]

Get the kwargs to use for QdrantVectorDB.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[str, Set[str], str]`	The kwargs to use, the imports and the embedding function body if used.

Source code in waldiez/exporting/agent/extras/rag/qdrant_extras.py

def get_qdrant_db_args(
    agent: WaldiezRagUserProxy, agent_name: str
) -> tuple[str, Set[str], str]:
    """Get the kwargs to use for QdrantVectorDB.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[str, Set[str], str]
        The kwargs to use, the imports and the embedding function body if used.
    """
    client_str, to_import_client = _get_qdrant_client_string(agent)
    embedding_function_arg, to_import_embedding, embedding_function_body = (
        _get_qdrant_embedding_function_string(agent, agent_name)
    )
    to_import = (
        {to_import_client, to_import_embedding}
        if to_import_embedding
        else {to_import_client}
    )
    kwarg_string = (
        f"            client={client_str}," + "\n"
        f"            embedding_function={embedding_function_arg}," + "\n"
    )
    return kwarg_string, to_import, embedding_function_body

vector_db_extras ¶

Vector DB exporting utils for RAG user agents.

VectorDBExtras `dataclass` ¶

VectorDBExtras(
    before_arg: str,
    vector_db_arg: str,
    imports: set[str] = set[str](),
)

Vector DB exporting extras for RAG user agents.

get_vector_db_extras ¶

get_vector_db_extras(
    agent: WaldiezRagUserProxy, agent_name: str
) -> VectorDBExtras

Get the RAG user vector db string.

The vector db can be one of the following: "vector_db": ChromaVectorDB(....) "vector_db": QdrantVectorDB(....) "vector_db": MongoDBAtlasVectorDB(....) "vector_db": PGVectorDB(....)

If a custom embedding function is to be used, it's name will be in the arg and its definition will be before the arg.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezRagUserProxy`	The agent.	required
`agent_name`	`str`	The agent's name.	required

Returns:

Type	Description
`tuple[str, str, Set[str]]`	The content before the arg if any, the arg and the related imports.

Source code in waldiez/exporting/agent/extras/rag/vector_db_extras.py

def get_vector_db_extras(
    agent: WaldiezRagUserProxy,
    agent_name: str,
) -> VectorDBExtras:
    """Get the RAG user vector db string.

    The vector db can be one of the following:
    "vector_db": ChromaVectorDB(....)
    "vector_db": QdrantVectorDB(....)
    "vector_db": MongoDBAtlasVectorDB(....)
    "vector_db": PGVectorDB(....)

    If a custom embedding function is to be used,
    it's name will be in the arg and its definition will be before the arg.

    Parameters
    ----------
    agent : WaldiezRagUserProxy
        The agent.
    agent_name : str
        The agent's name.

    Returns
    -------
    tuple[str, str, Set[str]]
        The content before the arg if any, the arg and the related imports.
    """
    before = ""
    imports: set[str] = set()
    ef_body: str = ""
    db_imports: set[str] = set()
    kwarg_string = ""
    content_before = ""
    vdb_class = "ChromaVectorDB"
    if agent.retrieve_config.vector_db == "chroma":
        imports.add(
            "from autogen.agentchat.contrib.vectordb.chromadb import ChromaVectorDB"
        )
        kwarg_string, db_imports, ef_body, content_before = get_chroma_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "qdrant":
        vdb_class = "QdrantVectorDB"
        imports.add(
            "from autogen.agentchat.contrib.vectordb.qdrant import QdrantVectorDB"
        )
        kwarg_string, db_imports, ef_body = get_qdrant_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "mongodb":
        vdb_class = "MongoDBAtlasVectorDB"
        imports.add(
            "from autogen.agentchat.contrib.vectordb.mongodb import MongoDBAtlasVectorDB"
        )
        kwarg_string, db_imports, ef_body = get_mongodb_db_args(
            agent, agent_name
        )
    if agent.retrieve_config.vector_db == "pgvector":
        imports.add(
            "from autogen.agentchat.contrib.vectordb.pgvectordb import PGVectorDB"
        )
        vdb_class = "PGVectorDB"
        kwarg_string, db_imports, ef_body = get_pgvector_db_args(
            agent, agent_name
        )
    if content_before:
        before += "\n" + f"{content_before}"
    if ef_body:
        before += "\n" + f"{ef_body}" + "\n"
    imports.update(db_imports)
    kwarg_string += _get_metadata_arg(agent)
    vdb_arg = f"{vdb_class}(" + "\n"
    vdb_arg += kwarg_string + "        )"
    return VectorDBExtras(
        before_arg=before,
        vector_db_arg=vdb_arg,
        imports=imports,
    )

rag_user_proxy_agent_extras ¶

RAG user proxy agent configuration processor.

RagUserProxyAgentProcessor ¶

RagUserProxyAgentProcessor(
    agent: WaldiezAgent,
    agent_name: str,
    model_names: dict[str, str],
    path_resolver: PathResolver | None = None,
    serializer: Serializer | None = None,
)

Processor for RAG user proxy agent configuration.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The Waldiez RAG user proxy agent to process.	required
`model_names`	`dict[str, str]`	A mapping from model id to model name.	required
`path_resolver`	`PathResolver \| None`	Optional path resolver for resolving paths. Defaults to DefaultPathResolver if not provided.	`None`
`serializer`	`Serializer \| None`	Optional serializer for the RAG configuration. Defaults to DefaultSerializer if not provided.	`None`

Source code in waldiez/exporting/agent/extras/rag_user_proxy_agent_extras.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_name: str,
    model_names: dict[str, str],
    path_resolver: PathResolver | None = None,
    serializer: Serializer | None = None,
):
    """Initialize the processor.

    Parameters
    ----------
    agent : WaldiezAgent
        The Waldiez RAG user proxy agent to process.
    model_names : dict[str, str]
        A mapping from model id to model name.
    path_resolver : PathResolver | None
        Optional path resolver for resolving paths.
        Defaults to DefaultPathResolver if not provided.
    serializer : Serializer | None
        Optional serializer for the RAG configuration.
        Defaults to DefaultSerializer if not provided.
    """
    self.agent = agent
    self.agent_name = agent_name
    self.model_names = model_names
    self.path_resolver = path_resolver or DefaultPathResolver()
    self.serializer = serializer or DefaultSerializer()

process ¶

process() -> RAGUserExtras

Process RAG user proxy agent configuration.

Returns:

Type	Description
`RAGConfig`	The processed result containing extra arguments, before content, imports, and environment variables.

Source code in waldiez/exporting/agent/extras/rag_user_proxy_agent_extras.py

def process(self) -> RAGUserExtras:
    """Process RAG user proxy agent configuration.

    Returns
    -------
    RAGConfig
        The processed result containing extra arguments, before content,
        imports, and environment variables.
    """
    result = RAGUserExtras(self.agent.id)
    if not self.agent.is_rag_user or not isinstance(
        self.agent, WaldiezRagUserProxy
    ):
        return result
    # Get the extra args
    vector_db_extras = get_vector_db_extras(
        agent=self.agent, agent_name=self.agent_name
    )
    before_agent, arg_value = self._get_retrieve_config(vector_db_extras)
    if before_agent:
        self._before_agent += before_agent
    if arg_value:
        retrieve_arg = InstanceArgument(
            instance_id=self.agent.id,
            name="retrieve_config",
            value=arg_value,
            tabs=1,
        )
        result.add_arg(retrieve_arg)
    for import_statement in vector_db_extras.imports:
        result.add_import(
            ImportStatement(
                statement=import_statement,
                position=ImportPosition.THIRD_PARTY,
            )
        )
    if self._before_agent:
        result.prepend_before_agent(self._before_agent)
    return result

reasoning_agent_extras ¶

Reasoning agent configuration processor.

ReasoningAgentProcessor ¶

ReasoningAgentProcessor(
    agent: WaldiezAgent, serializer: Serializer | None
)

Processor for reasoning agent configuration.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The Waldiez agent to process.	required
`serializer`	`Serializer \| None`	Optional serializer for the reasoning configuration. Defaults to DefaultSerializer if not provided.	required

Source code in waldiez/exporting/agent/extras/reasoning_agent_extras.py

def __init__(self, agent: WaldiezAgent, serializer: Serializer | None):
    """Initialize the processor with the agent and serializer.

    Parameters
    ----------
    agent : WaldiezAgent
        The Waldiez agent to process.
    serializer : Serializer | None
        Optional serializer for the reasoning configuration.
        Defaults to DefaultSerializer if not provided.
    """
    self.agent = agent
    self.serializer = serializer or DefaultSerializer()

process ¶

process(
    code_execution_config: Optional[
        CodeExecutionConfig
    ] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[
        SystemMessageConfig
    ] = None,
) -> ReasoningExtras

Process reasoning agent configuration.

Parameters:

Name	Type	Description	Default
`code_execution_config`	`CodeExecutionConfig`	Configuration for code execution, if applicable.	`None`
`termination_config`	`TerminationConfig`	Configuration for termination, if applicable.	`None`
`system_message_config`	`SystemMessageConfig`	Configuration for system messages, if applicable.	`None`

Returns:

Type	Description
`ReasoningExtras`	The processed result containing extra arguments, before content, imports, and environment variables.

Source code in waldiez/exporting/agent/extras/reasoning_agent_extras.py

def process(
    self,
    code_execution_config: Optional[CodeExecutionConfig] = None,
    termination_config: Optional[TerminationConfig] = None,
    system_message_config: Optional[SystemMessageConfig] = None,
) -> ReasoningExtras:
    """Process reasoning agent configuration.

    Parameters
    ----------
    code_execution_config : CodeExecutionConfig, optional
        Configuration for code execution, if applicable.
    termination_config : TerminationConfig, optional
        Configuration for termination, if applicable.
    system_message_config : SystemMessageConfig, optional
        Configuration for system messages, if applicable.

    Returns
    -------
    ReasoningExtras
        The processed result containing extra arguments, before content,
        imports, and environment variables.
    """
    result = ReasoningExtras(
        instance_id=self.agent.id,
        code_execution_config=code_execution_config,
        termination_config=termination_config,
        system_message_config=system_message_config,
    )
    if not self.agent.is_reasoning or not isinstance(
        self.agent, WaldiezReasoningAgent
    ):  # pragma: no cover
        return result

    reasoning_config = self.agent.get_reasoning_config()
    serialized = self.serializer.serialize(reasoning_config)
    reason_arg = InstanceArgument(
        instance_id=self.agent.id,
        name="reason_config",
        value=serialized,
        tabs=1,
    )
    result.add_arg(reason_arg)
    verbose_arg = InstanceArgument(
        instance_id=self.agent.id,
        name="verbose",
        value=self.agent.data.verbose,
        tabs=1,
    )
    result.add_arg(verbose_arg)
    return result

factory ¶

Factory for creating agent exporter.

create_agent_exporter ¶

create_agent_exporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    initial_chats: list[WaldiezAgentConnection],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[
        Callable[[WaldiezAgent], list[str]]
    ] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any
) -> AgentExporter

Create an agent exporter.

Parameters:

Name	Type	Description	Default
`agent`	`WaldiezAgent`	The agent to export.	required
`agent_names`	`dict[str, str]`	Mapping of agent IDs to names.	required
`models`	`tuple[list[WaldiezModel], dict[str, str]]`	All models and model names mapping.	required
`chats`	`tuple[list[WaldiezChat], dict[str, str]]`	All chats and chat names mapping.	required
`tool_names`	`dict[str, str]`	Mapping of tool IDs to names.	required
`is_async`	`bool`	Whether the flow is async, by default False	`False`
`for_notebook`	`bool`	Whether exporting for notebook, by default False	`False`
`cache_seed`	`Optional[int]`	Cache seed if any, by default None	`None`
`initial_chats`	`Optional[list[WaldiezAgentConnection]]`	Initial chats for group managers, by default None	required
`group_chat_members`	`Optional[list[WaldiezAgent]]`	Group chat members if group manager, by default None	`None`
`arguments_resolver`	`Optional[Callable]`	Function to resolve additional arguments, by default None	`None`
`output_dir`	`Optional[Union[str, Path]]`	Output directory for generated files, by default None	`None`
`context`	`Optional[ExporterContext]`	Exporter context with dependencies, by default None	`None`
`**kwargs`	`Any`	Additional keyword arguments.	`{}`

Returns:

Type	Description
`AgentExporter`	The created agent exporter.

Source code in waldiez/exporting/agent/factory.py

def create_agent_exporter(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    models: tuple[list[WaldiezModel], dict[str, str]],
    chats: tuple[list[WaldiezChat], dict[str, str]],
    tool_names: dict[str, str],
    initial_chats: list[WaldiezAgentConnection],
    is_async: bool = False,
    for_notebook: bool = False,
    cache_seed: Optional[int] = None,
    group_chat_members: Optional[list[WaldiezAgent]] = None,
    arguments_resolver: Optional[Callable[[WaldiezAgent], list[str]]] = None,
    output_dir: Optional[Union[str, Path]] = None,
    context: Optional[ExporterContext] = None,
    **kwargs: Any,
) -> AgentExporter:
    """Create an agent exporter.

    Parameters
    ----------
    agent : WaldiezAgent
        The agent to export.
    agent_names : dict[str, str]
        Mapping of agent IDs to names.
    models : tuple[list[WaldiezModel], dict[str, str]]
        All models and model names mapping.
    chats : tuple[list[WaldiezChat], dict[str, str]]
        All chats and chat names mapping.
    tool_names : dict[str, str]
        Mapping of tool IDs to names.
    is_async : bool, optional
        Whether the flow is async, by default False
    for_notebook : bool, optional
        Whether exporting for notebook, by default False
    cache_seed : Optional[int], optional
        Cache seed if any, by default None
    initial_chats : Optional[list[WaldiezAgentConnection]], optional
        Initial chats for group managers, by default None
    group_chat_members : Optional[list[WaldiezAgent]], optional
        Group chat members if group manager, by default None
    arguments_resolver : Optional[Callable], optional
        Function to resolve additional arguments, by default None
    output_dir : Optional[Union[str, Path]], optional
        Output directory for generated files, by default None
    context : Optional[ExporterContext], optional
        Exporter context with dependencies, by default None
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    AgentExporter
        The created agent exporter.
    """
    if context is None:
        context = get_default_exporter_context()
    return AgentExporter(
        agent=agent,
        agent_names=agent_names,
        models=models,
        chats=chats,
        tool_names=tool_names,
        is_async=is_async,
        for_notebook=for_notebook,
        cache_seed=cache_seed,
        initial_chats=initial_chats,
        group_chat_members=group_chat_members,
        arguments_resolver=arguments_resolver,
        output_dir=output_dir,
        context=context,
        **kwargs,
    )

processor ¶

Processor for generating Waldiez agent definitions.

AgentProcessingResult `dataclass` ¶

AgentProcessingResult(content: str = '')

Result from processing an agent.

AgentProcessor ¶

AgentProcessor(
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    arguments_resolver: Callable[[WaldiezAgent], list[str]],
    extras: StandardExtras,
)

Processor for main agent definition generation.

Source code in waldiez/exporting/agent/processor.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_names: dict[str, str],
    arguments_resolver: Callable[[WaldiezAgent], list[str]],
    extras: StandardExtras,
):
    self.agent = agent
    self.agent_names = agent_names
    self.arguments_resolver = arguments_resolver
    self.extras = extras

get_auto_reply_arg ¶

get_auto_reply_arg() -> str

Get the default auto reply argument.

Returns:

Type	Description
`str`	The default auto reply argument.

Source code in waldiez/exporting/agent/processor.py

def get_auto_reply_arg(self) -> str:
    """Get the default auto reply argument.

    Returns
    -------
    str
        The default auto reply argument.
    """
    default_auto_reply = '""'
    if self.agent.data.agent_default_auto_reply:
        # Escape the default auto reply string
        default_auto_reply = json.dumps(
            self.agent.data.agent_default_auto_reply
        )
    return f"    default_auto_reply={default_auto_reply}"

get_description_arg ¶

get_description_arg() -> str

Get the agent description.

Returns:

Type	Description
`str`	The agent description.

Source code in waldiez/exporting/agent/processor.py

def get_description_arg(self) -> str:
    """Get the agent description.

    Returns
    -------
    str
        The agent description.
    """
    description = (
        json.dumps(self.agent.description)
        if self.agent.description
        else '""'
    )
    return f"    description={description}"

get_human_input_mode_arg ¶

get_human_input_mode_arg() -> str

Get the human input mode argument.

Returns:

Type	Description
`str`	The human input mode argument.

Source code in waldiez/exporting/agent/processor.py

def get_human_input_mode_arg(self) -> str:
    """Get the human input mode argument.

    Returns
    -------
    str
        The human input mode argument.
    """
    return f'    human_input_mode="{self.agent.data.human_input_mode}"'

get_max_consecutive_auto_reply_arg ¶

get_max_consecutive_auto_reply_arg() -> str

Get the maximum consecutive auto reply argument.

Returns:

Type	Description
`str`	The maximum consecutive auto reply argument.

Source code in waldiez/exporting/agent/processor.py

def get_max_consecutive_auto_reply_arg(self) -> str:
    """Get the maximum consecutive auto reply argument.

    Returns
    -------
    str
        The maximum consecutive auto reply argument.
    """
    value = self.agent.data.max_consecutive_auto_reply
    return f"    max_consecutive_auto_reply={value}"

get_name_arg ¶

get_name_arg() -> str

Get the agent name argument.

Returns:

Type	Description
`str`	The agent name argument.

Source code in waldiez/exporting/agent/processor.py

def get_name_arg(self) -> str:
    """Get the agent name argument.

    Returns
    -------
    str
        The agent name argument.
    """
    return f'    name="{self.agent_names[self.agent.id]}",'

process ¶

process() -> AgentProcessingResult

Process the agent and generate its definition.

Returns:

Type	Description
`str`	The generated agent definition string.

Source code in waldiez/exporting/agent/processor.py

    def process(self) -> AgentProcessingResult:
        """Process the agent and generate its definition.

        Returns
        -------
        str
            The generated agent definition string.
        """
        if (
            isinstance(self.extras, GroupManagerExtras)
            and self._should_skip_group_manager()
        ):
            return AgentProcessingResult(content="")
        agent_name = self.agent_names[self.agent.id]

        ag2_class = self.agent.ag2_class
        # Build extra arguments
        extra_args = ""
        for arg in self.extras.extra_args:
            extra_args += arg.get_content(append_new_line=True)

        # Get additional arguments from resolver
        resolved_args = self.arguments_resolver(self.agent)
        if resolved_args:  # pragma: no branch
            extra_args += ",\n".join(resolved_args)
        # Build the agent definition
        agent_str = f"""{agent_name} = {ag2_class}(
{self.get_name_arg()}
{self.get_description_arg()},{self.extras.get_system_message_arg()}
{self.get_human_input_mode_arg()},
{self.get_max_consecutive_auto_reply_arg()},
{self.get_auto_reply_arg()},
{self.extras.get_code_execution_arg()}
{self.extras.get_termination_arg()}
{extra_args}
)"""

        return AgentProcessingResult(content=agent_str)

system_message ¶

Termination message configuration for Waldiez agents.

SystemMessageProcessor ¶

SystemMessageProcessor(agent: WaldiezAgent)

Processor for system message configuration.

Source code in waldiez/exporting/agent/system_message.py

def __init__(
    self,
    agent: WaldiezAgent,
):
    self.agent = agent

process ¶

process() -> SystemMessageConfig

Process system message configuration.

Returns:

Type	Description
`SystemMessageConfig`	The processed system message configuration.

Source code in waldiez/exporting/agent/system_message.py

def process(self) -> SystemMessageConfig:
    """Process system message configuration.

    Returns
    -------
    SystemMessageConfig
        The processed system message configuration.
    """
    if not self.agent.data.system_message:
        return SystemMessageConfig()

    return SystemMessageConfig(
        before_agent_conent="",
        system_message_arg=json.dumps(self.agent.data.system_message),
    )

termination ¶

Termination message configuration for Waldiez agents.

TerminationProcessor ¶

TerminationProcessor(agent: WaldiezAgent, agent_name: str)

Processor for termination message configuration.

Source code in waldiez/exporting/agent/termination.py

def __init__(
    self,
    agent: WaldiezAgent,
    agent_name: str,
):
    self.agent = agent
    self.agent_name = agent_name

process ¶

process() -> TerminationConfig

Process termination configuration.

Returns:

Type	Description
`TerminationConfig`	The processed termination configuration.

Source code in waldiez/exporting/agent/termination.py

def process(self) -> TerminationConfig:
    """Process termination configuration.

    Returns
    -------
    TerminationConfig
        The processed termination configuration.
    """
    if self.agent.data.termination.type == "none":
        return TerminationConfig(
            termination_arg="None",
            before_content="",
        )
    if self.agent.data.termination.type == "keyword":
        return TerminationConfig(
            termination_arg=self.agent.data.termination.string,
            before_content="",
        )
    if self.agent.data.termination.type == "method":
        content, function_name = (
            self.agent.data.termination.get_termination_function(
                name_suffix=self.agent_name
            )
        )
        return TerminationConfig(
            termination_arg=function_name,
            before_content="\n\n" + content + "\n",
        )
    return TerminationConfig()

Agents

AgentExporter ¶

create_reasoning_extras ¶

extras property ¶

generate_main_content ¶

code_execution ¶

CodeExecutionProcessor ¶

process ¶

create_agent_exporter ¶

exporter ¶

AgentExporter ¶

create_reasoning_extras ¶

extras property ¶

generate_main_content ¶

fallback_args_resolver ¶

extras ¶

CaptainAgentProcessor ¶

process ¶

GroupManagerProcessor ¶

is_pattern_strategy staticmethod ¶

process ¶

GroupMemberAgentProcessor ¶

process ¶

ReasoningAgentProcessor ¶

process ¶

captain_agent_extras ¶

CaptainAgentProcessor ¶

process ¶

group_manager_agent_extas ¶

GroupManagerProcessor ¶

is_pattern_strategy staticmethod ¶

process ¶

group_member_extras ¶

GroupMemberAgentProcessor ¶

process ¶

GroupMemberProcessorResult dataclass ¶

handoffs ¶

AfterWorkProcessor ¶

process ¶

AfterWorkResult dataclass ¶

HandoffProcessor ¶

process ¶

HandoffResult dataclass ¶

TargetResult dataclass ¶

TransitionTargetProcessor ¶

process ¶

after_work ¶

AfterWorkProcessor ¶

process ¶

AfterWorkResult dataclass ¶

available ¶

TransitionAvailableProcessor ¶

process ¶

TransitionAvailableResult dataclass ¶

condition ¶

ConditionProcessor ¶

process ¶

ConditionResult dataclass ¶

handoff ¶

HandoffProcessor ¶

process ¶

HandoffResult dataclass ¶

target ¶

TargetResult dataclass ¶

TransitionTargetProcessor ¶

process ¶

rag ¶

VectorDBExtras dataclass ¶

chroma_extras ¶

get_chroma_db_args ¶

get_vector_db_extras ¶

mongo_extras ¶

get_mongodb_db_args ¶

pgvector_extras ¶

get_pgvector_db_args ¶

qdrant_extras ¶

get_qdrant_db_args ¶

vector_db_extras ¶

VectorDBExtras dataclass ¶

get_vector_db_extras ¶

extras `property` ¶

extras `property` ¶

is_pattern_strategy `staticmethod` ¶

is_pattern_strategy `staticmethod` ¶

GroupMemberProcessorResult `dataclass` ¶

AfterWorkResult `dataclass` ¶

HandoffResult `dataclass` ¶

TargetResult `dataclass` ¶

AfterWorkResult `dataclass` ¶

TransitionAvailableResult `dataclass` ¶

ConditionResult `dataclass` ¶

HandoffResult `dataclass` ¶

TargetResult `dataclass` ¶

VectorDBExtras `dataclass` ¶

VectorDBExtras `dataclass` ¶

AgentProcessingResult `dataclass` ¶