Runner

Run a waldiez flow.

The flow is first converted to an autogen flow with agents, chats, and tools. We then chown to temporary directory and: either import and call the flow's main() (if not isolated), or run the flow in a subprocess (if isolated). Before running the flow, any additional environment variables specified in the waldiez file are set.

WaldiezRunner ¶

WaldiezRunner(
    waldiez: Waldiez,
    output_path: str | Path | None = None,
    uploads_root: str | Path | None = None,
    structured_io: bool = False,
    isolated: bool = False,
    threaded: bool = False,
    skip_patch_io: bool = True,
)

Bases: WaldiezBaseRunner

Waldiez runner class.

Parameters:

Name	Type	Description	Default
`waldiez`	`Waldiez`	The waldiez flow to run.	required
`output_path`	`str \| Path \| None`	The path to the output directory where the results will be stored. If None, a temporary directory will be used.	`None`
`uploads_root`	`str \| Path \| None`	The root directory for uploads. If None, the default uploads directory will be used.	`None`
`structured_io`	`bool`	If True, the flow will use structured IO instead of the default 'input/print'.	`False`
`isolated`	`bool`	If True, the flow will be run in an isolated subprocess. Defaults to False.	`False`
`threaded`	`bool`	If True, the flow will be run in a threaded manner. Defaults to False.	`False`
`skip_patch_io`	`bool`	If True, the IO patching will be skipped. Defaults to True.	`True`

Returns:

Type	Description
`WaldiezBaseRunner`	The runner instance that will execute the flow.

Source code in waldiez/runner.py

def __init__(
    self,
    waldiez: Waldiez,
    output_path: str | Path | None = None,
    uploads_root: str | Path | None = None,
    structured_io: bool = False,
    isolated: bool = False,
    threaded: bool = False,
    skip_patch_io: bool = True,
) -> None:
    """Create a new Waldiez runner.

    Parameters
    ----------
    waldiez : Waldiez
        The waldiez flow to run.
    output_path : str | Path | None, optional
        The path to the output directory where the results will be stored.
        If None, a temporary directory will be used.
    uploads_root : str | Path | None, optional
        The root directory for uploads. If None, the default uploads
        directory will be used.
    structured_io : bool, optional
        If True, the flow will use
        structured IO instead of the default 'input/print'.
    isolated : bool, optional
        If True, the flow will be run in an isolated subprocess.
        Defaults to False.
    threaded : bool, optional
        If True, the flow will be run in a threaded manner.
        Defaults to False.
    skip_patch_io : bool, optional
        If True, the IO patching will be skipped.
        Defaults to True.

    Returns
    -------
    WaldiezBaseRunner
        The runner instance that will execute the flow.
    """
    super().__init__(
        waldiez,
        output_path=output_path,
        uploads_root=uploads_root,
        structured_io=structured_io,
        isolated=isolated,
        threaded=threaded,
        skip_patch_io=skip_patch_io,
    )
    if isolated:
        self._implementation = WaldiezSubprocessRunner(
            waldiez,
            output_path=output_path,
            uploads_root=uploads_root,
            structured_io=structured_io,
            isolated=True,
            threaded=threaded,
            skip_patch_io=skip_patch_io,
        )
    else:
        self._implementation = WaldiezImportRunner(
            waldiez,
            output_path=output_path,
            uploads_root=uploads_root,
            structured_io=structured_io,
            isolated=False,
            threaded=threaded,
            skip_patch_io=skip_patch_io,
        )

a_start `async` ¶

a_start(
    output_path: str | Path | None,
    uploads_root: str | Path | None,
    structured_io: bool | None = None,
    skip_patch_io: bool | None = None,
    skip_mmd: bool = False,
    skip_timeline: bool = False,
) -> None

Asynchronously start the flow.

Parameters:

Name	Type	Description	Default
`output_path`	`str \| Path \| None`	The output path, by default None.	required
`uploads_root`	`str \| Path \| None`	The runtime uploads root.	required
`structured_io`	`bool \| None`	Whether to use structured IO instead of the default 'input/print'.	`None`
`skip_patch_io`	`bool \| None = None`	Whether to skip patching I/O, by default None.	`None`
`skip_mmd`	`bool = False`	Whether to skip generating the mermaid diagram, by default False.	`False`
`skip_timeline`	`bool = False`	Whether to skip generating the timeline JSON, by default False.	`False`

Source code in waldiez/runner.py

async def a_start(
    self,
    output_path: str | Path | None,
    uploads_root: str | Path | None,
    structured_io: bool | None = None,
    skip_patch_io: bool | None = None,
    skip_mmd: bool = False,
    skip_timeline: bool = False,
) -> None:
    """Asynchronously start the flow.

    Parameters
    ----------
    output_path : str | Path | None
        The output path, by default None.
    uploads_root : str | Path | None
        The runtime uploads root.
    structured_io : bool | None
        Whether to use structured IO instead of the default 'input/print'.
    skip_patch_io : bool | None = None
        Whether to skip patching I/O, by default None.
    skip_mmd : bool = False
        Whether to skip generating the mermaid diagram, by default False.
    skip_timeline : bool = False
        Whether to skip generating the timeline JSON, by default False.
    """
    await self._implementation.a_start(
        output_path=output_path,
        uploads_root=uploads_root,
        structured_io=structured_io,
        skip_patch_io=skip_patch_io,
        skip_mmd=skip_mmd,
        skip_timeline=skip_timeline,
    )

load `classmethod` ¶

load(
    waldiez_file: str | Path,
    name: str | None = None,
    description: str | None = None,
    tags: list[str] | None = None,
    requirements: list[str] | None = None,
    output_path: str | Path | None = None,
    uploads_root: str | Path | None = None,
    structured_io: bool = False,
    isolated: bool = False,
    threaded: bool = False,
    skip_patch_io: bool = True,
) -> WaldiezRunner

Load a waldiez flow from a file and create a runner.

Parameters:

Name	Type	Description	Default
`waldiez_file`	`str \| Path`	The path to the waldiez file.	required
`name`	`str \| None`	The name of the flow. If None, the name from the waldiez file will be used.	`None`
`description`	`str \| None`	The description of the flow. If None, the description from the waldiez file will be used.	`None`
`tags`	`list[str] \| None`	The tags for the flow. If None, no tags will be set.	`None`
`requirements`	`list[str] \| None`	The requirements for the flow. If None, no requirements will be set.	`None`
`output_path`	`str \| Path \| None`	The path to the output directory where the results will be stored. If None, a temporary directory will be used.	`None`
`uploads_root`	`str \| Path \| None`	The root directory for uploads. If None, the default uploads directory will be used.	`None`
`structured_io`	`bool`	If True, the flow will use structured IO instead of the default 'input/print'.	`False`
`isolated`	`bool`	If True, the flow will be run in an isolated subprocess. Defaults to False.	`False`
`threaded`	`bool`	If True, the flow will be run in a threaded manner. Defaults to False.	`False`
`skip_patch_io`	`bool`	If True, the IO patching will be skipped. Defaults to True.	`True`

Returns:

Type	Description
`WaldiezBaseRunner`	The runner instance that will execute the flow.

Source code in waldiez/runner.py

@classmethod
def load(
    cls,
    waldiez_file: str | Path,
    name: str | None = None,
    description: str | None = None,
    tags: list[str] | None = None,
    requirements: list[str] | None = None,
    output_path: str | Path | None = None,
    uploads_root: str | Path | None = None,
    structured_io: bool = False,
    isolated: bool = False,
    threaded: bool = False,
    skip_patch_io: bool = True,
) -> "WaldiezRunner":
    """Load a waldiez flow from a file and create a runner.

    Parameters
    ----------
    waldiez_file : str | Path
        The path to the waldiez file.
    name : str | None, optional
        The name of the flow.
        If None, the name from the waldiez file will be used.
    description : str | None, optional
        The description of the flow.
        If None, the description from the waldiez file will be used.
    tags : list[str] | None, optional
        The tags for the flow. If None, no tags will be set.
    requirements : list[str] | None, optional
        The requirements for the flow. If None, no requirements will be set.
    output_path : str | Path | None, optional
        The path to the output directory where the results will be stored.
        If None, a temporary directory will be used.
    uploads_root : str | Path | None, optional
        The root directory for uploads. If None, the default uploads
        directory will be used.
    structured_io : bool, optional
        If True, the flow will use
        structured IO instead of the default 'input/print'.
    isolated : bool, optional
        If True, the flow will be run in an isolated subprocess.
        Defaults to False.
    threaded : bool, optional
        If True, the flow will be run in a threaded manner.
        Defaults to False.
    skip_patch_io : bool, optional
        If True, the IO patching will be skipped.
        Defaults to True.

    Returns
    -------
    WaldiezBaseRunner
        The runner instance that will execute the flow.
    """
    waldiez = Waldiez.load(
        waldiez_file,
        name=name,
        description=description,
        tags=tags,
        requirements=requirements,
    )
    return cls(
        waldiez,
        output_path=output_path,
        uploads_root=uploads_root,
        structured_io=structured_io,
        isolated=isolated,
        threaded=threaded,
        skip_patch_io=skip_patch_io,
    )

start ¶

start(
    output_path: str | Path | None,
    uploads_root: str | Path | None,
    structured_io: bool | None = None,
    skip_patch_io: bool | None = None,
    skip_mmd: bool = False,
    skip_timeline: bool = False,
) -> None

Start the flow.

Parameters:

Name	Type	Description	Default
`output_path`	`str \| Path \| None`	The output path, by default None.	required
`uploads_root`	`str \| Path \| None`	The runtime uploads root.	required
`structured_io`	`bool \| None`	Whether to use structured IO instead of the default 'input/print'. If None, it will use the value from the context.	`None`
`skip_patch_io`	`bool \| None = None`	Whether to skip patching I/O, by default None. If None, it will use the value from the context.	`None`
`skip_mmd`	`bool = False`	Whether to skip generating the mermaid diagram.	`False`
`skip_timeline`	`bool = False`	Whether to skip generating the timeline JSON.	`False`

Source code in waldiez/runner.py

def start(
    self,
    output_path: str | Path | None,
    uploads_root: str | Path | None,
    structured_io: bool | None = None,
    skip_patch_io: bool | None = None,
    skip_mmd: bool = False,
    skip_timeline: bool = False,
) -> None:
    """Start the flow.

    Parameters
    ----------
    output_path : str | Path | None
        The output path, by default None.
    uploads_root : str | Path | None
        The runtime uploads root.
    structured_io : bool | None
        Whether to use structured IO instead of the default 'input/print'.
        If None, it will use the value from the context.
    skip_patch_io : bool | None = None
        Whether to skip patching I/O, by default None.
        If None, it will use the value from the context.
    skip_mmd : bool = False
        Whether to skip generating the mermaid diagram.
    skip_timeline : bool = False
        Whether to skip generating the timeline JSON.
    """
    self._implementation.start(
        output_path=output_path,
        uploads_root=uploads_root,
        structured_io=structured_io,
        skip_patch_io=skip_patch_io,
        skip_mmd=skip_mmd,
        skip_timeline=skip_timeline,
    )

Runner

WaldiezRunner ¶

a_start async ¶

load classmethod ¶

start ¶

a_start `async` ¶

load `classmethod` ¶