Skip to content

Common

Common utils for all models.

WaldiezBase

Bases: BaseModel

Base model class to inherit from.

It contains the default configuration for all models. It also model_dumps by alias by default.

model_dump(**kwargs: Any) -> Dict[str, Any]

Dump the model to a dictionary.

Parameters:

NameTypeDescriptionDefault
**kwargsAny

Additional keyword arguments.

{}

Returns:

TypeDescription
Dict[str, Any]

The dictionary representation of the model.

Source code in waldiez/models/common/base.py
def model_dump(self, **kwargs: Any) -> Dict[str, Any]:
    """Dump the model to a dictionary.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    Dict[str, Any]
        The dictionary representation of the model.
    """
    by_alias = kwargs.pop("by_alias", None)
    if by_alias is None:
        by_alias = True
    if not isinstance(by_alias, bool):
        by_alias = True
    return super().model_dump(by_alias=by_alias, **kwargs)

model_dump_json(**kwargs: Any) -> str

Dump the model to a JSON string.

Parameters:

NameTypeDescriptionDefault
**kwargsAny

Additional keyword arguments.

{}

Returns:

TypeDescription
str

The JSON string.

Source code in waldiez/models/common/base.py
def model_dump_json(self, **kwargs: Any) -> str:
    """Dump the model to a JSON string.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    str
        The JSON string.
    """
    by_alias = kwargs.pop("by_alias", None)
    if by_alias is None:
        by_alias = True
    if not isinstance(by_alias, bool):
        by_alias = True
    return super().model_dump_json(by_alias=by_alias, **kwargs)

ag2_version

Get the autogen version.

get_autogen_version() -> str cached

Get the autogen version.

Returns:

TypeDescription
str

The autogen version.

Raises:

TypeDescription
ValueError

If pyautogen is not installed.

Source code in waldiez/models/common/ag2_version.py
@cache
def get_autogen_version() -> str:
    """Get the autogen version.

    Returns
    -------
    str
        The autogen version.

    Raises
    ------
    ValueError
        If pyautogen is not installed.
    """
    # pylint: disable=import-outside-toplevel
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        try:
            from autogen.version import __version__ as ag2  # type: ignore
        except ImportError as error:  # pragma: no cover
            raise ValueError("pyautogen is not installed.") from error
    return ag2

base

Base class to inherit from.

WaldiezBase

Bases: BaseModel

Base model class to inherit from.

It contains the default configuration for all models. It also model_dumps by alias by default.

model_dump(**kwargs: Any) -> Dict[str, Any]

Dump the model to a dictionary.

Parameters:

NameTypeDescriptionDefault
**kwargsAny

Additional keyword arguments.

{}

Returns:

TypeDescription
Dict[str, Any]

The dictionary representation of the model.

Source code in waldiez/models/common/base.py
def model_dump(self, **kwargs: Any) -> Dict[str, Any]:
    """Dump the model to a dictionary.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    Dict[str, Any]
        The dictionary representation of the model.
    """
    by_alias = kwargs.pop("by_alias", None)
    if by_alias is None:
        by_alias = True
    if not isinstance(by_alias, bool):
        by_alias = True
    return super().model_dump(by_alias=by_alias, **kwargs)

model_dump_json(**kwargs: Any) -> str

Dump the model to a JSON string.

Parameters:

NameTypeDescriptionDefault
**kwargsAny

Additional keyword arguments.

{}

Returns:

TypeDescription
str

The JSON string.

Source code in waldiez/models/common/base.py
def model_dump_json(self, **kwargs: Any) -> str:
    """Dump the model to a JSON string.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    str
        The JSON string.
    """
    by_alias = kwargs.pop("by_alias", None)
    if by_alias is None:
        by_alias = True
    if not isinstance(by_alias, bool):
        by_alias = True
    return super().model_dump_json(by_alias=by_alias, **kwargs)

check_function(code_string: str, function_name: str, function_args: List[str]) -> Tuple[bool, str]

Check the function.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string to check.

required
function_namestr

The expected method name.

required
function_argsList[str]

The expected method arguments.

required

Returns:

TypeDescription
Tuple[bool, str]

If valid, True and the function body (only), no extra lines. If invalid, False and the error message.

Source code in waldiez/models/common/method_utils.py
def check_function(
    code_string: str,
    function_name: str,
    function_args: List[str],
) -> Tuple[bool, str]:
    """Check the function.

    Parameters
    ----------
    code_string : str
        The code string to check.
    function_name : str
        The expected method name.
    function_args : List[str]
        The expected method arguments.
    Returns
    -------
    Tuple[bool, str]
        If valid, True and the function body (only), no extra lines.
        If invalid, False and the error message.
    """
    error, tree = parse_code_string(code_string)
    if error is not None or tree is None:
        return False, error or "Invalid code"
    return _validate_function_body(
        tree,
        code_string,
        function_name,
        function_args,
    )

date_utils

Date utilities.

now() -> str

Get the current date and time in UTC.

Returns:

TypeDescription
str

The current date and time in UTC.

Source code in waldiez/models/common/date_utils.py
def now() -> str:
    """Get the current date and time in UTC.

    Returns
    -------
    str
        The current date and time in UTC.
    """
    return (
        datetime.now(tz=timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )

dict_utils

Dictionary related utilities.

update_dict(original: Dict[str, Any]) -> Dict[str, Any]

Try to determine the type of the dictionary values.

Parameters:

NameTypeDescriptionDefault
originalDict[str, Any]

The original dictionary.

required

Returns:

TypeDescription
Dict[str, Any]

The updated dictionary with values converted to the detected types.

Source code in waldiez/models/common/dict_utils.py
def update_dict(original: Dict[str, Any]) -> Dict[str, Any]:
    """

    Try to determine the type of the dictionary values.

    Parameters
    ----------
    original : Dict[str, Any]
        The original dictionary.

    Returns
    -------
    Dict[str, Any]
        The updated dictionary with values converted to the detected types.
    """
    new_dict: Dict[str, Any] = {}
    for key, value in original.items():
        value_lower = str(value).lower()
        if value_lower in ("none", "null"):
            new_dict[key] = None
        elif value_lower in ("true", "false"):
            new_dict[key] = value_lower == "true"
        elif str(value).isdigit():
            new_dict[key] = int(value)
        elif str(value).replace(".", "").isdigit():
            try:
                new_dict[key] = float(value)
            except ValueError:  # pragma: no cover
                new_dict[key] = value
        else:
            new_dict[key] = value
    return new_dict

gather_code_imports(code_string: str, is_interop: bool) -> Tuple[List[str], List[str]]

Gather the imports from the code string.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required
is_interopbool

If True, make sure the interoperability import is present.

required

Returns:

TypeDescription
Tuple[List[str], List[str]]

The standard library imports and the third party imports.

Source code in waldiez/models/common/method_utils.py
def gather_code_imports(
    code_string: str,
    is_interop: bool,
) -> Tuple[List[str], List[str]]:
    """Gather the imports from the code string.

    Parameters
    ----------
    code_string : str
        The code string.
    is_interop : bool
        If True, make sure the interoperability import is present.

    Returns
    -------
    Tuple[List[str], List[str]]
        The standard library imports and the third party imports.
    """
    standard_lib_imports: List[str] = []
    third_party_imports: List[str] = []
    tree = parso.parse(code_string)  # type: ignore
    for node in tree.iter_imports():
        if node.type == "import_name":
            full_import_statement = node.get_code().strip()
            module_name = (
                node.get_code().replace("import", "").strip().split(" ")[0]
            )
            if not module_name:
                continue
            if is_standard_library(module_name):
                standard_lib_imports.append(full_import_statement)
            else:
                third_party_imports.append(full_import_statement)
        elif node.type == "import_from":
            full_import_statement = node.get_code().strip()
            module_name = (
                node.get_code().replace("from", "").strip().split(" ")[0]
            )
            if not module_name:
                continue
            if is_standard_library(module_name):
                standard_lib_imports.append(full_import_statement)
            else:
                third_party_imports.append(full_import_statement)
    if is_interop and (
        "from autogen.interop import Interoperability"
        not in third_party_imports
    ):
        third_party_imports.append(
            "from autogen.interop import Interoperability"
        )
    # sorted_standard_lib_imports =  # first import x, then from a import b
    sorted_standard_lib_imports = sorted(
        [stmt for stmt in standard_lib_imports if stmt.startswith("import ")]
    ) + sorted(
        [stmt for stmt in standard_lib_imports if stmt.startswith("from ")]
    )
    sorted_third_party_imports = sorted(
        [stmt for stmt in third_party_imports if stmt.startswith("import ")]
    ) + sorted(
        [stmt for stmt in third_party_imports if stmt.startswith("from ")]
    )
    return sorted_standard_lib_imports, sorted_third_party_imports

generate_function(function_name: str, function_args: List[str], function_types: Tuple[List[str], str], function_body: str, types_as_comments: bool = False) -> str

Generate a function.

Parameters:

NameTypeDescriptionDefault
function_namestr

The function name.

required
function_argsList[str]

The function arguments.

required
function_typesTuple[List[str], str]

The function types.

required
function_bodystr

The function body.

required
types_as_commentsbool

Include the type hints as comments (or in the function signature) (default is False).

False

Returns:

TypeDescription
str

The generated function.

Source code in waldiez/models/common/method_utils.py
def generate_function(
    function_name: str,
    function_args: List[str],
    function_types: Tuple[List[str], str],
    function_body: str,
    types_as_comments: bool = False,
) -> str:
    """Generate a function.

    Parameters
    ----------
    function_name : str
        The function name.
    function_args : List[str]
        The function arguments.
    function_types : Tuple[List[str], str]
        The function types.
    function_body : str
        The function body.
    types_as_comments : bool, optional
        Include the type hints as comments (or in the function signature)
        (default is False).
    Returns
    -------
    str
        The generated function.
    """
    if len(function_name) > MAX_VAR_NAME_LENGTH:
        function_name = function_name[:MAX_VAR_NAME_LENGTH]
    function_string = f"def {function_name}("
    if not function_args:
        function_string += ")"
    else:
        function_string += "\n"
        for arg, arg_type in zip(function_args, function_types[0]):
            if types_as_comments:
                function_string += f"    {arg},  # type: {arg_type}" + "\n"
            else:
                function_string += f"    {arg}: {arg_type}," + "\n"
        function_string += ")"
    if types_as_comments:
        function_string += ":\n"
        function_string += "    # type: (...) -> " + function_types[1]
    else:
        function_string += " -> " + function_types[1] + ":"
    function_string += "\n" if not function_body.startswith("\n") else ""
    function_string += f"{function_body}"
    if not function_string.endswith("\n"):
        function_string += "\n"
    return function_string

get_autogen_version() -> str cached

Get the autogen version.

Returns:

TypeDescription
str

The autogen version.

Raises:

TypeDescription
ValueError

If pyautogen is not installed.

Source code in waldiez/models/common/ag2_version.py
@cache
def get_autogen_version() -> str:
    """Get the autogen version.

    Returns
    -------
    str
        The autogen version.

    Raises
    ------
    ValueError
        If pyautogen is not installed.
    """
    # pylint: disable=import-outside-toplevel
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        try:
            from autogen.version import __version__ as ag2  # type: ignore
        except ImportError as error:  # pragma: no cover
            raise ValueError("pyautogen is not installed.") from error
    return ag2

get_function(code_string: str, function_name: str) -> str

Get the function signature and body.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required
function_namestr

The function name.

required

Returns:

TypeDescription
str

The function signature and body.

Source code in waldiez/models/common/method_utils.py
def get_function(
    code_string: str,
    function_name: str,
) -> str:
    """Get the function signature and body.

    Parameters
    ----------
    code_string : str
        The code string.
    function_name : str
        The function name.

    Returns
    -------
    str
        The function signature and body.
    """
    tree = parso.parse(code_string)  # type: ignore
    for node in tree.iter_funcdefs():
        if node.name.value == function_name:
            return node.get_code()
    return ""

method_utils

Function related utilities.

check_function(code_string: str, function_name: str, function_args: List[str]) -> Tuple[bool, str]

Check the function.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string to check.

required
function_namestr

The expected method name.

required
function_argsList[str]

The expected method arguments.

required

Returns:

TypeDescription
Tuple[bool, str]

If valid, True and the function body (only), no extra lines. If invalid, False and the error message.

Source code in waldiez/models/common/method_utils.py
def check_function(
    code_string: str,
    function_name: str,
    function_args: List[str],
) -> Tuple[bool, str]:
    """Check the function.

    Parameters
    ----------
    code_string : str
        The code string to check.
    function_name : str
        The expected method name.
    function_args : List[str]
        The expected method arguments.
    Returns
    -------
    Tuple[bool, str]
        If valid, True and the function body (only), no extra lines.
        If invalid, False and the error message.
    """
    error, tree = parse_code_string(code_string)
    if error is not None or tree is None:
        return False, error or "Invalid code"
    return _validate_function_body(
        tree,
        code_string,
        function_name,
        function_args,
    )

gather_code_imports(code_string: str, is_interop: bool) -> Tuple[List[str], List[str]]

Gather the imports from the code string.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required
is_interopbool

If True, make sure the interoperability import is present.

required

Returns:

TypeDescription
Tuple[List[str], List[str]]

The standard library imports and the third party imports.

Source code in waldiez/models/common/method_utils.py
def gather_code_imports(
    code_string: str,
    is_interop: bool,
) -> Tuple[List[str], List[str]]:
    """Gather the imports from the code string.

    Parameters
    ----------
    code_string : str
        The code string.
    is_interop : bool
        If True, make sure the interoperability import is present.

    Returns
    -------
    Tuple[List[str], List[str]]
        The standard library imports and the third party imports.
    """
    standard_lib_imports: List[str] = []
    third_party_imports: List[str] = []
    tree = parso.parse(code_string)  # type: ignore
    for node in tree.iter_imports():
        if node.type == "import_name":
            full_import_statement = node.get_code().strip()
            module_name = (
                node.get_code().replace("import", "").strip().split(" ")[0]
            )
            if not module_name:
                continue
            if is_standard_library(module_name):
                standard_lib_imports.append(full_import_statement)
            else:
                third_party_imports.append(full_import_statement)
        elif node.type == "import_from":
            full_import_statement = node.get_code().strip()
            module_name = (
                node.get_code().replace("from", "").strip().split(" ")[0]
            )
            if not module_name:
                continue
            if is_standard_library(module_name):
                standard_lib_imports.append(full_import_statement)
            else:
                third_party_imports.append(full_import_statement)
    if is_interop and (
        "from autogen.interop import Interoperability"
        not in third_party_imports
    ):
        third_party_imports.append(
            "from autogen.interop import Interoperability"
        )
    # sorted_standard_lib_imports =  # first import x, then from a import b
    sorted_standard_lib_imports = sorted(
        [stmt for stmt in standard_lib_imports if stmt.startswith("import ")]
    ) + sorted(
        [stmt for stmt in standard_lib_imports if stmt.startswith("from ")]
    )
    sorted_third_party_imports = sorted(
        [stmt for stmt in third_party_imports if stmt.startswith("import ")]
    ) + sorted(
        [stmt for stmt in third_party_imports if stmt.startswith("from ")]
    )
    return sorted_standard_lib_imports, sorted_third_party_imports

generate_function(function_name: str, function_args: List[str], function_types: Tuple[List[str], str], function_body: str, types_as_comments: bool = False) -> str

Generate a function.

Parameters:

NameTypeDescriptionDefault
function_namestr

The function name.

required
function_argsList[str]

The function arguments.

required
function_typesTuple[List[str], str]

The function types.

required
function_bodystr

The function body.

required
types_as_commentsbool

Include the type hints as comments (or in the function signature) (default is False).

False

Returns:

TypeDescription
str

The generated function.

Source code in waldiez/models/common/method_utils.py
def generate_function(
    function_name: str,
    function_args: List[str],
    function_types: Tuple[List[str], str],
    function_body: str,
    types_as_comments: bool = False,
) -> str:
    """Generate a function.

    Parameters
    ----------
    function_name : str
        The function name.
    function_args : List[str]
        The function arguments.
    function_types : Tuple[List[str], str]
        The function types.
    function_body : str
        The function body.
    types_as_comments : bool, optional
        Include the type hints as comments (or in the function signature)
        (default is False).
    Returns
    -------
    str
        The generated function.
    """
    if len(function_name) > MAX_VAR_NAME_LENGTH:
        function_name = function_name[:MAX_VAR_NAME_LENGTH]
    function_string = f"def {function_name}("
    if not function_args:
        function_string += ")"
    else:
        function_string += "\n"
        for arg, arg_type in zip(function_args, function_types[0]):
            if types_as_comments:
                function_string += f"    {arg},  # type: {arg_type}" + "\n"
            else:
                function_string += f"    {arg}: {arg_type}," + "\n"
        function_string += ")"
    if types_as_comments:
        function_string += ":\n"
        function_string += "    # type: (...) -> " + function_types[1]
    else:
        function_string += " -> " + function_types[1] + ":"
    function_string += "\n" if not function_body.startswith("\n") else ""
    function_string += f"{function_body}"
    if not function_string.endswith("\n"):
        function_string += "\n"
    return function_string

get_function(code_string: str, function_name: str) -> str

Get the function signature and body.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required
function_namestr

The function name.

required

Returns:

TypeDescription
str

The function signature and body.

Source code in waldiez/models/common/method_utils.py
def get_function(
    code_string: str,
    function_name: str,
) -> str:
    """Get the function signature and body.

    Parameters
    ----------
    code_string : str
        The code string.
    function_name : str
        The function name.

    Returns
    -------
    str
        The function signature and body.
    """
    tree = parso.parse(code_string)  # type: ignore
    for node in tree.iter_funcdefs():
        if node.name.value == function_name:
            return node.get_code()
    return ""

is_standard_library(module_name: str) -> bool

Check if the module is part of the standard library.

Parameters:

NameTypeDescriptionDefault
module_namestr

The module name.

required

Returns:

TypeDescription
bool

True if the module is part of the standard library.

Source code in waldiez/models/common/method_utils.py
def is_standard_library(module_name: str) -> bool:
    """Check if the module is part of the standard library.

    Parameters
    ----------
    module_name : str
        The module name.

    Returns
    -------
    bool
        True if the module is part of the standard library.
    """
    if module_name in sys.builtin_module_names:
        return True
    try:
        spec = importlib.util.find_spec(module_name)
    except BaseException:  # pylint: disable=broad-except
        return False
    if spec is None or not spec.origin:
        return False
    if "site-packages" in spec.origin:
        return False
    if spec.origin.startswith(sys.prefix) or spec.origin == "frozen":
        return True
    stdlib_path = str(Path(sysconfig.get_path("stdlib")).resolve())
    return spec.origin.startswith(stdlib_path)

parse_code_string(code_string: str) -> Tuple[Optional[str], Optional[ast.Module]]

Parse the code string.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required

Returns:

TypeDescription
Tuple[Optional[str], Optional[Module]]

If valid, None and the ast module. If invalid, the error message and None.

Source code in waldiez/models/common/method_utils.py
def parse_code_string(
    code_string: str,
) -> Tuple[Optional[str], Optional[ast.Module]]:
    """Parse the code string.

    Parameters
    ----------
    code_string : str
        The code string.

    Returns
    -------
    Tuple[Optional[str], Optional[ast.Module]]
        If valid, None and the ast module.
        If invalid, the error message and None.
    """
    # pylint: disable=broad-except
    try:
        tree = ast.parse(code_string)
    except SyntaxError as e:
        return f"SyntaxError: {e}, in " + "\n" + f"{code_string}", None
    except BaseException as e:  # pragma: no cover
        return f"Invalid code: {e}, in " + "\n" + f"{code_string}", None
    return None, tree

now() -> str

Get the current date and time in UTC.

Returns:

TypeDescription
str

The current date and time in UTC.

Source code in waldiez/models/common/date_utils.py
def now() -> str:
    """Get the current date and time in UTC.

    Returns
    -------
    str
        The current date and time in UTC.
    """
    return (
        datetime.now(tz=timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )

parse_code_string(code_string: str) -> Tuple[Optional[str], Optional[ast.Module]]

Parse the code string.

Parameters:

NameTypeDescriptionDefault
code_stringstr

The code string.

required

Returns:

TypeDescription
Tuple[Optional[str], Optional[Module]]

If valid, None and the ast module. If invalid, the error message and None.

Source code in waldiez/models/common/method_utils.py
def parse_code_string(
    code_string: str,
) -> Tuple[Optional[str], Optional[ast.Module]]:
    """Parse the code string.

    Parameters
    ----------
    code_string : str
        The code string.

    Returns
    -------
    Tuple[Optional[str], Optional[ast.Module]]
        If valid, None and the ast module.
        If invalid, the error message and None.
    """
    # pylint: disable=broad-except
    try:
        tree = ast.parse(code_string)
    except SyntaxError as e:
        return f"SyntaxError: {e}, in " + "\n" + f"{code_string}", None
    except BaseException as e:  # pragma: no cover
        return f"Invalid code: {e}, in " + "\n" + f"{code_string}", None
    return None, tree

update_dict(original: Dict[str, Any]) -> Dict[str, Any]

Try to determine the type of the dictionary values.

Parameters:

NameTypeDescriptionDefault
originalDict[str, Any]

The original dictionary.

required

Returns:

TypeDescription
Dict[str, Any]

The updated dictionary with values converted to the detected types.

Source code in waldiez/models/common/dict_utils.py
def update_dict(original: Dict[str, Any]) -> Dict[str, Any]:
    """

    Try to determine the type of the dictionary values.

    Parameters
    ----------
    original : Dict[str, Any]
        The original dictionary.

    Returns
    -------
    Dict[str, Any]
        The updated dictionary with values converted to the detected types.
    """
    new_dict: Dict[str, Any] = {}
    for key, value in original.items():
        value_lower = str(value).lower()
        if value_lower in ("none", "null"):
            new_dict[key] = None
        elif value_lower in ("true", "false"):
            new_dict[key] = value_lower == "true"
        elif str(value).isdigit():
            new_dict[key] = int(value)
        elif str(value).replace(".", "").isdigit():
            try:
                new_dict[key] = float(value)
            except ValueError:  # pragma: no cover
                new_dict[key] = value
        else:
            new_dict[key] = value
    return new_dict