Recipe

conatus.recipes.recipe.Recipe dataclass

Recipe(
    steps: list[RuntimeStateStep] = list(),
    actions: list[Action] = list(),
)

A recipe.

steps class-attribute instance-attribute

steps: list[RuntimeStateStep] = field(default_factory=list)

The steps of the recipe.

actions class-attribute instance-attribute

actions: list[Action] = field(default_factory=list)

The actions of the recipe.

get_all_imports

get_all_imports(
    *, skip_action_imports: Literal[True]
) -> str

get_all_imports(
    *, skip_action_imports: bool = False
) -> str | None

get_all_imports(
    *, skip_action_imports: bool = False
) -> str | None

Get all the imports of the recipe.

PARAMETER	DESCRIPTION
skip_action_imports	Whether to skip the imports of the actions. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str | None	The code of the imports of the recipe, or None if the recipe contains non-retrievable actions.

Source code in conatus/recipes/recipe.py

def get_all_imports(  # noqa: C901, PLR0912
    self, *, skip_action_imports: bool = False
) -> str | None:
    """Get all the imports of the recipe.

    Args:
        skip_action_imports: Whether to skip the imports of the actions.

    Returns:
        The code of the imports of the recipe, or None if the recipe
            contains non-retrievable actions.
    """
    imports: dict[str, set[str]] = {}
    as_imports: list[tuple[str, str, str]] = []

    if not skip_action_imports:
        for action in self.actions:
            info = action.retrievability_info
            if not info.recipe_retrievable:
                return None
            # statement like "from xxx import yyy as zzz"
            if info.origin_qualname != action.name:
                as_imports.append(
                    (info.origin_module, info.origin_qualname, action.name)
                )
            elif info.origin_module not in imports:
                imports[info.origin_module] = {info.origin_qualname}
            else:
                imports[info.origin_module].add(info.origin_qualname)

    for step in self.steps:
        for module, qualname in step.import_statements():
            if module not in imports:
                imports[module] = {qualname}
            else:
                imports[module].add(qualname)

    code = ""
    for module in sorted(imports.keys()):
        if module in imports:
            code += f"from {module} import {', '.join(imports[module])}\n"
    for module, qualname, name in as_imports:
        code += f"from {module} import {qualname} as {name}\n"
    return code

replace_terminate staticmethod

replace_terminate(
    line: str, task_definition: TaskDefinition
) -> str

Replace 'terminate(...)' statements with 'return ...' ones.

PARAMETER	DESCRIPTION
line	The Python statement to modify. TYPE: str
task_definition	The definition of the task. TYPE: TaskDefinition

RETURNS	DESCRIPTION
str	The new statement, potentially modified to a return statement.

RAISES	DESCRIPTION
ValueError	If the assignment in the terminate statement is manifestly incorrect.

Source code in conatus/recipes/recipe.py

@staticmethod
def replace_terminate(line: str, task_definition: TaskDefinition) -> str:
    """Replace 'terminate(...)' statements with 'return ...' ones.

    Args:
        line: The Python statement to modify.
        task_definition: The definition of the task.

    Returns:
        The new statement, potentially modified to a return
            statement.

    Raises:
        ValueError: If the assignment in the terminate statement is
            manifestly incorrect.
    """
    outputs_form = task_definition.outputs_form
    inner: str = line[len("terminate(") : -1].strip()
    if not inner:
        if outputs_form == "dict":
            return "return {}"
        if outputs_form in {"list", "singleton"}:
            return "return "
        if outputs_form in {"none", None}:
            return "return None"
    parts: list[str] = inner.split(",")
    keys: list[str] = []
    values: list[str] = []
    for part in parts:
        part_stripped = part.strip()
        if "=" not in part_stripped:
            msg = f"Invalid argument assignment: {part_stripped}"
            raise ValueError(msg)
        key, value = part_stripped.split("=", 1)
        keys.append(key.strip())
        values.append(value.strip())

    if outputs_form == "dict":
        items: list[str] = []
        for k, v in zip(keys, values, strict=True):
            items.append(f'"{k}": {v}')
        return "return {" + ", ".join(items) + "}"
    return "return " + ", ".join(values)

to_code

to_code(
    task_definition: TaskDefinition,
    *,
    skip_action_imports: Literal[True]
) -> str

to_code(
    task_definition: TaskDefinition,
    *,
    skip_action_imports: bool = False
) -> str | None

to_code(
    task_definition: TaskDefinition,
    *,
    skip_action_imports: bool = False
) -> str | None

Convert the recipe to code.

PARAMETER	DESCRIPTION
task_definition	The definition of the task. TYPE: TaskDefinition
skip_action_imports	Whether to skip the imports of the actions. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str | None	The code of the recipe.

Source code in conatus/recipes/recipe.py

def to_code(
    self,
    task_definition: TaskDefinition,
    *,
    skip_action_imports: bool = False,
) -> str | None:
    """Convert the recipe to code.

    Args:
        task_definition: The definition of the task.
        skip_action_imports: Whether to skip the imports of the actions.

    Returns:
        The code of the recipe.
    """
    code = ""
    imports = self.get_all_imports(skip_action_imports=skip_action_imports)
    if imports is None:
        return None
    code += imports

    code += "\n"

    code += f"def {task_definition.name}("
    arguments: list[str] = []
    for input_name, input_type in task_definition.inputs.items():
        type_hint_str = process_typehint(
            type_hint=input_type,
            return_type="str",
            allow_qualified_names=False,
        )
        arguments.append(f"{input_name}: {type_hint_str}")
    code += ", ".join(arguments) + "):\n"
    # TODO(lemeb): Add outputs type hints to the recipe
    # CTUS-70
    code += (
        '    """\n'
        + "\n".join(
            [
                f"    {line}"
                for line in task_definition.user_prompt.split("\n")
            ]
        )
        + '\n    """\n'
    )

    for step in self.steps:
        if step.step_number == 0:
            continue
        step_code = step.code()
        if step_code:
            for line in step_code.split("\n"):
                if line.startswith("terminate("):
                    return_line = self.replace_terminate(
                        line, task_definition
                    )
                    code += f"    {return_line}\n"
                else:
                    code += f"    {line}\n"
    return code