Action utils

conatus.actions.utils.docstring

Docstring utilities.

More information can be found in the documentation about Action internals.

Backlinks

Some of the code for these utilities comes from Pydantic AI.

DocstringStyle module-attribute

DocstringStyle = Literal['google', 'numpy', 'sphinx']

The supported docstring styles.

generate_docstring

generate_docstring(
    doc: str,
    sig: Signature,
    style: DocstringStyle | None = None,
) -> list[DocstringSection]

Generate a structured representation of the docstring from Griffe.

PARAMETER	DESCRIPTION
doc	The docstring to parse. TYPE: str
sig	The function signature. TYPE: Signature
style	The docstring style. TYPE: DocstringStyle | None DEFAULT: None

RETURNS	DESCRIPTION
list[DocstringSection]	A structured representation of the docstring as a list of Griffe DocstringSection objects.

Source code in conatus/actions/utils/docstring.py

def generate_docstring(
    doc: str,
    sig: Signature,
    style: DocstringStyle | None = None,
) -> list[DocstringSection]:
    """Generate a structured representation of the docstring from Griffe.

    Args:
        doc: The docstring to parse.
        sig: The function signature.
        style: The docstring style.

    Returns:
        A structured representation of the docstring as a list of Griffe
            `DocstringSection` objects.
    """
    # NOTE: There are a few workarounds used in this function to deal with
    # the fact that we're tapping into Griffe's internals.

    # from: https://github.com/mkdocstrings/griffe/issues/293
    @contextmanager
    def disable_logging():  # type: ignore[no-untyped-def] # noqa: ANN202
        """Temporarily disable logging."""
        old_level = logging.root.getEffectiveLevel()
        logging.root.setLevel(logging.CRITICAL + 1)
        yield
        logging.root.setLevel(old_level)

    # see https://github.com/mkdocstrings/griffe/issues/293
    parent = cast("GriffeObject", sig)  # pyright: ignore[reportInvalidCast]

    docstring = Docstring(
        doc,
        lineno=1,
        parser=style or infer_docstring_style(doc),
        parent=parent,
    )
    with disable_logging():
        return docstring.parse()

info_from_docstring

info_from_docstring(
    func: Callable[..., ParamType],
    sig: Signature,
    *,
    style: DocstringStyle | None = None
) -> FunctionInfoFromDocstring

Extract function and parameter descriptions from a docstring.

PARAMETER	DESCRIPTION
func	The function to extract the info from. TYPE: Callable[..., ParamType]
sig	The function signature. TYPE: Signature
style	The docstring style. TYPE: DocstringStyle | None DEFAULT: None

RETURNS	DESCRIPTION
FunctionInfoFromDocstring	A structured representation of the function info extracted from the
FunctionInfoFromDocstring	docstring.

RAISES	DESCRIPTION
DocstringParsingError	If the docstring could not be parsed.

Source code in conatus/actions/utils/docstring.py

def info_from_docstring(
    func: Callable[..., ParamType],
    sig: Signature,
    *,
    style: DocstringStyle | None = None,
) -> FunctionInfoFromDocstring:
    """Extract function and parameter descriptions from a docstring.

    Args:
        func: The function to extract the info from.
        sig: The function signature.
        style: The docstring style.

    Returns:
        A structured representation of the function info extracted from the
        docstring.

    Raises:
        DocstringParsingError: If the docstring could not be parsed.
    """
    doc = func.__doc__
    if doc is None:
        return FunctionInfoFromDocstring(
            description=None, parameters=OrderedDict(), returns=[]
        )
    sections = generate_docstring(doc, sig, style)
    # NOTE: As you can see from above, we're playing a little fast and
    # loose with Griffe's Docstring, since it expects a proper parent to
    # be passed to the Docstring initialization. This means that, internally,
    # Griffe might be confused. As of this writing, in particular, Griffe
    # silently throws an AttributeError in
    # _griffe/expressions.py(1200)get_expression() (commit 2a8078a)
    # because it expects `parent` to have a `module` attribute. This
    # cascades in such a way that what frequently gets returned as an
    # annotation is not of the type `str | Expr | None` as expected, but
    # simply the raw annotation from the signature. Therefore, we have to
    # make unevaluated_type_hint more lenient.

    parameters = next(
        (p for p in sections if p.kind == DocstringSectionKind.parameters), None
    )
    params_dict: OrderedDict[str, ParamInfoFromDocstring] = (
        _handle_parameters(parameters) if parameters else OrderedDict()
    )

    main_desc = ""
    if main := next(
        (p for p in sections if p.kind == DocstringSectionKind.text), None
    ):
        if not (isinstance(main.value, str)):  # pyright: ignore[reportAny] # pragma: no cover
            msg = f"Expected a str, got {type(main.value)}"  # pyright: ignore[reportAny]
            raise DocstringParsingError(msg)
        main_desc = main.value

    returns = next(
        (p for p in sections if p.kind == DocstringSectionKind.returns), None
    )
    return_params = _handle_returns(returns) if returns else []

    return FunctionInfoFromDocstring(
        description=main_desc, parameters=params_dict, returns=return_params
    )

infer_docstring_style

infer_docstring_style(doc: str) -> DocstringStyle | None

Simplistic docstring style inference.

The potential styles are Sphinx, Google, and Numpy. The user is responsible for handling a None value.

PARAMETER	DESCRIPTION
doc	The docstring to infer the style from. TYPE: str

RETURNS	DESCRIPTION
DocstringStyle | None	The inferred docstring style. None if no style was inferred.

Source code in conatus/actions/utils/docstring.py

def infer_docstring_style(doc: str) -> DocstringStyle | None:
    """Simplistic docstring style inference.

    The potential styles are Sphinx, Google, and Numpy. The user is
    responsible for handling a None value.

    Args:
        doc: The docstring to infer the style from.

    Returns:
        The inferred docstring style. None if no style was inferred.
    """
    for pattern, replacements, style in _docstring_style_patterns:
        matches = (
            re.search(
                pattern.format(replacement), doc, re.IGNORECASE | re.MULTILINE
            )
            for replacement in replacements
        )
        if any(matches):
            return style
    return None

conatus.actions.utils.schema_extraction

Utilities to extract schema information from a function.

info_from_signature

info_from_signature(
    sig: Signature,
    func: FunctionType | None = None,
    *,
    remove_self: bool = False
) -> FunctionInfoFromSignature

Extract information from a function signature.

This method extracts the schema of a function, including its parameters and return type.

PARAMETER	DESCRIPTION
sig	The signature of the function. TYPE: Signature
func	The function to extract the schema from. (Optional, and only here as fallback for ForwardRefs.) TYPE: FunctionType | None DEFAULT: None
remove_self	Whether to remove the first argument if it's self. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
FunctionInfoFromSignature	The extracted information, which includes the function's parameters and return type.

Source code in conatus/actions/utils/schema_extraction.py

def info_from_signature(
    sig: Signature,
    func: FunctionType | None = None,
    *,
    remove_self: bool = False,
) -> FunctionInfoFromSignature:
    """Extract information from a function signature.

    This method extracts the schema of a function, including its
    parameters and return type.

    Args:
        sig: The signature of the function.
        func: The function to extract the schema from. (Optional, and only here
            as fallback for ForwardRefs.)
        remove_self: Whether to remove the first argument if it's `self`.

    Returns:
        The extracted information, which includes the function's parameters
            and return type.
    """
    params: OrderedDict[str, ParamInfoFromSignature] = OrderedDict()
    for param in sig.parameters.values():
        if remove_self and param.name == "self":
            continue
        kind = ParamKind(param.kind)
        raw_type_hint: TypeOfType = get_annotation(param, func)
        is_json_serializable, json_serializable_subtype = (
            is_json_serializable_type(raw_type_hint)
        )
        description = get_description_from_typehint(raw_type_hint)
        default_value = get_default_value(param)
        is_required = (
            default_value == ...
            if kind
            in {
                ParamKind.POSITIONAL_ONLY,
                ParamKind.POSITIONAL_OR_KEYWORD,
                ParamKind.KEYWORD_ONLY,
            }
            else False
        )
        params[param.name] = ParamInfoFromSignature(
            raw_type_hint=raw_type_hint,
            kind=kind,
            description=description,
            default_value=default_value,
            is_required=is_required,
            is_json_serializable=is_json_serializable,
            json_serializable_subtype=json_serializable_subtype,
        )
    raw_type_hint = get_return_annotation(sig)
    description = get_description_from_typehint(raw_type_hint)
    return_value = ReturnInfoFromSignature(
        raw_type_hint=raw_type_hint,
        description=description,
    )
    return FunctionInfoFromSignature(
        parameters=params,
        returns=return_value,
    )

reconcile_description

reconcile_description(
    docstring: str | None, desc: str | None
) -> str | None

Reconcile the description extracted from the docstring and the decorator.

PARAMETER	DESCRIPTION
docstring	The description extracted from the docstring. TYPE: str | None
desc	The description passed in the decorator. TYPE: str | None

RETURNS	DESCRIPTION
str | None	The reconciled description.

Source code in conatus/actions/utils/schema_extraction.py

def reconcile_description(
    docstring: str | None, desc: str | None
) -> str | None:
    """Reconcile the description extracted from the docstring and the decorator.

    Args:
        docstring: The description extracted from the docstring.
        desc: The description passed in the decorator.

    Returns:
        The reconciled description.
    """
    if desc is not None:
        return desc
    return docstring

reconcile_params

reconcile_params(
    params_sig: OrderedDict[str, ParamInfoFromSignature],
    params_doc: OrderedDict[str, ParamInfoFromDocstring],
    *,
    override_type_hint_for_llm: bool,
    fn_name: str
) -> OrderedDict[str, ParamInfo]

Reconcile the parameters extracted from the signature and docstring.

PARAMETER	DESCRIPTION
params_sig	The parameters extracted from the function's signature. TYPE: OrderedDict[str, ParamInfoFromSignature]
params_doc	The parameters extracted from the function's docstring. TYPE: OrderedDict[str, ParamInfoFromDocstring]
override_type_hint_for_llm	Whether to override the type hint passed to the LLM. TYPE: bool
fn_name	The name of the function. TYPE: str

RETURNS	DESCRIPTION
OrderedDict[str, ParamInfo]	The reconciled parameters.

Source code in conatus/actions/utils/schema_extraction.py

def reconcile_params(
    params_sig: OrderedDict[str, ParamInfoFromSignature],
    params_doc: OrderedDict[str, ParamInfoFromDocstring],
    *,
    override_type_hint_for_llm: bool,
    fn_name: str,
) -> OrderedDict[str, ParamInfo]:
    """Reconcile the parameters extracted from the signature and docstring.

    Args:
        params_sig: The parameters extracted from the function's signature.
        params_doc: The parameters extracted from the function's docstring.
        override_type_hint_for_llm: Whether to override the type hint passed to
            the LLM.
        fn_name: The name of the function.

    Returns:
        The reconciled parameters.
    """
    logger.debug("Starting to reconcile parameters...")
    params: OrderedDict[str, ParamInfo] = OrderedDict()
    for param_name, param_info_sig in params_sig.items():
        param_info_doc: ParamInfoFromDocstring | None
        if param_name not in params_doc:
            msg = (
                f"{fn_name}: Parameter {param_name} was not "
                "found in the docstring."
            )
            logger.warning(msg)
            param_info_doc = None
        else:
            param_info_doc = params_doc[param_name]
        logger.debug("Reconciling %s from signature and docstring", param_name)
        # We take the type hint from the signature. No exception.
        type_hint = param_info_sig.raw_type_hint
        # We allow the user to override the type hint passed to the LLM.
        type_hint_for_llm = (
            process_typehint(param_info_doc.raw_type_hint, return_type="str")
            if override_type_hint_for_llm and param_info_doc
            else process_typehint(
                param_info_sig.raw_type_hint, return_type="str"
            )
        )
        # We take the description from the signature if it exists.
        # Otherwise, we take it from the docstring.
        description = (
            param_info_sig.description
            or (param_info_doc and param_info_doc.description)
            or None
        )
        default_value = param_info_sig.default_value
        is_required = param_info_sig.is_required
        params[param_name] = ParamInfo(
            type_hint=type_hint,
            type_hint_for_llm=type_hint_for_llm,
            kind=param_info_sig.kind,
            description=description,
            default_value=default_value,
            is_required=is_required,
            is_json_serializable=param_info_sig.is_json_serializable,
            json_serializable_subtype=param_info_sig.json_serializable_subtype,
        )
        logger.debug("Parameters reconciled: \n%s", params[param_name])
    return params

check_docstring_signature_tuple_compatibility

check_docstring_signature_tuple_compatibility(
    return_sig: ReturnInfoFromSignature,
    return_doc: list[ReturnInfoFromDocstring],
) -> bool

Check that a signature and a docstring have same return (tuple) values.

If the signature of a function indicates that a tuple is returned (which means multiple values), we check that the docstring also indicates that a tuple is returned, and that the number of values is the same.

In other words:

def fn() -> tuple[str, str]: ...

should correspond to a docstring like this:

Returns:
    (str): First value
    (str): Second value

If that's the case, we can blend the information contained in both the signature and the docstring. Otherwise we can't.

PARAMETER	DESCRIPTION
return_sig	The signature of the function. TYPE: ReturnInfoFromSignature
return_doc	The docstring of the function. TYPE: list[ReturnInfoFromDocstring]

RETURNS	DESCRIPTION
bool	Whether the signature and the docstring are compatible.

Source code in conatus/actions/utils/schema_extraction.py

def check_docstring_signature_tuple_compatibility(
    return_sig: ReturnInfoFromSignature,
    return_doc: list[ReturnInfoFromDocstring],
) -> bool:
    """Check that a signature and a docstring have same return (tuple) values.

    If the signature of a function indicates that a tuple is returned (which
    means multiple values), we check that the docstring also indicates that
    a tuple is returned, and that the number of values is the same.

    In other words:

    ```python
    def fn() -> tuple[str, str]: ...
    ```

    should correspond to a docstring like this:

    ```
    Returns:
        (str): First value
        (str): Second value
    ```

    If that's the case, we can blend the information contained in both the
    signature and the docstring. Otherwise we can't.

    Args:
        return_sig: The signature of the function.
        return_doc: The docstring of the function.

    Returns:
        Whether the signature and the docstring are compatible.
    """
    if get_origin(return_sig.raw_type_hint) not in {tuple, tuple}:
        return False
    if len(return_doc) <= 1:
        return False

    return len(get_args(return_sig.raw_type_hint)) == len(return_doc)

reconcile_return

reconcile_return(
    return_sig: ReturnInfoFromSignature,
    return_doc: list[ReturnInfoFromDocstring],
    *,
    override_type_hint_for_llm: bool,
    passive: bool = False
) -> ReturnInfo | list[ReturnInfo]

Reconcile the return value extracted from the signature and docstring.

PARAMETER	DESCRIPTION
return_doc	The return value extracted from the docstring. TYPE: list[ReturnInfoFromDocstring]
return_sig	The return value extracted from the signature. TYPE: ReturnInfoFromSignature
override_type_hint_for_llm	Whether to override the type hint passed to the LLM. TYPE: bool
passive	Whether the function is passive. For more information, see Action.passive. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
ReturnInfo | list[ReturnInfo]	The reconciled return value.

RAISES	DESCRIPTION
PassiveActionCannotReturnError	If the function is passive and the return value is not None or Any.

Source code in conatus/actions/utils/schema_extraction.py

def reconcile_return(
    return_sig: ReturnInfoFromSignature,
    return_doc: list[ReturnInfoFromDocstring],
    *,
    override_type_hint_for_llm: bool,
    passive: bool = False,
) -> ReturnInfo | list[ReturnInfo]:
    """Reconcile the return value extracted from the signature and docstring.

    Args:
        return_doc: The return value extracted from the docstring.
        return_sig: The return value extracted from the signature.
        override_type_hint_for_llm: Whether to override the type hint passed to
            the LLM.
        passive: Whether the function is passive. For more information, see
            [`Action.passive`][conatus.actions.action.Action.passive].

    Returns:
        The reconciled return value.

    Raises:
        PassiveActionCannotReturnError: If the function is passive and the
            return value is not `None` or `Any`.
    """
    logger.debug("Starting to reconcile return value...")
    is_tuple_in_signature: bool = get_origin(return_sig.raw_type_hint) in {
        tuple,
        tuple,
    }
    is_tuple_in_docstring = len(return_doc) > 1
    signature_docstring_compatible = (
        check_docstring_signature_tuple_compatibility(return_sig, return_doc)
    )
    implied_sig_str = str(return_sig.raw_type_hint)
    implied_doc_str = ", ".join(
        [str(return_info.raw_type_hint) for return_info in return_doc]
    )
    if passive and implied_sig_str not in {"None", "typing.Any", "Any"}:
        msg = (
            "Passive actions cannot return anything. "
            "The return value %s is incorrect"
            "Please return 'None' or 'Any' instead."
        )
        logger.error(msg, implied_sig_str)
        raise PassiveActionCannotReturnError(msg, implied_sig_str)

    if is_tuple_in_signature:
        logger.debug(
            "More than one return detected."
            " Let's see if the docstring mentions a Union..."
        )
        if signature_docstring_compatible:
            logger.debug(
                "Consistent return values between the signature"
                "and the docstring: %s vs %s",
                implied_sig_str,
                implied_doc_str,
            )
        else:
            return_doc = [
                ReturnInfoFromDocstring(
                    name=None, raw_type_hint=Any, description=None
                )
                for _ in range(len(get_args(return_sig.raw_type_hint)))
            ]
        j: int = len(return_doc)
        return_infos: list[ReturnInfo] = [
            _reconcile_one_return(
                return_sig,
                return_doc[i],
                override_type_hint_for_llm=override_type_hint_for_llm,
                just_one_return=False,
                i=i,
            )
            for i in range(j)
        ]
        return return_infos

    if is_tuple_in_docstring and not signature_docstring_compatible:
        logger.warning(
            "Inconsistent return values: %s in signature, %s in docstring",
            implied_sig_str,
            implied_doc_str,
        )
        logger.warning(
            "We are only relying on the signature "
            "and will concatenate the description in the docstring"
        )
        return_doc = _flatten_return_info_docstring(return_doc)
    return _reconcile_one_return(
        return_sig,
        return_doc[0]
        if len(return_doc) > 0
        else ReturnInfoFromDocstring(
            name=None, raw_type_hint=Any, description=None
        ),
        override_type_hint_for_llm=override_type_hint_for_llm,
        just_one_return=True,
    )

reconcile_info

reconcile_info(
    func_info_sig: FunctionInfoFromSignature,
    func_info_doc: FunctionInfoFromDocstring,
    *,
    func: Callable[..., ParamType],
    desc: str | None,
    override_type_hint_for_llm: bool,
    passive: bool = False
) -> FunctionInfo

Reconcile the information extracted from the signature and docstring.

See the precise logic at the top of this file.

PARAMETER	DESCRIPTION
func_info_sig	The information extracted from the function's signature. TYPE: FunctionInfoFromSignature
func_info_doc	The information extracted from the function's docstring. TYPE: FunctionInfoFromDocstring
func	The function to extract the schema from. TYPE: Callable[..., ParamType]
desc	The description of the function as passed in Action or action. TYPE: str | None
override_type_hint_for_llm	Whether to override the type hint passed to the LLM. TYPE: bool
passive	Whether the function is passive. For more information, see Action.passive. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
FunctionInfo	The reconciled information, which includes the function's name and
FunctionInfo	parameters.

Source code in conatus/actions/utils/schema_extraction.py

def reconcile_info(
    func_info_sig: FunctionInfoFromSignature,
    func_info_doc: FunctionInfoFromDocstring,
    *,
    func: Callable[..., ParamType],
    desc: str | None,
    override_type_hint_for_llm: bool,
    passive: bool = False,
) -> FunctionInfo:
    """Reconcile the information extracted from the signature and docstring.

    See the precise logic at the top of this file.

    Args:
        func_info_sig: The information extracted from the function's signature.
        func_info_doc: The information extracted from the function's docstring.
        func: The function to extract the schema from.
        desc: The description of the function as passed in `Action` or `action`.
        override_type_hint_for_llm: Whether to override the type
            hint passed to the LLM.
        passive: Whether the function is passive. For more information, see
            [`Action.passive`][conatus.actions.action.Action.passive].

    Returns:
        The reconciled information, which includes the function's name and
        parameters.
    """
    # Description
    description = reconcile_description(func_info_doc.description, desc)
    fn_name = func.__qualname__.replace(".", "-")
    # Params
    full_params = reconcile_params(
        func_info_sig.parameters,
        func_info_doc.parameters,
        override_type_hint_for_llm=override_type_hint_for_llm,
        fn_name=fn_name,
    )
    # Return values
    full_returns = reconcile_return(
        func_info_sig.returns,
        func_info_doc.returns,
        override_type_hint_for_llm=override_type_hint_for_llm,
        passive=passive,
    )
    return FunctionInfo(
        fn_name=fn_name,
        description=description,
        parameters=full_params,
        returns=full_returns,
    )

extract_function_info

extract_function_info(
    func: FunctionType,
    *,
    is_action_function: bool = False,
    override_type_hint_for_llm: bool = False,
    desc: str | None = None,
    passive: bool = False
) -> FunctionInfo

Extract information from a function.

This method extracts the schema of a function, including its parameters and return type.

PARAMETER	DESCRIPTION
func	The function to extract the schema from. TYPE: FunctionType
is_action_function	Whether the function is an action function. This will tell the schema extractor to remove self if it's the first argument in the signature. TYPE: bool DEFAULT: False
override_type_hint_for_llm	Whether to override the type hint passed to the LLM. TYPE: bool DEFAULT: False
desc	The description of the function. Defaults to None. TYPE: str | None DEFAULT: None
passive	Whether the function is passive. For more information, see Action.passive. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
FunctionInfo	The extracted information, which includes the function's name,
FunctionInfo	parameters, and a Pydantic model representing the function's
FunctionInfo	inputs.

Source code in conatus/actions/utils/schema_extraction.py

def extract_function_info(
    func: FunctionType,
    *,
    is_action_function: bool = False,
    override_type_hint_for_llm: bool = False,
    desc: str | None = None,
    passive: bool = False,
) -> FunctionInfo:
    """Extract information from a function.

    This method extracts the schema of a function, including its
    parameters and return type.

    Args:
        func: The function to extract the schema from.
        is_action_function: Whether the function is an action function. This
            will tell the schema extractor to remove `self` if it's the first
            argument in the signature.
        override_type_hint_for_llm: Whether to override the type hint passed to
            the LLM.
        desc: The description of the function. Defaults to None.
        passive: Whether the function is passive. For more information, see
            [`Action.passive`][conatus.actions.action.Action.passive].

    Returns:
        The extracted information, which includes the function's name,
        parameters, and a Pydantic model representing the function's
        inputs.
    """
    sig = signature(func)
    func_info_sig = info_from_signature(
        sig, func=func, remove_self=is_action_function
    )
    func_info_doc = info_from_docstring(func, sig)
    return reconcile_info(
        func_info_sig,
        func_info_doc,
        func=func,
        desc=desc,
        override_type_hint_for_llm=override_type_hint_for_llm,
        passive=passive,
    )

get_enhanced_annotation

get_enhanced_annotation(
    annotation: TypeOfType, kind: ParamKind
) -> TypeOfType

Take positional and keyword-only arguments into account.

We also swap None for NoneType for Pydantic compatibility.

PARAMETER	DESCRIPTION
annotation	The original annotation. TYPE: TypeOfType
kind	The kind of parameter. TYPE: ParamKind

RETURNS	DESCRIPTION
TypeOfType	The enhanced annotation.

Source code in conatus/actions/utils/schema_extraction.py

def get_enhanced_annotation(
    annotation: TypeOfType, kind: ParamKind
) -> TypeOfType:
    """Take positional and keyword-only arguments into account.

    We also swap `None` for `NoneType` for Pydantic compatibility.

    Args:
        annotation: The original annotation.
        kind: The kind of parameter.

    Returns:
        The enhanced annotation.
    """
    if kind == ParamKind.VAR_POSITIONAL:
        # NOTE: This might fail for type aliases... If this breaks you
        # will know why :)
        # https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
        return list[annotation]  # type: ignore[valid-type]
    if kind == ParamKind.VAR_KEYWORD:
        # NOTE: Ditto. See above.
        return dict[str, annotation]  # type: ignore[valid-type]
    annotation = annotation if annotation is not None else NoneType
    return cast("type", annotation)

build_parameter_schema

build_parameter_schema(
    parameter_name: str,
    parameter_info: ParamInfo,
    compatible_variables: dict[str, list[str]] | None,
    *,
    use_test_variables_if_none: bool = False
) -> tuple[type, FieldInfo]

Build the schema for a single parameter.

PARAMETER	DESCRIPTION
parameter_name	The name of the parameter. TYPE: str
parameter_info	The parameter information. TYPE: ParamInfo
compatible_variables	Mapping of parameter names to variable names. TYPE: dict[str, list[str]] | None
use_test_variables_if_none	Whether to use the test variables if compatible_variables is None. If set to True, this ensures that the function's JSON schema is valid even if some of its parameters are not JSON serializable. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
tuple[type, FieldInfo]	A tuple of the type annotation and field info for the parameter.

RAISES	DESCRIPTION
NoCompatibleVariablesFoundForNonJSONSerializableError	If no compatible variables are found for a non-JSON serializable argument.

Source code in conatus/actions/utils/schema_extraction.py

def build_parameter_schema(
    parameter_name: str,
    parameter_info: ParamInfo,
    compatible_variables: dict[str, list[str]] | None,
    *,
    use_test_variables_if_none: bool = False,
) -> tuple[type, FieldInfo]:
    """Build the schema for a single parameter.

    Args:
        parameter_name: The name of the parameter.
        parameter_info: The parameter information.
        compatible_variables: Mapping of parameter names to variable names.
        use_test_variables_if_none: Whether to use the test variables if
            `compatible_variables` is `None`. If set to [`True`][True], this
            ensures that the function's JSON schema is valid even if some of
            its parameters are not JSON serializable.

    Returns:
        A tuple of the type annotation and field info for the parameter.

    Raises:
        NoCompatibleVariablesFoundForNonJSONSerializableError: If no
            compatible variables are found for a non-JSON serializable
            argument.
    """
    desc: str = parameter_info.description or NO_DESC
    if parameter_info.kind == ParamKind.VAR_POSITIONAL:
        th_llm = f"list[{parameter_info.type_hint_for_llm}]"
    elif parameter_info.kind == ParamKind.VAR_KEYWORD:
        th_llm = f"dict[str, {parameter_info.type_hint_for_llm}]"
    else:
        th_llm = parameter_info.type_hint_for_llm
    desc = f"(type: {th_llm}) {desc}"
    default_value = parameter_info.default_value

    annotation_lst: list[Enum | TypeOfType] = []

    # Allow variables to be passed by value if available.
    if parameter_info.is_json_serializable:
        annotation = parameter_info.json_serializable_subtype
        json_ser_enhanced_annotation = get_enhanced_annotation(
            annotation, parameter_info.kind
        )
        annotation_lst.append(json_ser_enhanced_annotation)

    # Allow variables to be passed by reference if available.
    use_test_variables = (
        compatible_variables is None and use_test_variables_if_none
    )
    compatible_variables_present = (
        compatible_variables is not None
        and parameter_name in compatible_variables
        and len(compatible_variables[parameter_name]) > 0
    )

    if use_test_variables or compatible_variables_present:
        variables = (
            ["test_var0"]
            if compatible_variables is None
            else compatible_variables[parameter_name]
        )
        formatted_variables = [(var, f"<<var:{var}>>") for var in variables]
        enum_name = f"{parameter_name}_possible_variables"
        var_enum = Enum(enum_name, formatted_variables)  # type: ignore[misc]
        var_enum.__doc__ = (
            f" You can pass '{parameter_name}' by reference with a formatted "
            "reference '<<var:{name}>>' to a variable compatible with type "
            f"'{parameter_info.type_hint_for_llm}' among {variables}"
        )
        annotation_lst.append(var_enum)

    if len(annotation_lst) == 0:
        msg = (
            f"Parameter {parameter_name} is not JSON serializable, and no "
            "compatible variables were found. If you are calling this "
            "function directly, it means you need to flesh out your "
            "compatible variables for this parameter. If you are calling "
            "this function indirectly, please report this bug."
        )
        raise NoCompatibleVariablesFoundForNonJSONSerializableError(msg)

    return (
        cast("type", cast("object", Union[tuple(annotation_lst)])),
        cast("FieldInfo", Field(default_value, description=desc)),
    )

build_return_schema

build_return_schema(
    function_info: FunctionInfo,
    all_variables: list[str] | None,
    *,
    use_test_variables_if_none: bool = False
) -> dict[str, tuple[type, FieldInfo]]

Build the return schema for the JSON model.

PARAMETER	DESCRIPTION
function_info	The function information. TYPE: FunctionInfo
all_variables	A list of all available variable names. TYPE: list[str] | None
use_test_variables_if_none	Whether to use the test variables if all_variables is None. If set to True, this ensures that the function's JSON schema is valid even if its return value is not JSON serializable. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, tuple[type, FieldInfo]]	A dictionary with the key "return" mapping to a tuple containing the type annotation and field info for the return value.

Source code in conatus/actions/utils/schema_extraction.py

def build_return_schema(
    function_info: FunctionInfo,
    all_variables: list[str] | None,
    *,
    use_test_variables_if_none: bool = False,
) -> dict[str, tuple[type, FieldInfo]]:
    """Build the return schema for the JSON model.

    Args:
        function_info: The function information.
        all_variables: A list of all available variable names.
        use_test_variables_if_none: Whether to use the test variables if
            `all_variables` is `None`. If set to [`True`][True], this ensures
            that the function's JSON schema is valid even if its return value
            is not JSON serializable.

    Returns:
        A dictionary with the key "return" mapping to a tuple containing the
            type annotation and field info for the return value.
    """
    # TODO(lemeb): Offer the possibility to mandate the model to return
    # a specific existing variable (ideally compatible with the type hint).
    # CTUS-72

    # Skip if the function has no return value.
    if (
        not isinstance(function_info.returns, list)
        and function_info.returns.type_hint is None
    ):
        return {}

    # If no variables are provided, use a default one.
    if all_variables is None:
        if use_test_variables_if_none:
            all_variables = ["test_var0"]
        else:
            return {}

    # If there are variables, build the return schema.
    if len(all_variables) > 0:
        n_of_returns = (
            len(function_info.returns)
            if isinstance(function_info.returns, list)
            else 1
        )
        all_variables_for_enum = [(var, var) for var in all_variables]
        if n_of_returns == 1:
            var_ref_type = Enum(  # type: ignore[misc]
                "possible_return_assignment", all_variables_for_enum
            )
            description_return = (
                "If you want this action to assign the return value to a "
                "variable, pass the name of the variable in this `return` "
                "parameter. If you pass a null value, we will create a new "
                "variable automatically.\n"
            )
        else:
            var_ref_type = Tuple[(str | None,) * n_of_returns]  # type: ignore[misc, assignment]
            description_return = (
                "For each of the values, you can optionally assign it to a "
                "variable by passing its name in the list of arguments. "
                "If you pass a null value, we will create a new variable "
                "automatically. For reference, the available variables are: ["
                + ", ".join(all_variables)
                + "]\n"
            )
        description_return += "This is OPTIONAL. Only use it if it makes sense."
        return {
            "return": (
                cast("type", Optional[var_ref_type]),
                cast("FieldInfo", Field(..., description=description_return)),
            )
        }
    return {}

generate_pydantic_json_schema_model

generate_pydantic_json_schema_model(
    function_info: FunctionInfo,
    compatible_variables: (
        dict[str, list[str]] | None
    ) = None,
    all_variables: list[str] | None = None,
    *,
    args_without_variable_references: (
        list[str] | None
    ) = None,
    use_test_variables_if_none: bool = False
) -> type[BaseModel]

Generate the Pydantic model representing the function's JSON schema.

Unlike the action's input_model , which is used to validate the inputs and outputs of the action function, the json_schema is used to display the JSON schema of the action function in the LLM. The main difference here is that arguments whose types are not JSON serializable are converted to string references; the role of the LLM is to choose among the runtime variables that are compatible with the argument in question.

This function is called when the Action is being created. It is also called after each LLM step, to update the JSON schema model with the latest compatible variables. The variables need to be passed to this function; otherwise, the Enums will be filled with test values.

PARAMETER	DESCRIPTION
function_info	The schema of the function. TYPE: FunctionInfo
compatible_variables	The variables that are compatible with the function's arguments; of the format {arg_name: [var_name]}. TYPE: dict[str, list[str]] | None DEFAULT: None
all_variables	The names of all the available variables. TYPE: list[str] | None DEFAULT: None
args_without_variable_references	The names of the arguments for which we forbid variable references. TYPE: list[str] | None DEFAULT: None
use_test_variables_if_none	Whether to use the test variables if compatible_variables or all_variables are None. This ensures that the function's JSON schema is valid even if some of its parameters are not JSON serializable. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
type[BaseModel]	The Pydantic model representing the function's JSON schema.

Source code in conatus/actions/utils/schema_extraction.py

def generate_pydantic_json_schema_model(
    function_info: FunctionInfo,
    compatible_variables: dict[str, list[str]] | None = None,
    all_variables: list[str] | None = None,
    *,
    args_without_variable_references: list[str] | None = None,
    use_test_variables_if_none: bool = False,
) -> type[BaseModel]:
    """Generate the Pydantic model representing the function's JSON schema.

    Unlike the action's [`input_model`
    ][conatus.actions.action.Action.input_model], which is used to validate the
    inputs and outputs of the action function, the [`json_schema`
    ][conatus.actions.action.Action.json_schema] is used to
    display the JSON schema of the action function in the LLM. The main
    difference here is that arguments whose types are not JSON serializable are
    converted to string references; the role of the LLM is to choose among the
    runtime variables that are compatible with the argument in question.

    This function is called when the [`Action`][conatus.actions.action.Action]
    is being created. It is also called after each LLM step, to update the
    JSON schema model with the latest compatible variables. The variables
    need to be passed to this function; otherwise, the [`Enum`][enum.Enum]s
    will be filled with test values.

    Args:
        function_info: The schema of the function.
        compatible_variables: The variables that are compatible with the
            function's arguments; of the format `{arg_name: [var_name]}`.
        all_variables: The names of all the available variables.
        args_without_variable_references: The names of the arguments for
            which we forbid variable references.
        use_test_variables_if_none: Whether to use the test variables if
            `compatible_variables` or `all_variables` are `None`. This ensures
            that the function's JSON schema is valid even if some of its
            parameters are not JSON serializable.

    Returns:
        The Pydantic model representing the function's JSON schema.
    """
    to_pass_to_create_model: dict[str, tuple[type, FieldInfo]] = {}

    args_without_variable_references = args_without_variable_references or []
    for parameter_name, parameter_info in function_info.parameters.items():
        # If we pass `None`, we'll end up with the test variable values.
        if parameter_name in args_without_variable_references:
            compatible_variables = {}
        to_pass_to_create_model[parameter_name] = build_parameter_schema(
            parameter_name=parameter_name,
            parameter_info=parameter_info,
            compatible_variables=compatible_variables,
            use_test_variables_if_none=use_test_variables_if_none,
        )

    return_schema = build_return_schema(
        function_info=function_info,
        all_variables=all_variables,
        use_test_variables_if_none=use_test_variables_if_none,
    )
    to_pass_to_create_model.update(return_schema)

    return cast(
        "type[BaseModel]",
        create_model(  # type: ignore[call-overload] # pyright: ignore[reportCallIssue]
            f"{function_info.fn_name}JSONSchema",
            __config__=cfg_all_types,
            __doc__=function_info.description,
            **to_pass_to_create_model,  # pyright: ignore[reportArgumentType]
        ),
    )

generate_pydantic_models

generate_pydantic_models(
    function_info: FunctionInfo,
    *,
    use_test_variables_if_none: bool = False
) -> FunctionPydanticModels

Generate Pydantic models from a function's schema.

PARAMETER	DESCRIPTION
function_info	The schema of the function. TYPE: FunctionInfo
use_test_variables_if_none	Whether to use the test variables if compatible_variables or all_variables are None. This ensures that the function's JSON schema is valid even if some of its parameters are not JSON serializable. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
FunctionPydanticModels	The Pydantic models representing the function's inputs.

Source code in conatus/actions/utils/schema_extraction.py

def generate_pydantic_models(
    function_info: FunctionInfo,
    *,
    use_test_variables_if_none: bool = False,
) -> FunctionPydanticModels:
    """Generate Pydantic models from a function's schema.

    Args:
        function_info: The schema of the function.
        use_test_variables_if_none: Whether to use the test variables if
            `compatible_variables` or `all_variables` are `None`. This ensures
            that the function's JSON schema is valid even if some of its
            parameters are not JSON serializable.

    Returns:
        The Pydantic models representing the function's inputs.
    """
    # Start with parameters
    all_fields: dict[str, tuple[type, FieldInfo]] = {}
    for name, param in function_info.parameters.items():
        desc: str = param.description or NO_DESC
        annotation = param.type_hint
        enhanced_annotation = get_enhanced_annotation(annotation, param.kind)
        all_fields[name] = (
            cast("type", enhanced_annotation),
            cast(
                "FieldInfo",
                Field(param.default_value, description=desc),
            ),
        )

    # Continue with return values
    returns = function_info.returns
    if not isinstance(returns, list) and returns.type_hint is None:
        output_model = None
    else:
        # TODO(lemeb): Use `tuple[*list(ret.type_hint for ret in returns)]`
        # once we drop Python 3.10 support.
        # CTUS-54
        returns_type = (
            tuple[tuple(ret.type_hint for ret in returns)]  # type: ignore[misc]
            if isinstance(returns, list)
            else returns.type_hint
        )
        output_model = create_model(
            f"{function_info.fn_name}OutputModel",
            __config__=cfg_all_types,
            returns=(returns_type, ...),  # pyright: ignore[reportAny]
        )

    input_model = cast(
        "type[BaseModel]",
        create_model(  # type: ignore[call-overload] # pyright: ignore[reportCallIssue]
            f"{function_info.fn_name}InputModel",
            __config__=cfg_all_types,
            __doc__=function_info.description,
            **all_fields,  # pyright: ignore[reportArgumentType]
        ),
    )

    json_schema = generate_pydantic_json_schema_model(
        function_info=function_info,
        use_test_variables_if_none=use_test_variables_if_none,
    )

    return FunctionPydanticModels(
        input_model=input_model,
        output_model=output_model,
        json_schema=json_schema,
        all_fields=all_fields,
    )

conatus.actions.utils.function_info_types

Types used in the actions module.

ParamKind

Bases: IntEnum

Copy of the _ParameterKind enum from the inspect module.

Since this enum is private, we don't want to take any risks.

ArgForPrint

Bases: ArbitraryBaseModel

Argument for pretty printing (works for both parameters and returns).

ATTRIBUTE	DESCRIPTION
type_hint	The type hint of the argument. TYPE: str | TypeOfType
description	The description of the argument. TYPE: str | None
name	The name of the argument. TYPE: str | None
kind	The kind of the argument. TYPE: ParamKind | None
default_value	The default value of the argument. TYPE: ParamType | None
is_required	Whether the argument is required. TYPE: bool | None
is_json_serializable	Whether the argument is JSON serializable. TYPE: bool | None
json_serializable_subtype	The subtype of the argument that is JSON serializable. TYPE: TypeOfType

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the argument.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. Defaults to True. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the argument.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the argument.

    Args:
        colors: Whether to use colors in the output. Defaults to True.

    Returns:
        The pretty print of the argument.
    """
    repr_str: str = ""
    json_ok = ""
    if self.name:
        repr_str += f"- {blue('Name', colors=colors)}: {self.name}\n"

    if self.is_json_serializable is not None:
        json_ok = " " + (
            ("(" + green("JSON OK", colors=colors) + ")")
            if self.is_json_serializable
            else ("(" + grey("Not JSON OK", colors=colors) + ")")
        )
    if self.json_serializable_subtype != ...:
        json_ok += grey(
            f"\n-- JSON-serializable subtype: "
            f"{self.json_serializable_subtype}",
            colors=colors,
        )

    desc = self.description or grey("<No description>", colors=colors)
    fa = " (if in Annotated)" if self.from_signature else ""
    repr_str += f"- {blue('Description' + fa, colors=colors)}: {desc}\n"
    th = cast("type", self.type_hint)
    type_hint, type_type_hint = str(th), str(type(th))
    repr_str += f"- {blue('Type hint', colors=colors)}: "
    repr_str += f"{type_hint}{json_ok}\n"
    repr_str += (
        grey(f"-- Type of type hint: {type_type_hint}", colors=colors)
        + "\n"
        if PRINT_TYPE_OF_TYPE
        else ""
    )

    if self.type_hint_for_llm:
        repr_str += (
            f"- {blue('Type hint shown to the LLM', colors=colors)}: "
            f"'{self.type_hint_for_llm}'\n"
        )
    if self.is_required is not None:
        repr_str += f"- {blue('Required', colors=colors)}:"
        repr_str += f" {self.is_required}\n"
    if self.kind:
        repr_str += f"- {blue('Kind of parameter', colors=colors)}: "
        repr_str += f"{self.kind}\n"
    if (dv := self.default_value) is not None:
        default_value = (
            str(dv)
            if dv != Ellipsis
            else grey("<No default value>", colors=colors)
        )
        repr_str += f"- {blue('Default value', colors=colors)}: "
        repr_str += f"{default_value}\n"
    return repr_str

ParamInfoFromDocstring

Bases: ArbitraryBaseModel

Information about a raw parameter (according to the docstring).

ATTRIBUTE	DESCRIPTION
raw_type_hint	The type hint of the parameter. TYPE: str | TypeOfType
description	The description of the parameter. TYPE: str | None
default_value	The default value of the parameter. TYPE: ParamType | None

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the parameter.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. Defaults to True. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the parameter.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the parameter.

    Args:
        colors: Whether to use colors in the output. Defaults to True.

    Returns:
        The pretty print of the parameter.
    """
    return ArgForPrint(
        type_hint=self.raw_type_hint,
        description=self.description,
        default_value=self.default_value,
    ).pretty_print(colors=colors)

ReturnInfoFromDocstring

Bases: ArbitraryBaseModel

Information about a raw return parameter (according to the docstring).

ATTRIBUTE	DESCRIPTION
name	The name of the return parameter. TYPE: str | None
raw_type_hint	The type hint of the return parameter. TYPE: str | TypeOfType
description	The description of the return parameter. TYPE: str | None

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the return parameter.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. Defaults to True. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the return parameter.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the return parameter.

    Args:
        colors: Whether to use colors in the output. Defaults to True.

    Returns:
        The pretty print of the return parameter.
    """
    return ArgForPrint(
        name=self.name,
        type_hint=self.raw_type_hint,
        description=self.description,
    ).pretty_print(colors=colors)

FunctionInfoFromDocstring

Bases: ArbitraryBaseModel

Structured function info from docstring.

ATTRIBUTE	DESCRIPTION
description	The description of the function. TYPE: str | None
parameters	The parameters of the function. TYPE: OrderedDict[str, ParamInfoFromDocstring]
returns	The return parameters of the function. (Can be multiple) TYPE: list[ReturnInfoFromDocstring]

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the function info.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the function info.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the function info.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the function info.
    """
    repr_str = "Function info (inferred from docstring):\n"
    if len(self.parameters) == 0:
        repr_str += grey("  - No parameter information\n", colors=colors)
    for param_name, param_info in self.parameters.items():
        repr_str += (
            f"  - {green('Parameter ' + param_name, colors=colors)}:\n"
        )
        repr_str += add_indentation(
            param_info.pretty_print(colors=colors), 4
        )
    if len(self.returns) == 0:
        repr_str += grey("  - No return information\n", colors=colors)
    for i, return_info in enumerate(self.returns):
        repr_str += f"  - {green('Return #' + str(i), colors=colors)}:\n"
        repr_str += add_indentation(
            return_info.pretty_print(colors=colors), 4
        )
    return repr_str

ReturnInfoFromSignature

Bases: ArbitraryBaseModel

Information about a raw return parameter (according to the signature).

ATTRIBUTE	DESCRIPTION
raw_type_hint	The raw type hint of the return parameter (for Pydantic). TYPE: TypeOfType
description	The description of the return parameter. TYPE: str | None

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the function info.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the function info.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the function info.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the function info.
    """
    return ArgForPrint(
        type_hint=self.raw_type_hint,
        description=self.description,
        from_signature=True,
    ).pretty_print(colors=colors)

ParamInfoFromSignature

Bases: ArbitraryBaseModel

Information about a raw parameter (according to the signature).

ATTRIBUTE	DESCRIPTION
raw_type_hint	The raw type hint of the parameter (for Pydantic). TYPE: TypeOfType
kind	The kind of the parameter. TYPE: ParamKind
description	The description of the parameter. TYPE: str | None
default_value	The default value of the parameter. TYPE: ParamType
is_required	Whether the parameter is required. TYPE: bool
is_json_serializable	Whether the parameter is JSON serializable. TYPE: bool
json_serializable_subtype	The subtype of the parameter that is JSON serializable. TYPE: TypeOfType

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the parameter.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the parameter.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the parameter.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the parameter.
    """
    return ArgForPrint(
        type_hint=self.raw_type_hint,
        kind=self.kind,
        default_value=self.default_value,
        is_required=self.is_required,
        description=self.description,
        from_signature=True,
        is_json_serializable=self.is_json_serializable,
        json_serializable_subtype=self.json_serializable_subtype,
    ).pretty_print(colors=colors)

FunctionInfoFromSignature

Bases: ArbitraryBaseModel

Structured function info from signature.

ATTRIBUTE	DESCRIPTION
parameters	The parameters of the function. TYPE: OrderedDict[str, ParamInfoFromSignature]
returns	The return parameter of the function. TYPE: ReturnInfoFromSignature

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the function info.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the function info.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the function info.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the function info.
    """
    repr_str = "Function info (inferred from signature):\n"
    if len(self.parameters) == 0:
        repr_str += grey("  - No parameter information\n", colors=colors)
    for param_name, param_info in self.parameters.items():
        repr_str += f"  - {green('Parameter ' + param_name, colors=colors)}"
        repr_str += ":\n"
        repr_str += add_indentation(
            param_info.pretty_print(colors=colors), 4
        )
    repr_str += f"  - {green('Returns', colors=colors)}:\n"
    repr_str += add_indentation(self.returns.pretty_print(colors=colors), 4)
    return repr_str

ReturnInfo

Bases: ArbitraryBaseModel

Information about a return parameter.

ATTRIBUTE	DESCRIPTION
type_hint	The type hint of the return parameter. TYPE: TypeOfType
type_hint_for_llm	The type hint of the return parameter for the LLM. TYPE: str
description	The description of the return parameter. TYPE: str | None

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the return parameter.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the return parameter.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the return parameter.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the return parameter.
    """
    return ArgForPrint(
        type_hint=self.type_hint,
        type_hint_for_llm=self.type_hint_for_llm,
        description=self.description,
    ).pretty_print(colors=colors)

ParamInfo

Bases: ArbitraryBaseModel

Information about a parameter.

ATTRIBUTE	DESCRIPTION
type_hint	The raw type hint of the parameter. TYPE: TypeOfType
type_hint_for_llm	The type hint of the parameter for the LLM. TYPE: str
kind	The kind of the parameter. TYPE: ParamKind
description	The description of the parameter. TYPE: str | None
default_value	The default value of the parameter. TYPE: ParamType
is_required	Whether the parameter is required. TYPE: bool
is_json_serializable	Whether the parameter can be serialized to JSON. TYPE: bool
json_serializable_subtype	The subtype of the parameter that can be serialized to JSON. TYPE: TypeOfType

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the parameter.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the parameter.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the parameter.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the parameter.
    """
    return ArgForPrint(
        type_hint=self.type_hint,
        type_hint_for_llm=self.type_hint_for_llm,
        kind=self.kind,
        description=self.description,
        default_value=self.default_value,
        is_required=self.is_required,
        is_json_serializable=self.is_json_serializable,
        json_serializable_subtype=self.json_serializable_subtype,
    ).pretty_print(colors=colors)

FunctionInfo

Bases: ArbitraryBaseModel

Structured function info.

ATTRIBUTE	DESCRIPTION
fn_name	The name of the function. TYPE: str
description	The description of the function. TYPE: str | None
parameters	The parameters of the function. TYPE: OrderedDict[str, ParamInfo]
returns	The return parameter of the function. (Can be multiple) TYPE: ReturnInfo | list[ReturnInfo]

pretty_print

pretty_print(*, colors: bool = True) -> str

Returns the pretty print string of the function info.

PARAMETER	DESCRIPTION
colors	Whether to use colors in the output. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The pretty print of the function info.

Source code in conatus/actions/utils/function_info_types.py

def pretty_print(self, *, colors: bool = True) -> str:
    """Returns the pretty print string of the function info.

    Args:
        colors: Whether to use colors in the output.

    Returns:
        The pretty print of the function info.
    """
    repr_str = "Function info:\n"
    repr_str += f"- {green('Name', colors=colors)}: '{self.fn_name}'\n"
    repr_str += f"- {green('Description', colors=colors)}: "
    repr_str += f"'{self.description}'\n"
    if len(self.parameters) == 0:
        repr_str += grey("  - No parameter information\n", colors=colors)
    for param_name, param_info in self.parameters.items():
        repr_str += "  - "
        repr_str += f"{green('Parameter ' + param_name, colors=colors)}:\n"
        repr_str += add_indentation(
            param_info.pretty_print(colors=colors), 4
        )
    if isinstance(self.returns, list):
        for i, return_info in enumerate(self.returns):
            repr_str += "  - "
            repr_str += f"{green('Return ' + str(i), colors=colors)}:\n"
            repr_str += add_indentation(
                return_info.pretty_print(colors=colors), 4
            )
    else:
        repr_str += "  - "
        repr_str += f"{green('Returns', colors=colors)}:\n"
        repr_str += add_indentation(
            self.returns.pretty_print(colors=colors), 4
        )
    return repr_str

FunctionPydanticModels dataclass

FunctionPydanticModels(
    input_model: type[BaseModel],
    output_model: type[BaseModel] | None,
    json_schema: type[BaseModel],
    all_fields: dict[str, tuple[type, FieldInfo]],
)

Pydantic models for the function info.

input_model instance-attribute

input_model: type[BaseModel]

Pydantic model for the input of the function.

This model is used to perform type checking of the input parameters.

output_model instance-attribute

output_model: type[BaseModel] | None

Pydantic model for the output of the function.

This model is used to perform type checking of the output parameters.

json_schema instance-attribute

json_schema: type[BaseModel]

Pydantic model for the JSON schema of the function.

The main difference with input_model is that this model is meant to be regenerated at runtime depending on the runtime variables. It is also used to generate the OpenAI JSON schema of the action.

all_fields instance-attribute

all_fields: dict[str, tuple[type, FieldInfo]]

All fields of the function.

The keys are the names of the fields, and the values are tuples of the type of the field and the FieldInfo instance. We use this internally to create the input_model.

This is mostly useful at agent runtime, when we can build a big Pydantic model of all the parameters of all the actions, which we can then use to check the compatibility of runtime variables.

add_indentation

add_indentation(text: str, indentation: int) -> str

Add indentation to a text.

If we detect a newline at the end of the text, we add a newline at the end of the result. That way we don't get an ugly last line with dangling indentation.

RETURNS	DESCRIPTION
str	The text with indentation added.

Source code in conatus/actions/utils/function_info_types.py

def add_indentation(text: str, indentation: int) -> str:
    """Add indentation to a text.

    If we detect a newline at the end of the text, we add a newline at the end
    of the result. That way we don't get an ugly last line with dangling
    indentation.

    Returns:
        The text with indentation added.
    """
    add_new_line = False
    if len(text) == 0:
        return text
    if text[-1] == "\n":
        text = text[:-1]
        add_new_line = True
    return "\n".join(" " * indentation + line for line in text.split("\n")) + (
        "\n" if add_new_line else ""
    )

conatus.actions.utils.type_hint_handling

Utilities handling type hints.

We do so here partially for type-checking, since so many values are Any or Unknown.

IsLiteral module-attribute

IsLiteral = type[ParamType]

The type of a Literal type.

IsAnnotated module-attribute

IsAnnotated = type[ParamType]

The type of an Annotated type.

OneOfCallables module-attribute

OneOfCallables = {Callable, Callable}

The type of a one-of callable.

CallableType

Bases: ArbitraryBaseModel

Type of a callable.

ATTRIBUTE	DESCRIPTION
params	The parameters of the callable. TYPE: list[TypeOfType]
return_	The return type of the callable. TYPE: TypeOfType

AnnotatedType

Bases: BaseModel

Type of an Annotated type.

ATTRIBUTE	DESCRIPTION
metadata	The metadata of the Annotated type. TYPE: ParamType

LiteralType

Bases: BaseModel

Type of a Literal type.

ATTRIBUTE	DESCRIPTION
values	The values of the Literal type. TYPE: list[ParamType]

ClassVarType

Bases: ArbitraryBaseModel

Type of a ClassVar type.

ATTRIBUTE	DESCRIPTION
type_	The type of the ClassVar type. TYPE: TypeOfType

FinalType

Bases: ArbitraryBaseModel

Type of a Final type.

ATTRIBUTE	DESCRIPTION
type_	The type of the Final type. TYPE: TypeOfType

NotRequiredType

Bases: ArbitraryBaseModel

Type of a NotRequired type.

ATTRIBUTE	DESCRIPTION
type_	The type of the NotRequired type. TYPE: TypeOfType

ForwardRefType

Bases: ArbitraryBaseModel

Type of a ForwardRef type.

ATTRIBUTE	DESCRIPTION
type_	The type of the ForwardRef type. TYPE: TypeOfType

get_restricted_annotation

get_restricted_annotation(
    type_hint: TypeOfType,
) -> TypeOfTypeRestrictive

Get the restricted annotation of a type hint.

PARAMETER	DESCRIPTION
type_hint	The type hint to get the restricted annotation from. TYPE: TypeOfType

RETURNS	DESCRIPTION
TypeOfTypeRestrictive	The restricted annotation of the type hint.

Source code in conatus/actions/utils/type_hint_handling.py

def get_restricted_annotation(
    type_hint: TypeOfType,
) -> TypeOfTypeRestrictive:
    """Get the restricted annotation of a type hint.

    Args:
        type_hint: The type hint to get the restricted annotation from.

    Returns:
        The restricted annotation of the type hint.
    """
    new_type_hint = type_hint
    if get_origin(type_hint) in AnnotatedPythonTypes:
        new_type_hint = AnnotatedType(metadata=type_hint.__metadata__)  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownArgumentType]
    elif get_origin(type_hint) is Literal:
        new_type_hint = LiteralType(values=type_hint.__args__)  # type: ignore[union-attr, arg-type] # pyright: ignore[reportAttributeAccessIssue, reportArgumentType, reportUnknownMemberType]
    elif get_origin(type_hint) is ClassVar:
        new_type_hint = ClassVarType(type_=type_hint.__args__[0])  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue, reportUnknownArgumentType, reportUnknownMemberType]
    elif get_origin(type_hint) is Final:
        new_type_hint = FinalType(
            type_=get_restricted_annotation(type_hint.__args__[0])  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownArgumentType]
        )
    elif get_origin(type_hint) is NotRequired:
        new_type_hint = NotRequiredType(type_=type_hint.__args__[0])  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownArgumentType]
    # NOTE: ForwardRefs are cursed, so we're excluding from test coverage
    # for now.
    elif get_origin(type_hint) is ForwardRef:  # pragma: no cover
        new_type_hint = ForwardRefType(type_=type_hint.__forward_arg__)  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownArgumentType]
    return new_type_hint

get_annotation

get_annotation(
    param: Parameter | DocstringParameter,
    func_for_forward_refs: (
        Callable[..., ParamType] | None
    ) = None,
) -> TypeOfType

Get the annotation of a parameter.

PARAMETER	DESCRIPTION
param	The parameter to get the annotation from. TYPE: Parameter | DocstringParameter
func_for_forward_refs	The function to get the annotation from. (Optional, and only here as fallback for ForwardRefs.) TYPE: Callable[..., ParamType] | None DEFAULT: None

RETURNS	DESCRIPTION
TypeOfType	The annotation of the parameter.

Source code in conatus/actions/utils/type_hint_handling.py

def get_annotation(
    param: Parameter | DocstringParameter,
    func_for_forward_refs: ABCCallable[..., ParamType] | None = None,
) -> TypeOfType:
    """Get the annotation of a parameter.

    Args:
        param: The parameter to get the annotation from.
        func_for_forward_refs: The function to get the annotation from.
            (Optional, and only here as fallback for ForwardRefs.)

    Returns:
        The annotation of the parameter.
    """
    annotation = cast("ParamType | type", param.annotation)
    func = func_for_forward_refs
    if func is not None and isinstance(annotation, str):
        annotation = cast(
            "ParamType | type",
            get_type_hints(
                func,
                globalns=func.__globals__,
                include_extras=True,
            )[param.name],
        )
    if isinstance(param, Parameter):
        return annotation if annotation != param.empty else Any
    return annotation if annotation != inspect._empty else Any  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]

get_return_annotation

get_return_annotation(sig: Signature) -> TypeOfType

Get the return annotation of a signature.

PARAMETER	DESCRIPTION
sig	The signature to get the return annotation from. TYPE: Signature

RETURNS	DESCRIPTION
TypeOfType	The return annotation of the signature.

Source code in conatus/actions/utils/type_hint_handling.py

def get_return_annotation(sig: Signature) -> TypeOfType:
    """Get the return annotation of a signature.

    Args:
        sig: The signature to get the return annotation from.

    Returns:
        The return annotation of the signature.
    """
    return_annotation = cast(
        "ParamType | type", getattr(sig, "return_annotation", None)
    )
    if return_annotation is None:
        return None
    if return_annotation == sig.empty:
        return Any
    return return_annotation

get_default_value

get_default_value(
    param: Parameter | DocstringParameter,
) -> ParamType

Get the default value of a parameter.

PARAMETER	DESCRIPTION
param	The parameter to get the default value from. TYPE: Parameter | DocstringParameter

RETURNS	DESCRIPTION
ParamType	The default value of the parameter.

Source code in conatus/actions/utils/type_hint_handling.py

def get_default_value(param: Parameter | DocstringParameter) -> ParamType:
    """Get the default value of a parameter.

    Args:
        param: The parameter to get the default value from.

    Returns:
        The default value of the parameter.
    """
    default = cast("ParamType | type", param.default)
    if isinstance(param, Parameter):
        return default if default != param.empty else ...
    return default if default != inspect._empty else ...  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]

get_description_from_typehint

get_description_from_typehint(
    type_hint: TypeOfType,
) -> str | None

Get the description from a type hint.

PARAMETER	DESCRIPTION
type_hint	The type hint to get the description from. TYPE: TypeOfType

RETURNS	DESCRIPTION
str | None	The description of the type hint.

Source code in conatus/actions/utils/type_hint_handling.py

def get_description_from_typehint(
    type_hint: TypeOfType,
) -> str | None:
    """Get the description from a type hint.

    Args:
        type_hint: The type hint to get the description from.

    Returns:
        The description of the type hint.
    """
    if get_origin(type_hint) in AnnotatedPythonTypes and (
        metadata := getattr(type_hint, "__metadata__", None)
    ):
        for m in cast("list[ParamType]", metadata):
            if isinstance(m, str):
                return m
    return None

is_json_serializable_value

is_json_serializable_value(value: ParamType) -> bool

Check whether a value is JSON-serializable.

PARAMETER	DESCRIPTION
value	The value to check. TYPE: ParamType

RETURNS	DESCRIPTION
bool	Whether the value is JSON-serializable.

Source code in conatus/actions/utils/type_hint_handling.py

def is_json_serializable_value(value: ParamType) -> bool:
    """Check whether a value is JSON-serializable.

    Args:
        value: The value to check.

    Returns:
        Whether the value is JSON-serializable.
    """
    try:
        _ = json.dumps(value)
    except TypeError:
        return False
    return True

is_json_serializable_type

is_json_serializable_type(
    type_hint: TypeOfType,
) -> tuple[bool, TypeOfType]

Check whether part of all of a type hint is JSON-serializable.

This function returns two values: a boolean indicating whether the type hint is JSON-serializable, and the part of the type hint that is JSON-serializable. If the type hint is never JSON-serializable, we return Ellipsis as the second value. If the type hint is always JSON-serializable, we return the type hint itself.

More information on why we need to check for JSON-serializable-ness, and why we need to determine a JSON-serializable subtype of a type hint can be found in the Action internals documentation.

PARAMETER	DESCRIPTION
type_hint	The type hint to check. TYPE: TypeOfType

RETURNS	DESCRIPTION
bool	Whether the type is SOMETIMES JSON-serializable.
TypeOfType	The part of the type hint that is JSON-serializable. If the type hint is JSON-serializable, we return Ellipsis.

Source code in conatus/actions/utils/type_hint_handling.py

def is_json_serializable_type(type_hint: TypeOfType) -> tuple[bool, TypeOfType]:  # noqa: PLR0911, PLR0912, C901
    """Check whether part of all of a type hint is JSON-serializable.

    This function returns two values: a boolean indicating whether the type
    hint is JSON-serializable, and the part of the type hint that is
    JSON-serializable. If the type hint is never JSON-serializable, we return
    Ellipsis as the second value. If the type hint is always JSON-serializable,
    we return the type hint itself.

    More information on why we need to check for JSON-serializable-ness,
    and why we need to determine a JSON-serializable subtype of a type hint
    can be found in the [`Action` internals](
    ../internals/action.md#what-is-a-json-serializable-type) documentation.

    Args:
        type_hint: The type hint to check.

    Returns:
        (bool): Whether the type is SOMETIMES JSON-serializable.
        (TypeOfType): The part of the type hint that is JSON-serializable. If
            the type hint is JSON-serializable, we return Ellipsis.
    """
    if type_hint in json_scalars or type_hint is None:
        return True, type_hint

    origin = cast("TypeOfType", get_origin(type_hint))
    args = cast("tuple[TypeOfType, ...]", get_args(type_hint))

    # If it's a Callable, that's not JSON-serializable
    if origin in OneOfCallables:
        return False, ...

    if isinstance(type_hint, TypeAliasType):
        return is_json_serializable_type(
            cast("TypeOfType", type_hint.__value__)
        )

    # Handle dataclasses, Pydantic, and TypedDict
    res: tuple[bool, TypeOfType] | None = (
        _is_json_serializable_type_handle_dataclass(type_hint)
    )
    if res:
        return res
    if is_typeddict(type_hint) or is_typeddict_ext(type_hint):
        return _is_json_serializable_type_handle_typeddict(type_hint)

    # If no origin, could be a plain type like `int` or `Any` or a forward ref
    if origin is None:
        return _is_json_serializable_type_handle_none_origin(type_hint)
    # Handle Union and Optional
    if origin is Union or origin is UnionType:  # pyright: ignore[reportDeprecated]
        return _is_json_serializable_type_handle_union(args)

    # Handle sequences and mappings
    if origin in {list, List, tuple, Tuple, Sequence, set}:  # noqa: UP006 # pyright: ignore[reportDeprecated]
        return _is_json_serializable_type_handle_sequence(args, type_hint)
    if origin in {dict, Dict, Mapping}:  # noqa: UP006 # pyright: ignore[reportDeprecated]
        return _is_json_serializable_type_handle_mapping(args, type_hint)

    # Handle wrapper types like Required, NotRequired, Annotated, Literal, etc.
    if origin in {Required, NotRequired, Annotated} or (
        str(origin).endswith("ClassVar") or str(origin).endswith("Final")
    ):
        existing, args = _detect_annotated_for_existing_variable(args)
        # This is a signal that the type hint is not JSON-serializable
        # because it's used to indicate that the variable is an existing
        # variable.
        if existing:
            return False, ...
        return is_json_serializable_type(args[0])
    if origin is Literal:
        return _is_json_serializable_type_handle_literal(args)

    # Handle Optional: we need to add None to the type hint
    # I have never see that happen though, so I think it's mostly for backward
    # compatibility
    if origin is Optional:  # pyright: ignore[reportDeprecated] # pragma: no cover
        return True, cast("type", is_json_serializable_type(args[0])[1]) | None

    # NOTE: NoneType is JSON-serializable, although it's not a type hint
    # Keeping this just in case this comes up
    if type_hint == NoneType:  # pragma: no cover
        return True, None

    # Too complex --> False
    return False, ...

process_typehint

process_typehint(
    type_hint: TypeOfType,
    return_type: Literal["str"],
    *,
    allow_qualified_names: bool = True
) -> str

process_typehint(
    type_hint: TypeOfType,
    return_type: Literal["set"],
    *,
    allow_qualified_names: bool = False
) -> set[type | str]

process_typehint(
    type_hint: TypeOfType,
    return_type: Literal["set", "str"] = "str",
    *,
    allow_qualified_names: bool = True
) -> str | set[type | str]

Process type hints.

This function recursively unpacks type hints. It has two uses:

1. To generate a type hint for the LLM. In this case, we want to return a string that is more likely to be understood by the LLM.
2. To get a set of all the types/literals encountered in the type hint. In this case, we want to return a set of the types/literals. We use this to be able to know all the imports that need to be added to the code.

Examples

Generating a type hint for the LLM

from typing import Union
from conatus.actions.utils.type_hint_handling import process_typehint
assert process_typehint(Union[int, str]) == "int | str"

Get all the types/literals encountered in the type hint

from conatus.actions.utils.type_hint_handling import process_typehint
assert process_typehint(int | str, return_type="set") == {int, str}

PARAMETER	DESCRIPTION
type_hint	The original type hint. TYPE: TypeOfType
return_type	The expected return type. If "set", return a set of the types/literals encountered in the type hint; if "str", return a human-readable string. TYPE: Literal['set', 'str'] DEFAULT: 'str'
allow_qualified_names	In string mode, whether to allow qualified names, e.g. pandas.core.frame.DataFrame. If this is False, we make sure that only the type name is returned, e.g. DataFrame. If you use this method to display a type to a LLM, you should set this to True. If you use this method to retrieve imports, you should set this to False. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str | set[type | str]	Either the type hint as a string or a set of the types.

Source code in conatus/actions/utils/type_hint_handling.py

def process_typehint(
    type_hint: TypeOfType,
    return_type: Literal["set", "str"] = "str",
    *,
    allow_qualified_names: bool = True,
) -> str | set[type | str]:
    """Process type hints.

    This function recursively unpacks type hints. It has two uses:

    1. To generate a type hint for the LLM. In this case, we want to return a
       string that is more likely to be understood by the LLM.
    2. To get a set of all the types/literals encountered in the type hint.
       In this case, we want to return a set of the types/literals. We use
       this to be able to know all the imports that need to be added to the
       code.

    # Examples

    ## Generating a type hint for the LLM

    ```python
    from typing import Union
    from conatus.actions.utils.type_hint_handling import process_typehint
    assert process_typehint(Union[int, str]) == "int | str"
    ```

    ## Get all the types/literals encountered in the type hint

    ```python
    from conatus.actions.utils.type_hint_handling import process_typehint
    assert process_typehint(int | str, return_type="set") == {int, str}
    ```

    Args:
        type_hint: The original type hint.
        return_type: The expected return type. If "set", return a set of the
            types/literals encountered in the type hint; if "str", return a
            human-readable string.
        allow_qualified_names: In string mode, whether to allow qualified names,
            e.g. `pandas.core.frame.DataFrame`. If this is False, we make sure
            that only the type name is returned, e.g. `DataFrame`. If you
            use this method to display a type to a LLM, you should set this
            to `True`. If you use this method to retrieve imports, you should
            set this to `False`.

    Returns:
        Either the type hint as a string or a set of the types.
    """
    # Unpack some Annotated values
    origin = cast("TypeOfType", get_origin(type_hint))
    args = cast("tuple[TypeOfType, ...]", get_args(type_hint))

    # If the type hint is Annotated, we unpack it
    if origin in AnnotatedPythonTypes:
        return _handle_annotated(
            origin=cast("type", origin),
            args=args,
            return_type=return_type,
        )

    if origin in {Union, UnionType}:  # pyright: ignore[reportDeprecated]
        return _handle_union(
            args=args,
            return_type=return_type,
        )

    if origin in {Literal, LiteralExt}:
        return _handle_literal(
            origin=origin,  # type: ignore[arg-type]
            args=args,
            return_type=return_type,
        )

    if len(get_args(type_hint)) > 0:
        return _handle_generic(
            type_hint=type_hint,
            origin=origin,
            args=args,
            return_type=return_type,
        )

    if isinstance(type_hint, list | tuple):
        type_hint = cast("list[TypeOfType] | tuple[TypeOfType, ...]", type_hint)
        return _handle_list_tuple(
            type_hint=type_hint,
            return_type=return_type,
        )

    return _handle_str_and_cleanup(
        type_hint=cast("type | str", type_hint),
        return_type=return_type,
        allow_qualified_names=allow_qualified_names,
    )

conatus.actions.retrievability

Module defining the retrievability of actions.

MaybeRetrievableActionInfo module-attribute

MaybeRetrievableActionInfo = (
    RetrievableActionInfo | NonRetrievableActionInfo
)

Type alias for a retrievable or non-retrievable action info.

We separate the two to ensure correctness for the type checker.

NonRetrievableActionInfo dataclass

NonRetrievableActionInfo(
    origin_type: ActionOriginType,
    recipe_retrievable: Literal[False],
    reason_why_not_retrievable: ReasonWhyNotRetrievable,
    origin_module: str,
    origin_qualname: str,
)

Information about a non-retrievable action.

This is the case when the action is not retrievable because it's defined in a <locals> namespace or because some changes are done to the Action that makes it non retrievable because some configuration values are not JSON serializable.

origin_type instance-attribute

origin_type: ActionOriginType

The origin of the action.

Can be either: - "subclass": The action was created through a subclass of Action. - "decorator": The action was a function decorated with @action or passed to the action function.

recipe_retrievable instance-attribute

recipe_retrievable: Literal[False]

The action is not retrievable.

For more information, see RetrievableActionInfo.recipe_retrievable .

reason_why_not_retrievable instance-attribute

reason_why_not_retrievable: ReasonWhyNotRetrievable

The reason why the action is not retrievable.

For more information, see NonRetrievableActionInfo.reason_why_not_retrievable .

origin_module instance-attribute

origin_module: str

The module where the action was originally defined.

For more information, see RetrievableActionInfo.origin_module .

origin_qualname instance-attribute

origin_qualname: str

The fully qualified name of the action as originally defined.

For more information, see RetrievableActionInfo.origin_qualname .

RetrievabilityInstructions dataclass

RetrievabilityInstructions(
    imports: list[str], qualname: str, changes: str
)

Instructions to retrieve the action.

imports instance-attribute

imports: list[str]

The imports needed to retrieve the action.

The list is structured such that you want to do from {imports[:-2]} import {imports[-1]}

qualname instance-attribute

qualname: str

The qualified name of the action after the imports have been executed.

changes instance-attribute

changes: str

The changes to the action that need to be made.

The changes are serialized as a JSON string.

This string might be {} if no changes are needed, and might contain values if the action is a result of a with_config call or if the user has changed the attributes of the action.

to_python_lines

to_python_lines() -> tuple[str, str | None]

Convert the retrievability instructions to a Python string.

RETURNS	DESCRIPTION
str	A tuple of two strings:
str | None	The import line
tuple[str, str | None]	The changes line

Source code in conatus/actions/retrievability.py

def to_python_lines(self) -> tuple[str, str | None]:
    """Convert the retrievability instructions to a Python string.

    Returns:
        A tuple of two strings:
        - The import line
        - The changes line
    """
    import_str = f"from {'.'.join(self.imports)} import {self.qualname}"
    if self.changes != "{}":
        changes_dict: dict[str, JSONType] = json.loads(self.changes)  # pyright: ignore[reportAny]
        changes_str = (
            f"{self.qualname} = {self.qualname}."
            "with_config("
            + ", ".join(
                [
                    f"{key}={json.dumps(value)}"
                    for key, value in changes_dict.items()
                ]
            )
            + ")"
        )
        return import_str, changes_str
    return import_str, None

RetrievableActionInfo dataclass

RetrievableActionInfo(
    origin_type: ActionOriginType,
    recipe_retrievable: Literal[True],
    reason_why_not_retrievable: None,
    origin_module: str,
    origin_qualname: str,
)

Information about a retrievable action.

It should be the default case that Action instances are retrievable, in that we can close the Python runtime, re-open it, and still be able to retrieve the action by running an import and changing its configuration parameters.

In this case: * recipe_retrievable is True * reason_why_not_retrievable is None * origin_module is the module where the action was originally defined * origin_qualname is the fully qualified name of the action as originally defined

Sometimes, the action is not retrievable. This happens, for example, if the action is defined in a <locals> namespace. In this case, Action will use the [NonRetrievableActionInfo] class to store the information about the action.

origin_type instance-attribute

origin_type: ActionOriginType

The origin of the action.

Can be either: - "subclass": The action was created through a subclass of Action. - "decorator": The action was a function decorated with @action or passed to the action function.

recipe_retrievable instance-attribute

recipe_retrievable: Literal[True]

Whether the action is retrievable in recipes.

This is evaluated based on whether the action can be imported like a normal Python function or class. In other words, it will not be retrievable if: - The action is part of a namespace (e.g. a function defined inside another function). - The action is defined in the Python REPL / a Jupyter notebook.

reason_why_not_retrievable instance-attribute

reason_why_not_retrievable: None

The reason why the action is not retrievable in recipes.

This attribute is set if _recipe_retrievable is False, and contains information about why the action is not retrievable. If the action is retrievable, this attribute is None.

origin_module instance-attribute

origin_module: str

The module where the action was originally defined.

To be more precise:

• If the action was created through a subclass of Action, this will be the module where the subclass was defined.
• If the action was created through a function decorated with @action or passed to the action function, this will be the module where the function was defined.

This attribute enables us to retrieve the original action or function when recipes are executed.

origin_qualname instance-attribute

origin_qualname: str

The fully qualified name of the action as originally defined.

To be more precise: - If the action was created through a subclass of Action, this will be the fully qualified name of the subclass. - If the action was created through a function decorated with @action or passed to the action function, this will be the fully qualified name of the function.

This attribute enables us to retrieve the original action or function when recipes are executed.

change_to_non_retrievable

change_to_non_retrievable(
    reason_why_not_retrievable: ReasonWhyNotRetrievable,
) -> NonRetrievableActionInfo

Change the action to non-retrievable.

This happens, for example, if some changes are done to the [Action] that makes it non retrievable because some configuration values are not JSON serializable.

PARAMETER	DESCRIPTION
reason_why_not_retrievable	The reason why the action is not retrievable. TYPE: ReasonWhyNotRetrievable

RETURNS	DESCRIPTION
NonRetrievableActionInfo	The non-retrievable action info.

Source code in conatus/actions/retrievability.py

def change_to_non_retrievable(
    self, reason_why_not_retrievable: ReasonWhyNotRetrievable
) -> NonRetrievableActionInfo:
    """Change the action to non-retrievable.

    This happens, for example, if some changes are done to the [`Action`]
    that makes it non retrievable because some configuration values are not
    JSON serializable.

    Args:
        reason_why_not_retrievable: The reason why the action is not
            retrievable.

    Returns:
        The non-retrievable action info.
    """
    return NonRetrievableActionInfo(
        origin_type=self.origin_type,
        recipe_retrievable=False,
        reason_why_not_retrievable=reason_why_not_retrievable,
        origin_module=self.origin_module,
        origin_qualname=self.origin_qualname,
    )

get_retrievability_instructions

get_retrievability_instructions(
    changes_as_json: str | None = None,
) -> RetrievabilityInstructions

Get the instructions to retrieve the action.

PARAMETER	DESCRIPTION
changes_as_json	The changes to the action as a JSON string. Defaults to None, which means no changes are needed. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
RetrievabilityInstructions	The retrievability instructions.

Source code in conatus/actions/retrievability.py

def get_retrievability_instructions(
    self, changes_as_json: str | None = None
) -> RetrievabilityInstructions:
    """Get the instructions to retrieve the action.

    Args:
        changes_as_json: The changes to the action as a JSON string.
            Defaults to `None`, which means no changes are needed.

    Returns:
        The retrievability instructions.
    """
    if changes_as_json is None:
        changes_as_json = "{}"

    return RetrievabilityInstructions(
        imports=self.origin_module.split("."),
        qualname=self.origin_qualname,
        changes=changes_as_json,
    )