tasks

__all__ = ['BaseTask', 'BaseTextInputTask', 'BaseMultiTextInputTask', 'PromptTask', 'ActionsSubtask', 'ToolkitTask', 'TextSummaryTask', 'ToolTask', 'RagTask', 'ExtractionTask', 'BaseImageGenerationTask', 'CodeExecutionTask', 'PromptImageGenerationTask', 'VariationImageGenerationTask', 'InpaintingImageGenerationTask', 'OutpaintingImageGenerationTask', 'ImageQueryTask', 'BaseAudioGenerationTask', 'TextToSpeechTask', 'StructureRunTask', 'AudioTranscriptionTask'] module-attribute

ActionsSubtask

Bases: BaseTask

Source code in griptape/tasks/actions_subtask.py

@define
class ActionsSubtask(BaseTask):
    THOUGHT_PATTERN = r"(?s)^Thought:\s*(.*?)$"
    ACTIONS_PATTERN = r"(?s)Actions:[^\[]*(\[.*\])"
    ANSWER_PATTERN = r"(?s)^Answer:\s?([\s\S]*)$"

    parent_task_id: Optional[str] = field(default=None, kw_only=True)
    thought: Optional[str] = field(default=None, kw_only=True)
    actions: list[ToolAction] = field(factory=list, kw_only=True)
    output: Optional[BaseArtifact] = field(default=None, init=False)
    _input: str | list | tuple | BaseArtifact | Callable[[BaseTask], BaseArtifact] = field(
        default=lambda task: task.full_context["args"][0] if task.full_context["args"] else TextArtifact(value=""),
        alias="input",
    )
    _memory: Optional[TaskMemory] = None

    @property
    def input(self) -> TextArtifact | ListArtifact:
        return self._process_task_input(self._input)

    @input.setter
    def input(self, value: str | list | tuple | BaseArtifact | Callable[[BaseTask], BaseArtifact]) -> None:
        self._input = value

    @property
    def origin_task(self) -> BaseTask:
        if self.parent_task_id:
            return self.structure.find_task(self.parent_task_id)
        else:
            raise Exception("ActionSubtask has no parent task.")

    @property
    def parents(self) -> list[BaseTask]:
        if isinstance(self.origin_task, ActionsSubtaskOriginMixin):
            return [self.origin_task.find_subtask(parent_id) for parent_id in self.parent_ids]
        else:
            raise Exception("ActionSubtask must be attached to a Task that implements ActionSubtaskOriginMixin.")

    @property
    def children(self) -> list[BaseTask]:
        if isinstance(self.origin_task, ActionsSubtaskOriginMixin):
            return [self.origin_task.find_subtask(child_id) for child_id in self.child_ids]
        else:
            raise Exception("ActionSubtask must be attached to a Task that implements ActionSubtaskOriginMixin.")

    def add_child(self, child: BaseTask) -> BaseTask:
        if child.id not in self.child_ids:
            self.child_ids.append(child.id)
        return child

    def add_parent(self, parent: BaseTask) -> BaseTask:
        if parent.id not in self.parent_ids:
            self.parent_ids.append(parent.id)
        return parent

    def attach_to(self, parent_task: BaseTask) -> None:
        self.parent_task_id = parent_task.id
        self.structure = parent_task.structure

        try:
            if isinstance(self.input, TextArtifact):
                self.__init_from_prompt(self.input.to_text())
            else:
                self.__init_from_artifacts(self.input)
        except Exception as e:
            logger.error("Subtask %s\nError parsing tool action: %s", self.origin_task.id, e)

            self.output = ErrorArtifact(f"ToolAction input parsing error: {e}", exception=e)

    def before_run(self) -> None:
        EventBus.publish_event(
            StartActionsSubtaskEvent(
                task_id=self.id,
                task_parent_ids=self.parent_ids,
                task_child_ids=self.child_ids,
                task_input=self.input,
                task_output=self.output,
                subtask_parent_task_id=self.parent_task_id,
                subtask_thought=self.thought,
                subtask_actions=self.actions_to_dicts(),
            ),
        )

        parts = [
            f"Subtask {self.id}",
            *([f"\nThought: {self.thought}"] if self.thought else []),
            f"\nActions: {self.actions_to_json()}",
        ]
        logger.info("".join(parts))

    def run(self) -> BaseArtifact:
        try:
            if any(isinstance(a.output, ErrorArtifact) for a in self.actions):
                errors = [a.output.value for a in self.actions if isinstance(a.output, ErrorArtifact)]

                self.output = ErrorArtifact("\n\n".join(errors))
            else:
                results = self.execute_actions(self.actions)

                actions_output = []
                for result in results:
                    tag, output = result
                    output.name = f"{tag} output"

                    actions_output.append(output)
                self.output = ListArtifact(actions_output)
        except Exception as e:
            logger.exception("Subtask %s\n%s", self.id, e)

            self.output = ErrorArtifact(str(e), exception=e)
        if self.output is not None:
            return self.output
        else:
            return ErrorArtifact("no tool output")

    def execute_actions(self, actions: list[ToolAction]) -> list[tuple[str, BaseArtifact]]:
        return utils.execute_futures_list([self.futures_executor.submit(self.execute_action, a) for a in actions])

    def execute_action(self, action: ToolAction) -> tuple[str, BaseArtifact]:
        if action.tool is not None:
            if action.path is not None:
                output = action.tool.execute(getattr(action.tool, action.path), self, action)
            else:
                output = ErrorArtifact("action path not found")
        else:
            output = ErrorArtifact("action name not found")
        action.output = output

        return action.tag, output

    def after_run(self) -> None:
        response = self.output.to_text() if isinstance(self.output, BaseArtifact) else str(self.output)

        EventBus.publish_event(
            FinishActionsSubtaskEvent(
                task_id=self.id,
                task_parent_ids=self.parent_ids,
                task_child_ids=self.child_ids,
                task_input=self.input,
                task_output=self.output,
                subtask_parent_task_id=self.parent_task_id,
                subtask_thought=self.thought,
                subtask_actions=self.actions_to_dicts(),
            ),
        )
        logger.info("Subtask %s\nResponse: %s", self.id, response)

    def actions_to_dicts(self) -> list[dict]:
        json_list = []

        for action in self.actions:
            json_dict = {}

            if action.tag:
                json_dict["tag"] = action.tag

            if action.name:
                json_dict["name"] = action.name

            if action.path:
                json_dict["path"] = action.path

            if action.input:
                json_dict["input"] = action.input

            json_list.append(json_dict)

        return json_list

    def actions_to_json(self) -> str:
        return json.dumps(self.actions_to_dicts(), indent=2)

    def _process_task_input(
        self,
        task_input: str | tuple | list | BaseArtifact | Callable[[BaseTask], BaseArtifact],
    ) -> TextArtifact | ListArtifact:
        if isinstance(task_input, (TextArtifact, ListArtifact)):
            return task_input
        elif isinstance(task_input, ActionArtifact):
            return ListArtifact([task_input])
        elif isinstance(task_input, Callable):
            return self._process_task_input(task_input(self))
        elif isinstance(task_input, str):
            return self._process_task_input(TextArtifact(task_input))
        elif isinstance(task_input, (list, tuple)):
            return ListArtifact([self._process_task_input(elem) for elem in task_input])
        else:
            raise ValueError(f"Invalid input type: {type(task_input)} ")

    def __init_from_prompt(self, value: str) -> None:
        thought_matches = re.findall(self.THOUGHT_PATTERN, value, re.MULTILINE)
        actions_matches = re.findall(self.ACTIONS_PATTERN, value, re.DOTALL)
        answer_matches = re.findall(self.ANSWER_PATTERN, value, re.MULTILINE)

        if self.thought is None and thought_matches:
            self.thought = thought_matches[-1]

        self.__parse_actions(actions_matches)

        # If there are no actions to take but an answer is provided, set the answer as the output.
        if len(self.actions) == 0 and self.output is None and answer_matches:
            self.output = TextArtifact(answer_matches[-1])

    def __init_from_artifacts(self, artifacts: ListArtifact) -> None:
        """Parses the input Artifacts to extract the thought and actions.

        Text Artifacts are used to extract the thought, and ToolAction Artifacts are used to extract the actions.

        Args:
            artifacts: The input Artifacts.

        Returns:
            None
        """
        self.actions = [
            self.__process_action_object(artifact.value.to_dict())
            for artifact in artifacts.value
            if isinstance(artifact, ActionArtifact)
        ]

        thoughts = [artifact.value for artifact in artifacts.value if isinstance(artifact, TextArtifact)]
        if thoughts:
            self.thought = thoughts[0]

    def __parse_actions(self, actions_matches: list[str]) -> None:
        if len(actions_matches) == 0:
            return
        try:
            data = actions_matches[-1]
            actions_list: list[dict] = json.loads(data, strict=False)

            self.actions = [self.__process_action_object(action_object) for action_object in actions_list]
        except json.JSONDecodeError as e:
            logger.exception("Subtask %s\nInvalid actions JSON: %s", self.origin_task.id, e)

            self.output = ErrorArtifact(f"Actions JSON decoding error: {e}", exception=e)

    def __process_action_object(self, action_object: dict) -> ToolAction:
        # Load action tag; throw exception if the key is not present
        action_tag = action_object["tag"]

        # Load action name; throw exception if the key is not present
        action_name = action_object["name"]

        # Load action method; throw exception if the key is not present
        action_path = action_object["path"]

        # Load optional input value; don't throw exceptions if key is not present
        if "input" in action_object:
            # Some LLMs don't support nested parameters and therefore won't generate "values".
            # So we need to manually add it here.
            if "values" not in action_object["input"]:
                action_object["input"] = {"values": action_object["input"]}

            # The schema library has a bug, where something like `Or(str, None)` doesn't get
            # correctly translated into JSON schema. For some optional input fields LLMs sometimes
            # still provide null value, which trips up the validator. The temporary solution that
            # works is to strip all key-values where value is null.
            action_input = remove_null_values_in_dict_recursively(action_object["input"])
        else:
            action_input = {}

        # Load the action itself
        if isinstance(self.origin_task, ActionsSubtaskOriginMixin):
            tool = self.origin_task.find_tool(action_name)
        else:
            raise Exception("ActionSubtask must be attached to a Task that implements ActionSubtaskOriginMixin.")

        action = ToolAction(tag=action_tag, name=action_name, path=action_path, input=action_input, tool=tool)

        if action.tool and action.input:
            self.__validate_action(action)

        return action

    def __validate_action(self, action: ToolAction) -> None:
        try:
            if action.path is not None:
                activity = getattr(action.tool, action.path)
            else:
                raise Exception("ToolAction path not found.")

            if activity is not None:
                activity_schema = action.tool.activity_schema(activity)
            else:
                raise Exception("Activity not found.")

            if activity_schema:
                activity_schema.validate(action.input)
        except schema.SchemaError as e:
            logger.exception("Subtask %s\nInvalid action JSON: %s", self.origin_task.id, e)

            action.output = ErrorArtifact(f"Activity input JSON validation error: {e}", exception=e)
        except SyntaxError as e:
            logger.exception("Subtask %s\nSyntax error: %s", self.origin_task.id, e)

            action.output = ErrorArtifact(f"Syntax error: {e}", exception=e)

ACTIONS_PATTERN = '(?s)Actions:[^\\[]*(\\[.*\\])' class-attribute instance-attribute

ANSWER_PATTERN = '(?s)^Answer:\\s?([\\s\\S]*)$' class-attribute instance-attribute

THOUGHT_PATTERN = '(?s)^Thought:\\s*(.*?)$' class-attribute instance-attribute

actions: list[ToolAction] = field(factory=list, kw_only=True) class-attribute instance-attribute

children: list[BaseTask] property

input: TextArtifact | ListArtifact property writable

origin_task: BaseTask property

output: Optional[BaseArtifact] = field(default=None, init=False) class-attribute instance-attribute

parent_task_id: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

parents: list[BaseTask] property

thought: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

__init_from_artifacts(artifacts)

Parses the input Artifacts to extract the thought and actions.

Text Artifacts are used to extract the thought, and ToolAction Artifacts are used to extract the actions.

Parameters:

Name	Type	Description	Default
artifacts	ListArtifact	The input Artifacts.	required

Returns:

Type	Description
None	None

Source code in griptape/tasks/actions_subtask.py

def __init_from_artifacts(self, artifacts: ListArtifact) -> None:
    """Parses the input Artifacts to extract the thought and actions.

    Text Artifacts are used to extract the thought, and ToolAction Artifacts are used to extract the actions.

    Args:
        artifacts: The input Artifacts.

    Returns:
        None
    """
    self.actions = [
        self.__process_action_object(artifact.value.to_dict())
        for artifact in artifacts.value
        if isinstance(artifact, ActionArtifact)
    ]

    thoughts = [artifact.value for artifact in artifacts.value if isinstance(artifact, TextArtifact)]
    if thoughts:
        self.thought = thoughts[0]

__init_from_prompt(value)

Source code in griptape/tasks/actions_subtask.py

def __init_from_prompt(self, value: str) -> None:
    thought_matches = re.findall(self.THOUGHT_PATTERN, value, re.MULTILINE)
    actions_matches = re.findall(self.ACTIONS_PATTERN, value, re.DOTALL)
    answer_matches = re.findall(self.ANSWER_PATTERN, value, re.MULTILINE)

    if self.thought is None and thought_matches:
        self.thought = thought_matches[-1]

    self.__parse_actions(actions_matches)

    # If there are no actions to take but an answer is provided, set the answer as the output.
    if len(self.actions) == 0 and self.output is None and answer_matches:
        self.output = TextArtifact(answer_matches[-1])

__parse_actions(actions_matches)

Source code in griptape/tasks/actions_subtask.py

def __parse_actions(self, actions_matches: list[str]) -> None:
    if len(actions_matches) == 0:
        return
    try:
        data = actions_matches[-1]
        actions_list: list[dict] = json.loads(data, strict=False)

        self.actions = [self.__process_action_object(action_object) for action_object in actions_list]
    except json.JSONDecodeError as e:
        logger.exception("Subtask %s\nInvalid actions JSON: %s", self.origin_task.id, e)

        self.output = ErrorArtifact(f"Actions JSON decoding error: {e}", exception=e)

__process_action_object(action_object)

Source code in griptape/tasks/actions_subtask.py

def __process_action_object(self, action_object: dict) -> ToolAction:
    # Load action tag; throw exception if the key is not present
    action_tag = action_object["tag"]

    # Load action name; throw exception if the key is not present
    action_name = action_object["name"]

    # Load action method; throw exception if the key is not present
    action_path = action_object["path"]

    # Load optional input value; don't throw exceptions if key is not present
    if "input" in action_object:
        # Some LLMs don't support nested parameters and therefore won't generate "values".
        # So we need to manually add it here.
        if "values" not in action_object["input"]:
            action_object["input"] = {"values": action_object["input"]}

        # The schema library has a bug, where something like `Or(str, None)` doesn't get
        # correctly translated into JSON schema. For some optional input fields LLMs sometimes
        # still provide null value, which trips up the validator. The temporary solution that
        # works is to strip all key-values where value is null.
        action_input = remove_null_values_in_dict_recursively(action_object["input"])
    else:
        action_input = {}

    # Load the action itself
    if isinstance(self.origin_task, ActionsSubtaskOriginMixin):
        tool = self.origin_task.find_tool(action_name)
    else:
        raise Exception("ActionSubtask must be attached to a Task that implements ActionSubtaskOriginMixin.")

    action = ToolAction(tag=action_tag, name=action_name, path=action_path, input=action_input, tool=tool)

    if action.tool and action.input:
        self.__validate_action(action)

    return action

__validate_action(action)

Source code in griptape/tasks/actions_subtask.py

def __validate_action(self, action: ToolAction) -> None:
    try:
        if action.path is not None:
            activity = getattr(action.tool, action.path)
        else:
            raise Exception("ToolAction path not found.")

        if activity is not None:
            activity_schema = action.tool.activity_schema(activity)
        else:
            raise Exception("Activity not found.")

        if activity_schema:
            activity_schema.validate(action.input)
    except schema.SchemaError as e:
        logger.exception("Subtask %s\nInvalid action JSON: %s", self.origin_task.id, e)

        action.output = ErrorArtifact(f"Activity input JSON validation error: {e}", exception=e)
    except SyntaxError as e:
        logger.exception("Subtask %s\nSyntax error: %s", self.origin_task.id, e)

        action.output = ErrorArtifact(f"Syntax error: {e}", exception=e)

actions_to_dicts()

Source code in griptape/tasks/actions_subtask.py

def actions_to_dicts(self) -> list[dict]:
    json_list = []

    for action in self.actions:
        json_dict = {}

        if action.tag:
            json_dict["tag"] = action.tag

        if action.name:
            json_dict["name"] = action.name

        if action.path:
            json_dict["path"] = action.path

        if action.input:
            json_dict["input"] = action.input

        json_list.append(json_dict)

    return json_list

actions_to_json()

Source code in griptape/tasks/actions_subtask.py

def actions_to_json(self) -> str:
    return json.dumps(self.actions_to_dicts(), indent=2)

add_child(child)

Source code in griptape/tasks/actions_subtask.py

def add_child(self, child: BaseTask) -> BaseTask:
    if child.id not in self.child_ids:
        self.child_ids.append(child.id)
    return child

add_parent(parent)

Source code in griptape/tasks/actions_subtask.py

def add_parent(self, parent: BaseTask) -> BaseTask:
    if parent.id not in self.parent_ids:
        self.parent_ids.append(parent.id)
    return parent

after_run()

Source code in griptape/tasks/actions_subtask.py

def after_run(self) -> None:
    response = self.output.to_text() if isinstance(self.output, BaseArtifact) else str(self.output)

    EventBus.publish_event(
        FinishActionsSubtaskEvent(
            task_id=self.id,
            task_parent_ids=self.parent_ids,
            task_child_ids=self.child_ids,
            task_input=self.input,
            task_output=self.output,
            subtask_parent_task_id=self.parent_task_id,
            subtask_thought=self.thought,
            subtask_actions=self.actions_to_dicts(),
        ),
    )
    logger.info("Subtask %s\nResponse: %s", self.id, response)

attach_to(parent_task)

Source code in griptape/tasks/actions_subtask.py

def attach_to(self, parent_task: BaseTask) -> None:
    self.parent_task_id = parent_task.id
    self.structure = parent_task.structure

    try:
        if isinstance(self.input, TextArtifact):
            self.__init_from_prompt(self.input.to_text())
        else:
            self.__init_from_artifacts(self.input)
    except Exception as e:
        logger.error("Subtask %s\nError parsing tool action: %s", self.origin_task.id, e)

        self.output = ErrorArtifact(f"ToolAction input parsing error: {e}", exception=e)

before_run()

Source code in griptape/tasks/actions_subtask.py

def before_run(self) -> None:
    EventBus.publish_event(
        StartActionsSubtaskEvent(
            task_id=self.id,
            task_parent_ids=self.parent_ids,
            task_child_ids=self.child_ids,
            task_input=self.input,
            task_output=self.output,
            subtask_parent_task_id=self.parent_task_id,
            subtask_thought=self.thought,
            subtask_actions=self.actions_to_dicts(),
        ),
    )

    parts = [
        f"Subtask {self.id}",
        *([f"\nThought: {self.thought}"] if self.thought else []),
        f"\nActions: {self.actions_to_json()}",
    ]
    logger.info("".join(parts))

execute_action(action)

Source code in griptape/tasks/actions_subtask.py

def execute_action(self, action: ToolAction) -> tuple[str, BaseArtifact]:
    if action.tool is not None:
        if action.path is not None:
            output = action.tool.execute(getattr(action.tool, action.path), self, action)
        else:
            output = ErrorArtifact("action path not found")
    else:
        output = ErrorArtifact("action name not found")
    action.output = output

    return action.tag, output

execute_actions(actions)

Source code in griptape/tasks/actions_subtask.py

def execute_actions(self, actions: list[ToolAction]) -> list[tuple[str, BaseArtifact]]:
    return utils.execute_futures_list([self.futures_executor.submit(self.execute_action, a) for a in actions])

run()

Source code in griptape/tasks/actions_subtask.py

def run(self) -> BaseArtifact:
    try:
        if any(isinstance(a.output, ErrorArtifact) for a in self.actions):
            errors = [a.output.value for a in self.actions if isinstance(a.output, ErrorArtifact)]

            self.output = ErrorArtifact("\n\n".join(errors))
        else:
            results = self.execute_actions(self.actions)

            actions_output = []
            for result in results:
                tag, output = result
                output.name = f"{tag} output"

                actions_output.append(output)
            self.output = ListArtifact(actions_output)
    except Exception as e:
        logger.exception("Subtask %s\n%s", self.id, e)

        self.output = ErrorArtifact(str(e), exception=e)
    if self.output is not None:
        return self.output
    else:
        return ErrorArtifact("no tool output")

AudioTranscriptionTask

Bases: BaseAudioInputTask

Source code in griptape/tasks/audio_transcription_task.py

@define
class AudioTranscriptionTask(BaseAudioInputTask):
    audio_transcription_engine: AudioTranscriptionEngine = field(
        default=Factory(lambda: AudioTranscriptionEngine()),
        kw_only=True,
    )

    def run(self) -> TextArtifact:
        return self.audio_transcription_engine.run(self.input)

audio_transcription_engine: AudioTranscriptionEngine = field(default=Factory(lambda: AudioTranscriptionEngine()), kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/audio_transcription_task.py

def run(self) -> TextArtifact:
    return self.audio_transcription_engine.run(self.input)

BaseAudioGenerationTask

Bases: ArtifactFileOutputMixin, RuleMixin, BaseTask, ABC

Source code in griptape/tasks/base_audio_generation_task.py

@define
class BaseAudioGenerationTask(ArtifactFileOutputMixin, RuleMixin, BaseTask, ABC):
    def before_run(self) -> None:
        super().before_run()

        logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

    def after_run(self) -> None:
        super().after_run()

        logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

after_run()

Source code in griptape/tasks/base_audio_generation_task.py

def after_run(self) -> None:
    super().after_run()

    logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

before_run()

Source code in griptape/tasks/base_audio_generation_task.py

def before_run(self) -> None:
    super().before_run()

    logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

BaseImageGenerationTask

Bases: ArtifactFileOutputMixin, RuleMixin, BaseTask, ABC

Provides a base class for image generation-related tasks.

Attributes:

Name	Type	Description
negative_rulesets	list[Ruleset]	List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
negative_rules	list[Rule]	List of negatively-weighted rules applied to the text prompt, if supported by the driver.
output_dir	list[Rule]	If provided, the generated image will be written to disk in output_dir.
output_file	list[Rule]	If provided, the generated image will be written to disk as output_file.

Source code in griptape/tasks/base_image_generation_task.py

@define
class BaseImageGenerationTask(ArtifactFileOutputMixin, RuleMixin, BaseTask, ABC):
    """Provides a base class for image generation-related tasks.

    Attributes:
        negative_rulesets: List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
        negative_rules: List of negatively-weighted rules applied to the text prompt, if supported by the driver.
        output_dir: If provided, the generated image will be written to disk in output_dir.
        output_file: If provided, the generated image will be written to disk as output_file.
    """

    NEGATIVE_RULESET_NAME = "Negative Ruleset"

    negative_rulesets: list[Ruleset] = field(factory=list, kw_only=True)
    negative_rules: list[Rule] = field(factory=list, kw_only=True)

    @negative_rulesets.validator  # pyright: ignore[reportAttributeAccessIssue]
    def validate_negative_rulesets(self, _: Attribute, negative_rulesets: list[Ruleset]) -> None:
        if not negative_rulesets:
            return

        if self.negative_rules:
            raise ValueError("Can't have both negative_rulesets and negative_rules specified.")

    @negative_rules.validator  # pyright: ignore[reportAttributeAccessIssue]
    def validate_negative_rules(self, _: Attribute, negative_rules: list[Rule]) -> None:
        if not negative_rules:
            return

        if self.negative_rulesets:
            raise ValueError("Can't have both negative_rules and negative_rulesets specified.")

    @property
    def all_negative_rulesets(self) -> list[Ruleset]:
        task_rulesets = []
        if self.negative_rulesets:
            task_rulesets = self.negative_rulesets

        elif self.negative_rules:
            task_rulesets = [Ruleset(name=self.NEGATIVE_RULESET_NAME, rules=self.negative_rules)]

        return task_rulesets

    def _read_from_file(self, path: str) -> ImageArtifact:
        logger.info("Reading image from %s", os.path.abspath(path))
        return ImageLoader().load(Path(path))

NEGATIVE_RULESET_NAME = 'Negative Ruleset' class-attribute instance-attribute

all_negative_rulesets: list[Ruleset] property

negative_rules: list[Rule] = field(factory=list, kw_only=True) class-attribute instance-attribute

negative_rulesets: list[Ruleset] = field(factory=list, kw_only=True) class-attribute instance-attribute

validate_negative_rules(_, negative_rules)

Source code in griptape/tasks/base_image_generation_task.py

@negative_rules.validator  # pyright: ignore[reportAttributeAccessIssue]
def validate_negative_rules(self, _: Attribute, negative_rules: list[Rule]) -> None:
    if not negative_rules:
        return

    if self.negative_rulesets:
        raise ValueError("Can't have both negative_rules and negative_rulesets specified.")

validate_negative_rulesets(_, negative_rulesets)

Source code in griptape/tasks/base_image_generation_task.py

@negative_rulesets.validator  # pyright: ignore[reportAttributeAccessIssue]
def validate_negative_rulesets(self, _: Attribute, negative_rulesets: list[Ruleset]) -> None:
    if not negative_rulesets:
        return

    if self.negative_rules:
        raise ValueError("Can't have both negative_rulesets and negative_rules specified.")

BaseMultiTextInputTask

Bases: RuleMixin, BaseTask, ABC

Source code in griptape/tasks/base_multi_text_input_task.py

@define
class BaseMultiTextInputTask(RuleMixin, BaseTask, ABC):
    DEFAULT_INPUT_TEMPLATE = "{{ args[0] }}"

    _input: tuple[str, ...] | tuple[TextArtifact, ...] | tuple[Callable[[BaseTask], TextArtifact], ...] = field(
        default=Factory(lambda self: (self.DEFAULT_INPUT_TEMPLATE,), takes_self=True),
        alias="input",
    )

    @property
    def input(self) -> ListArtifact:
        if all(isinstance(elem, TextArtifact) for elem in self._input):
            return ListArtifact([artifact for artifact in self._input if isinstance(artifact, TextArtifact)])
        elif all(isinstance(elem, Callable) for elem in self._input):
            return ListArtifact(
                [callable_input(self) for callable_input in self._input if isinstance(callable_input, Callable)]
            )
        else:
            return ListArtifact(
                [
                    TextArtifact(J2().render_from_string(input_template, **self.full_context))
                    for input_template in self._input
                    if isinstance(input_template, str)
                ],
            )

    @input.setter
    def input(
        self,
        value: tuple[str, ...] | tuple[TextArtifact, ...] | tuple[Callable[[BaseTask], TextArtifact], ...],
    ) -> None:
        self._input = value

    def before_run(self) -> None:
        super().before_run()

        joined_input = "\n".join([i.to_text() for i in self.input])
        logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, joined_input)

    def after_run(self) -> None:
        super().after_run()

        logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

DEFAULT_INPUT_TEMPLATE = '{{ args[0] }}' class-attribute instance-attribute

input: ListArtifact property writable

after_run()

Source code in griptape/tasks/base_multi_text_input_task.py

def after_run(self) -> None:
    super().after_run()

    logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

before_run()

Source code in griptape/tasks/base_multi_text_input_task.py

def before_run(self) -> None:
    super().before_run()

    joined_input = "\n".join([i.to_text() for i in self.input])
    logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, joined_input)

BaseTask

Bases: FuturesExecutorMixin, ABC

Source code in griptape/tasks/base_task.py

@define
class BaseTask(FuturesExecutorMixin, ABC):
    class State(Enum):
        PENDING = 1
        EXECUTING = 2
        FINISHED = 3

    id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True)
    state: State = field(default=State.PENDING, kw_only=True)
    parent_ids: list[str] = field(factory=list, kw_only=True)
    child_ids: list[str] = field(factory=list, kw_only=True)
    max_meta_memory_entries: Optional[int] = field(default=20, kw_only=True)
    structure: Optional[Structure] = field(default=None, kw_only=True)

    output: Optional[BaseArtifact] = field(default=None, init=False)
    context: dict[str, Any] = field(factory=dict, kw_only=True)

    def __rshift__(self, other: BaseTask) -> BaseTask:
        self.add_child(other)

        return other

    def __lshift__(self, other: BaseTask) -> BaseTask:
        self.add_parent(other)

        return other

    def __attrs_post_init__(self) -> None:
        if self.structure is not None:
            self.structure.add_task(self)

    @property
    @abstractmethod
    def input(self) -> BaseArtifact: ...

    @property
    def parents(self) -> list[BaseTask]:
        if self.structure is not None:
            return [self.structure.find_task(parent_id) for parent_id in self.parent_ids]
        raise ValueError("Structure must be set to access parents")

    @property
    def children(self) -> list[BaseTask]:
        if self.structure is not None:
            return [self.structure.find_task(child_id) for child_id in self.child_ids]
        raise ValueError("Structure must be set to access children")

    @property
    def parent_outputs(self) -> dict[str, str]:
        return {parent.id: parent.output.to_text() if parent.output else "" for parent in self.parents}

    @property
    def parents_output_text(self) -> str:
        return "\n".join([parent.output.to_text() for parent in self.parents if parent.output])

    @property
    def meta_memories(self) -> list[BaseMetaEntry]:
        if self.structure and self.structure.meta_memory:
            if self.max_meta_memory_entries:
                return self.structure.meta_memory.entries[: self.max_meta_memory_entries]
            else:
                return self.structure.meta_memory.entries
        else:
            return []

    def __str__(self) -> str:
        return str(self.output.value)

    def add_parents(self, parents: list[BaseTask]) -> None:
        for parent in parents:
            self.add_parent(parent)

    def add_parent(self, parent: BaseTask) -> BaseTask:
        if parent.id not in self.parent_ids:
            self.parent_ids.append(parent.id)

        if self.id not in parent.child_ids:
            parent.child_ids.append(self.id)

        if self.structure is not None:
            self.structure.add_task(parent)

        return self

    def add_children(self, children: list[BaseTask]) -> None:
        for child in children:
            self.add_child(child)

    def add_child(self, child: BaseTask) -> BaseTask:
        if child.id not in self.child_ids:
            self.child_ids.append(child.id)

        if self.id not in child.parent_ids:
            child.parent_ids.append(self.id)

        if self.structure is not None:
            self.structure.add_task(child)

        return self

    def preprocess(self, structure: Structure) -> BaseTask:
        self.structure = structure

        return self

    def is_pending(self) -> bool:
        return self.state == BaseTask.State.PENDING

    def is_finished(self) -> bool:
        return self.state == BaseTask.State.FINISHED

    def is_executing(self) -> bool:
        return self.state == BaseTask.State.EXECUTING

    def before_run(self) -> None:
        if self.structure is not None:
            EventBus.publish_event(
                StartTaskEvent(
                    task_id=self.id,
                    task_parent_ids=self.parent_ids,
                    task_child_ids=self.child_ids,
                    task_input=self.input,
                    task_output=self.output,
                ),
            )

    def after_run(self) -> None:
        if self.structure is not None:
            EventBus.publish_event(
                FinishTaskEvent(
                    task_id=self.id,
                    task_parent_ids=self.parent_ids,
                    task_child_ids=self.child_ids,
                    task_input=self.input,
                    task_output=self.output,
                ),
            )

    def execute(self) -> Optional[BaseArtifact]:
        try:
            self.state = BaseTask.State.EXECUTING

            self.before_run()

            self.output = self.run()

            self.after_run()
        except Exception as e:
            logger.exception("%s %s\n%s", self.__class__.__name__, self.id, e)

            self.output = ErrorArtifact(str(e), exception=e)
        finally:
            self.state = BaseTask.State.FINISHED

        return self.output

    def can_execute(self) -> bool:
        return self.state == BaseTask.State.PENDING and all(parent.is_finished() for parent in self.parents)

    def reset(self) -> BaseTask:
        self.state = BaseTask.State.PENDING
        self.output = None

        return self

    @abstractmethod
    def run(self) -> BaseArtifact: ...

    @property
    def full_context(self) -> dict[str, Any]:
        if self.structure:
            structure_context = self.structure.context(self)

            structure_context.update(self.context)

            return structure_context
        else:
            return {}

child_ids: list[str] = field(factory=list, kw_only=True) class-attribute instance-attribute

children: list[BaseTask] property

context: dict[str, Any] = field(factory=dict, kw_only=True) class-attribute instance-attribute

full_context: dict[str, Any] property

id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True) class-attribute instance-attribute

input: BaseArtifact abstractmethod property

max_meta_memory_entries: Optional[int] = field(default=20, kw_only=True) class-attribute instance-attribute

meta_memories: list[BaseMetaEntry] property

output: Optional[BaseArtifact] = field(default=None, init=False) class-attribute instance-attribute

parent_ids: list[str] = field(factory=list, kw_only=True) class-attribute instance-attribute

parent_outputs: dict[str, str] property

parents: list[BaseTask] property

parents_output_text: str property

state: State = field(default=State.PENDING, kw_only=True) class-attribute instance-attribute

structure: Optional[Structure] = field(default=None, kw_only=True) class-attribute instance-attribute

State

Bases: Enum

Source code in griptape/tasks/base_task.py

class State(Enum):
    PENDING = 1
    EXECUTING = 2
    FINISHED = 3

EXECUTING = 2 class-attribute instance-attribute

FINISHED = 3 class-attribute instance-attribute

PENDING = 1 class-attribute instance-attribute

__attrs_post_init__()

Source code in griptape/tasks/base_task.py

def __attrs_post_init__(self) -> None:
    if self.structure is not None:
        self.structure.add_task(self)

__lshift__(other)

Source code in griptape/tasks/base_task.py

def __lshift__(self, other: BaseTask) -> BaseTask:
    self.add_parent(other)

    return other

__rshift__(other)

Source code in griptape/tasks/base_task.py

def __rshift__(self, other: BaseTask) -> BaseTask:
    self.add_child(other)

    return other

__str__()

Source code in griptape/tasks/base_task.py

def __str__(self) -> str:
    return str(self.output.value)

add_child(child)

Source code in griptape/tasks/base_task.py

def add_child(self, child: BaseTask) -> BaseTask:
    if child.id not in self.child_ids:
        self.child_ids.append(child.id)

    if self.id not in child.parent_ids:
        child.parent_ids.append(self.id)

    if self.structure is not None:
        self.structure.add_task(child)

    return self

add_children(children)

Source code in griptape/tasks/base_task.py

def add_children(self, children: list[BaseTask]) -> None:
    for child in children:
        self.add_child(child)

add_parent(parent)

Source code in griptape/tasks/base_task.py

def add_parent(self, parent: BaseTask) -> BaseTask:
    if parent.id not in self.parent_ids:
        self.parent_ids.append(parent.id)

    if self.id not in parent.child_ids:
        parent.child_ids.append(self.id)

    if self.structure is not None:
        self.structure.add_task(parent)

    return self

add_parents(parents)

Source code in griptape/tasks/base_task.py

def add_parents(self, parents: list[BaseTask]) -> None:
    for parent in parents:
        self.add_parent(parent)

after_run()

Source code in griptape/tasks/base_task.py

def after_run(self) -> None:
    if self.structure is not None:
        EventBus.publish_event(
            FinishTaskEvent(
                task_id=self.id,
                task_parent_ids=self.parent_ids,
                task_child_ids=self.child_ids,
                task_input=self.input,
                task_output=self.output,
            ),
        )

before_run()

Source code in griptape/tasks/base_task.py

def before_run(self) -> None:
    if self.structure is not None:
        EventBus.publish_event(
            StartTaskEvent(
                task_id=self.id,
                task_parent_ids=self.parent_ids,
                task_child_ids=self.child_ids,
                task_input=self.input,
                task_output=self.output,
            ),
        )

can_execute()

Source code in griptape/tasks/base_task.py

def can_execute(self) -> bool:
    return self.state == BaseTask.State.PENDING and all(parent.is_finished() for parent in self.parents)

execute()

Source code in griptape/tasks/base_task.py

def execute(self) -> Optional[BaseArtifact]:
    try:
        self.state = BaseTask.State.EXECUTING

        self.before_run()

        self.output = self.run()

        self.after_run()
    except Exception as e:
        logger.exception("%s %s\n%s", self.__class__.__name__, self.id, e)

        self.output = ErrorArtifact(str(e), exception=e)
    finally:
        self.state = BaseTask.State.FINISHED

    return self.output

is_executing()

Source code in griptape/tasks/base_task.py

def is_executing(self) -> bool:
    return self.state == BaseTask.State.EXECUTING

is_finished()

Source code in griptape/tasks/base_task.py

def is_finished(self) -> bool:
    return self.state == BaseTask.State.FINISHED

is_pending()

Source code in griptape/tasks/base_task.py

def is_pending(self) -> bool:
    return self.state == BaseTask.State.PENDING

preprocess(structure)

Source code in griptape/tasks/base_task.py

def preprocess(self, structure: Structure) -> BaseTask:
    self.structure = structure

    return self

reset()

Source code in griptape/tasks/base_task.py

def reset(self) -> BaseTask:
    self.state = BaseTask.State.PENDING
    self.output = None

    return self

run() abstractmethod

Source code in griptape/tasks/base_task.py

@abstractmethod
def run(self) -> BaseArtifact: ...

BaseTextInputTask

Bases: RuleMixin, BaseTask, ABC

Source code in griptape/tasks/base_text_input_task.py

@define
class BaseTextInputTask(RuleMixin, BaseTask, ABC):
    DEFAULT_INPUT_TEMPLATE = "{{ args[0] }}"

    _input: str | TextArtifact | Callable[[BaseTask], TextArtifact] = field(
        default=DEFAULT_INPUT_TEMPLATE,
        alias="input",
    )

    @property
    def input(self) -> TextArtifact:
        if isinstance(self._input, TextArtifact):
            return self._input
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            return TextArtifact(J2().render_from_string(self._input, **self.full_context))

    @input.setter
    def input(self, value: str | TextArtifact | Callable[[BaseTask], TextArtifact]) -> None:
        self._input = value

    def before_run(self) -> None:
        super().before_run()

        logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

    def after_run(self) -> None:
        super().after_run()

        logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

DEFAULT_INPUT_TEMPLATE = '{{ args[0] }}' class-attribute instance-attribute

input: TextArtifact property writable

after_run()

Source code in griptape/tasks/base_text_input_task.py

def after_run(self) -> None:
    super().after_run()

    logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

before_run()

Source code in griptape/tasks/base_text_input_task.py

def before_run(self) -> None:
    super().before_run()

    logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

CodeExecutionTask

Bases: BaseTextInputTask

Source code in griptape/tasks/code_execution_task.py

@define
class CodeExecutionTask(BaseTextInputTask):
    run_fn: Callable[[CodeExecutionTask], BaseArtifact] = field(kw_only=True)

    def run(self) -> BaseArtifact:
        return self.run_fn(self)

run_fn: Callable[[CodeExecutionTask], BaseArtifact] = field(kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/code_execution_task.py

def run(self) -> BaseArtifact:
    return self.run_fn(self)

ExtractionTask

Bases: BaseTextInputTask

Source code in griptape/tasks/extraction_task.py

@define
class ExtractionTask(BaseTextInputTask):
    extraction_engine: BaseExtractionEngine = field(kw_only=True)
    args: dict = field(kw_only=True, factory=dict)

    def run(self) -> ListArtifact | ErrorArtifact:
        return self.extraction_engine.extract_artifacts(
            ListArtifact([self.input]), rulesets=self.all_rulesets, **self.args
        )

args: dict = field(kw_only=True, factory=dict) class-attribute instance-attribute

extraction_engine: BaseExtractionEngine = field(kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/extraction_task.py

def run(self) -> ListArtifact | ErrorArtifact:
    return self.extraction_engine.extract_artifacts(
        ListArtifact([self.input]), rulesets=self.all_rulesets, **self.args
    )

ImageQueryTask

Bases: BaseTask

A task that executes a natural language query on one or more input images.

Accepts a text prompt and a list of images as input in one of the following formats: - tuple of (template string, list[ImageArtifact]) - tuple of (TextArtifact, list[ImageArtifact]) - Callable that returns a tuple of (TextArtifact, list[ImageArtifact]).

Attributes:

Name	Type	Description
image_query_engine	ImageQueryEngine	The engine used to execute the query.

Source code in griptape/tasks/image_query_task.py

@define
class ImageQueryTask(BaseTask):
    """A task that executes a natural language query on one or more input images.

    Accepts a text prompt and a list of
    images as input in one of the following formats:
    - tuple of (template string, list[ImageArtifact])
    - tuple of (TextArtifact, list[ImageArtifact])
    - Callable that returns a tuple of (TextArtifact, list[ImageArtifact]).

    Attributes:
        image_query_engine: The engine used to execute the query.
    """

    image_query_engine: ImageQueryEngine = field(default=Factory(lambda: ImageQueryEngine()), kw_only=True)
    _input: (
        tuple[str, list[ImageArtifact]]
        | tuple[TextArtifact, list[ImageArtifact]]
        | Callable[[BaseTask], ListArtifact]
        | ListArtifact
    ) = field(default=None, alias="input")

    @property
    def input(self) -> ListArtifact:
        if isinstance(self._input, ListArtifact):
            return self._input
        elif isinstance(self._input, tuple):
            if isinstance(self._input[0], TextArtifact):
                query_text = self._input[0]
            else:
                query_text = TextArtifact(J2().render_from_string(self._input[0], **self.full_context))

            return ListArtifact([query_text, *self._input[1]])
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            raise ValueError(
                "Input must be a tuple of a TextArtifact and a list of ImageArtifacts or a callable that "
                "returns a tuple of a TextArtifact and a list of ImageArtifacts.",
            )

    @input.setter
    def input(
        self,
        value: (
            tuple[str, list[ImageArtifact]]
            | tuple[TextArtifact, list[ImageArtifact]]
            | Callable[[BaseTask], ListArtifact]
        ),
    ) -> None:
        self._input = value

    def run(self) -> TextArtifact:
        query = self.input.value[0]

        if all(isinstance(artifact, ImageArtifact) for artifact in self.input.value[1:]):
            image_artifacts = [
                image_artifact for image_artifact in self.input.value[1:] if isinstance(image_artifact, ImageArtifact)
            ]
        else:
            raise ValueError("All inputs after the query must be ImageArtifacts.")

        self.output = self.image_query_engine.run(query.value, image_artifacts)

        return self.output

image_query_engine: ImageQueryEngine = field(default=Factory(lambda: ImageQueryEngine()), kw_only=True) class-attribute instance-attribute

input: ListArtifact property writable

run()

Source code in griptape/tasks/image_query_task.py

def run(self) -> TextArtifact:
    query = self.input.value[0]

    if all(isinstance(artifact, ImageArtifact) for artifact in self.input.value[1:]):
        image_artifacts = [
            image_artifact for image_artifact in self.input.value[1:] if isinstance(image_artifact, ImageArtifact)
        ]
    else:
        raise ValueError("All inputs after the query must be ImageArtifacts.")

    self.output = self.image_query_engine.run(query.value, image_artifacts)

    return self.output

InpaintingImageGenerationTask

Bases: BaseImageGenerationTask

A task that modifies a select region within an image using a mask.

Accepts a text prompt, image, and mask as input in one of the following formats: - tuple of (template string, ImageArtifact, ImageArtifact) - tuple of (TextArtifact, ImageArtifact, ImageArtifact) - Callable that returns a tuple of (TextArtifact, ImageArtifact, ImageArtifact).

Attributes:

Name	Type	Description
image_generation_engine	InpaintingImageGenerationEngine	The engine used to generate the image.
negative_rulesets	InpaintingImageGenerationEngine	List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
negative_rules	InpaintingImageGenerationEngine	List of negatively-weighted rules applied to the text prompt, if supported by the driver.
output_dir	InpaintingImageGenerationEngine	If provided, the generated image will be written to disk in output_dir.
output_file	InpaintingImageGenerationEngine	If provided, the generated image will be written to disk as output_file.

Source code in griptape/tasks/inpainting_image_generation_task.py

@define
class InpaintingImageGenerationTask(BaseImageGenerationTask):
    """A task that modifies a select region within an image using a mask.

    Accepts a text prompt, image, and mask as
    input in one of the following formats:
    - tuple of (template string, ImageArtifact, ImageArtifact)
    - tuple of (TextArtifact, ImageArtifact, ImageArtifact)
    - Callable that returns a tuple of (TextArtifact, ImageArtifact, ImageArtifact).

    Attributes:
        image_generation_engine: The engine used to generate the image.
        negative_rulesets: List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
        negative_rules: List of negatively-weighted rules applied to the text prompt, if supported by the driver.
        output_dir: If provided, the generated image will be written to disk in output_dir.
        output_file: If provided, the generated image will be written to disk as output_file.
    """

    image_generation_engine: InpaintingImageGenerationEngine = field(
        default=Factory(lambda: InpaintingImageGenerationEngine()),
        kw_only=True,
    )
    _input: (
        tuple[str | TextArtifact, ImageArtifact, ImageArtifact] | Callable[[BaseTask], ListArtifact] | ListArtifact
    ) = field(default=None, alias="input")

    @property
    def input(self) -> ListArtifact:
        if isinstance(self._input, ListArtifact):
            return self._input
        elif isinstance(self._input, tuple):
            if isinstance(self._input[0], TextArtifact):
                input_text = self._input[0]
            else:
                input_text = TextArtifact(J2().render_from_string(self._input[0], **self.full_context))

            return ListArtifact([input_text, self._input[1], self._input[2]])
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            raise ValueError("Input must be a tuple of (text, image, mask) or a callable that returns such a tuple.")

    @input.setter
    def input(
        self,
        value: tuple[str | TextArtifact, ImageArtifact, ImageArtifact] | Callable[[BaseTask], ListArtifact],
    ) -> None:
        self._input = value

    def run(self) -> ImageArtifact:
        prompt_artifact = self.input[0]

        image_artifact = self.input[1]
        if not isinstance(image_artifact, ImageArtifact):
            raise ValueError("Image must be an ImageArtifact.")

        mask_artifact = self.input[2]
        if not isinstance(mask_artifact, ImageArtifact):
            raise ValueError("Mask must be an ImageArtifact.")

        output_image_artifact = self.image_generation_engine.run(
            prompts=[prompt_artifact.to_text()],
            image=image_artifact,
            mask=mask_artifact,
            rulesets=self.all_rulesets,
            negative_rulesets=self.negative_rulesets,
        )

        if self.output_dir or self.output_file:
            self._write_to_file(output_image_artifact)

        return output_image_artifact

image_generation_engine: InpaintingImageGenerationEngine = field(default=Factory(lambda: InpaintingImageGenerationEngine()), kw_only=True) class-attribute instance-attribute

input: ListArtifact property writable

run()

Source code in griptape/tasks/inpainting_image_generation_task.py

def run(self) -> ImageArtifact:
    prompt_artifact = self.input[0]

    image_artifact = self.input[1]
    if not isinstance(image_artifact, ImageArtifact):
        raise ValueError("Image must be an ImageArtifact.")

    mask_artifact = self.input[2]
    if not isinstance(mask_artifact, ImageArtifact):
        raise ValueError("Mask must be an ImageArtifact.")

    output_image_artifact = self.image_generation_engine.run(
        prompts=[prompt_artifact.to_text()],
        image=image_artifact,
        mask=mask_artifact,
        rulesets=self.all_rulesets,
        negative_rulesets=self.negative_rulesets,
    )

    if self.output_dir or self.output_file:
        self._write_to_file(output_image_artifact)

    return output_image_artifact

OutpaintingImageGenerationTask

Bases: BaseImageGenerationTask

A task that modifies an image outside the bounds of a mask.

Accepts a text prompt, image, and mask as input in one of the following formats: - tuple of (template string, ImageArtifact, ImageArtifact) - tuple of (TextArtifact, ImageArtifact, ImageArtifact) - Callable that returns a tuple of (TextArtifact, ImageArtifact, ImageArtifact).

Attributes:

Name	Type	Description
image_generation_engine	OutpaintingImageGenerationEngine	The engine used to generate the image.
negative_rulesets	OutpaintingImageGenerationEngine	List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
negative_rules	OutpaintingImageGenerationEngine	List of negatively-weighted rules applied to the text prompt, if supported by the driver.
output_dir	OutpaintingImageGenerationEngine	If provided, the generated image will be written to disk in output_dir.
output_file	OutpaintingImageGenerationEngine	If provided, the generated image will be written to disk as output_file.

Source code in griptape/tasks/outpainting_image_generation_task.py

@define
class OutpaintingImageGenerationTask(BaseImageGenerationTask):
    """A task that modifies an image outside the bounds of a mask.

    Accepts a text prompt, image, and mask as
    input in one of the following formats:
    - tuple of (template string, ImageArtifact, ImageArtifact)
    - tuple of (TextArtifact, ImageArtifact, ImageArtifact)
    - Callable that returns a tuple of (TextArtifact, ImageArtifact, ImageArtifact).

    Attributes:
        image_generation_engine: The engine used to generate the image.
        negative_rulesets: List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
        negative_rules: List of negatively-weighted rules applied to the text prompt, if supported by the driver.
        output_dir: If provided, the generated image will be written to disk in output_dir.
        output_file: If provided, the generated image will be written to disk as output_file.
    """

    image_generation_engine: OutpaintingImageGenerationEngine = field(
        default=Factory(lambda: OutpaintingImageGenerationEngine()),
        kw_only=True,
    )
    _input: (
        tuple[str | TextArtifact, ImageArtifact, ImageArtifact] | Callable[[BaseTask], ListArtifact] | ListArtifact
    ) = field(default=None, alias="input")

    @property
    def input(self) -> ListArtifact:
        if isinstance(self._input, ListArtifact):
            return self._input
        elif isinstance(self._input, tuple):
            if isinstance(self._input[0], TextArtifact):
                input_text = self._input[0]
            else:
                input_text = TextArtifact(J2().render_from_string(self._input[0], **self.full_context))

            return ListArtifact([input_text, self._input[1], self._input[2]])
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            raise ValueError("Input must be a tuple of (text, image, mask) or a callable that returns such a tuple.")

    @input.setter
    def input(
        self,
        value: tuple[str | TextArtifact, ImageArtifact, ImageArtifact] | Callable[[BaseTask], ListArtifact],
    ) -> None:
        self._input = value

    def run(self) -> ImageArtifact:
        prompt_artifact = self.input[0]

        image_artifact = self.input[1]
        if not isinstance(image_artifact, ImageArtifact):
            raise ValueError("Image must be an ImageArtifact.")

        mask_artifact = self.input[2]
        if not isinstance(mask_artifact, ImageArtifact):
            raise ValueError("Mask must be an ImageArtifact.")

        output_image_artifact = self.image_generation_engine.run(
            prompts=[prompt_artifact.to_text()],
            image=image_artifact,
            mask=mask_artifact,
            rulesets=self.all_rulesets,
            negative_rulesets=self.negative_rulesets,
        )

        if self.output_dir or self.output_file:
            self._write_to_file(output_image_artifact)

        return output_image_artifact

image_generation_engine: OutpaintingImageGenerationEngine = field(default=Factory(lambda: OutpaintingImageGenerationEngine()), kw_only=True) class-attribute instance-attribute

input: ListArtifact property writable

run()

Source code in griptape/tasks/outpainting_image_generation_task.py

def run(self) -> ImageArtifact:
    prompt_artifact = self.input[0]

    image_artifact = self.input[1]
    if not isinstance(image_artifact, ImageArtifact):
        raise ValueError("Image must be an ImageArtifact.")

    mask_artifact = self.input[2]
    if not isinstance(mask_artifact, ImageArtifact):
        raise ValueError("Mask must be an ImageArtifact.")

    output_image_artifact = self.image_generation_engine.run(
        prompts=[prompt_artifact.to_text()],
        image=image_artifact,
        mask=mask_artifact,
        rulesets=self.all_rulesets,
        negative_rulesets=self.negative_rulesets,
    )

    if self.output_dir or self.output_file:
        self._write_to_file(output_image_artifact)

    return output_image_artifact

PromptImageGenerationTask

Bases: BaseImageGenerationTask

Used to generate an image from a text prompt.

Accepts prompt as input in one of the following formats: - template string - TextArtifact - Callable that returns a TextArtifact.

Attributes:

Name	Type	Description
image_generation_engine	PromptImageGenerationEngine	The engine used to generate the image.
negative_rulesets	PromptImageGenerationEngine	List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
negative_rules	PromptImageGenerationEngine	List of negatively-weighted rules applied to the text prompt, if supported by the driver.
output_dir	PromptImageGenerationEngine	If provided, the generated image will be written to disk in output_dir.
output_file	PromptImageGenerationEngine	If provided, the generated image will be written to disk as output_file.

Source code in griptape/tasks/prompt_image_generation_task.py

@define
class PromptImageGenerationTask(BaseImageGenerationTask):
    """Used to generate an image from a text prompt.

    Accepts prompt as input in one of the following formats:
    - template string
    - TextArtifact
    - Callable that returns a TextArtifact.

    Attributes:
        image_generation_engine: The engine used to generate the image.
        negative_rulesets: List of negatively-weighted rulesets applied to the text prompt, if supported by the driver.
        negative_rules: List of negatively-weighted rules applied to the text prompt, if supported by the driver.
        output_dir: If provided, the generated image will be written to disk in output_dir.
        output_file: If provided, the generated image will be written to disk as output_file.
    """

    DEFAULT_INPUT_TEMPLATE = "{{ args[0] }}"

    _input: str | TextArtifact | Callable[[BaseTask], TextArtifact] = field(
        default=DEFAULT_INPUT_TEMPLATE, alias="input"
    )
    image_generation_engine: PromptImageGenerationEngine = field(
        default=Factory(lambda: PromptImageGenerationEngine()),
        kw_only=True,
    )

    @property
    def input(self) -> TextArtifact:
        if isinstance(self._input, TextArtifact):
            return self._input
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            return TextArtifact(J2().render_from_string(self._input, **self.full_context))

    @input.setter
    def input(self, value: TextArtifact) -> None:
        self._input = value

    def run(self) -> ImageArtifact:
        image_artifact = self.image_generation_engine.run(
            prompts=[self.input.to_text()],
            rulesets=self.all_rulesets,
            negative_rulesets=self.negative_rulesets,
        )

        if self.output_dir or self.output_file:
            self._write_to_file(image_artifact)

        return image_artifact

DEFAULT_INPUT_TEMPLATE = '{{ args[0] }}' class-attribute instance-attribute

image_generation_engine: PromptImageGenerationEngine = field(default=Factory(lambda: PromptImageGenerationEngine()), kw_only=True) class-attribute instance-attribute

input: TextArtifact property writable

run()

Source code in griptape/tasks/prompt_image_generation_task.py

def run(self) -> ImageArtifact:
    image_artifact = self.image_generation_engine.run(
        prompts=[self.input.to_text()],
        rulesets=self.all_rulesets,
        negative_rulesets=self.negative_rulesets,
    )

    if self.output_dir or self.output_file:
        self._write_to_file(image_artifact)

    return image_artifact

PromptTask

Bases: RuleMixin, BaseTask

Source code in griptape/tasks/prompt_task.py

@define
class PromptTask(RuleMixin, BaseTask):
    prompt_driver: BasePromptDriver = field(
        default=Factory(lambda: Defaults.drivers_config.prompt_driver), kw_only=True
    )
    generate_system_template: Callable[[PromptTask], str] = field(
        default=Factory(lambda self: self.default_system_template_generator, takes_self=True),
        kw_only=True,
    )
    _input: str | list | tuple | BaseArtifact | Callable[[BaseTask], BaseArtifact] = field(
        default=lambda task: task.full_context["args"][0] if task.full_context["args"] else TextArtifact(value=""),
        alias="input",
    )

    @property
    def input(self) -> BaseArtifact:
        return self._process_task_input(self._input)

    @input.setter
    def input(self, value: str | list | tuple | BaseArtifact | Callable[[BaseTask], BaseArtifact]) -> None:
        self._input = value

    output: Optional[BaseArtifact] = field(default=None, init=False)

    @property
    def prompt_stack(self) -> PromptStack:
        stack = PromptStack()
        memory = self.structure.conversation_memory

        system_template = self.generate_system_template(self)
        if system_template:
            stack.add_system_message(system_template)

        stack.add_user_message(self.input)

        if self.output:
            stack.add_assistant_message(self.output)

        if memory is not None:
            # insert memory into the stack right before the user messages
            memory.add_to_prompt_stack(self.prompt_driver, stack, 1 if system_template else 0)

        return stack

    def default_system_template_generator(self, _: PromptTask) -> str:
        return J2("tasks/prompt_task/system.j2").render(
            rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
        )

    def before_run(self) -> None:
        super().before_run()

        logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

    def after_run(self) -> None:
        super().after_run()

        logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

    def run(self) -> BaseArtifact:
        message = self.prompt_driver.run(self.prompt_stack)

        return message.to_artifact()

    def _process_task_input(
        self,
        task_input: str | tuple | list | BaseArtifact | Callable[[BaseTask], BaseArtifact],
    ) -> BaseArtifact:
        if isinstance(task_input, TextArtifact):
            task_input.value = J2().render_from_string(task_input.value, **self.full_context)

            return task_input
        elif isinstance(task_input, Callable):
            return self._process_task_input(task_input(self))
        elif isinstance(task_input, ListArtifact):
            return ListArtifact([self._process_task_input(elem) for elem in task_input.value])
        elif isinstance(task_input, BaseArtifact):
            return task_input
        elif isinstance(task_input, (list, tuple)):
            return ListArtifact([self._process_task_input(elem) for elem in task_input])
        else:
            return self._process_task_input(TextArtifact(task_input))

generate_system_template: Callable[[PromptTask], str] = field(default=Factory(lambda self: self.default_system_template_generator, takes_self=True), kw_only=True) class-attribute instance-attribute

input: BaseArtifact property writable

output: Optional[BaseArtifact] = field(default=None, init=False) class-attribute instance-attribute

prompt_driver: BasePromptDriver = field(default=Factory(lambda: Defaults.drivers_config.prompt_driver), kw_only=True) class-attribute instance-attribute

prompt_stack: PromptStack property

after_run()

Source code in griptape/tasks/prompt_task.py

def after_run(self) -> None:
    super().after_run()

    logger.info("%s %s\nOutput: %s", self.__class__.__name__, self.id, self.output.to_text())

before_run()

Source code in griptape/tasks/prompt_task.py

def before_run(self) -> None:
    super().before_run()

    logger.info("%s %s\nInput: %s", self.__class__.__name__, self.id, self.input.to_text())

default_system_template_generator(_)

Source code in griptape/tasks/prompt_task.py

def default_system_template_generator(self, _: PromptTask) -> str:
    return J2("tasks/prompt_task/system.j2").render(
        rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
    )

run()

Source code in griptape/tasks/prompt_task.py

def run(self) -> BaseArtifact:
    message = self.prompt_driver.run(self.prompt_stack)

    return message.to_artifact()

RagTask

Bases: BaseTextInputTask

Source code in griptape/tasks/rag_task.py

@define
class RagTask(BaseTextInputTask):
    rag_engine: RagEngine = field(kw_only=True, default=Factory(lambda: RagEngine()))

    def run(self) -> BaseArtifact:
        outputs = self.rag_engine.process_query(self.input.to_text()).outputs

        if len(outputs) > 0:
            return ListArtifact(outputs)
        else:
            return ErrorArtifact("empty output")

rag_engine: RagEngine = field(kw_only=True, default=Factory(lambda: RagEngine())) class-attribute instance-attribute

run()

Source code in griptape/tasks/rag_task.py

def run(self) -> BaseArtifact:
    outputs = self.rag_engine.process_query(self.input.to_text()).outputs

    if len(outputs) > 0:
        return ListArtifact(outputs)
    else:
        return ErrorArtifact("empty output")

StructureRunTask

Bases: BaseMultiTextInputTask

Task to run a Structure.

Attributes:

Name	Type	Description
driver	BaseStructureRunDriver	Driver to run the Structure.

Source code in griptape/tasks/structure_run_task.py

@define
class StructureRunTask(BaseMultiTextInputTask):
    """Task to run a Structure.

    Attributes:
        driver: Driver to run the Structure.
    """

    driver: BaseStructureRunDriver = field(kw_only=True)

    def run(self) -> BaseArtifact:
        return self.driver.run(*self.input)

driver: BaseStructureRunDriver = field(kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/structure_run_task.py

def run(self) -> BaseArtifact:
    return self.driver.run(*self.input)

TextSummaryTask

Bases: BaseTextInputTask

Source code in griptape/tasks/text_summary_task.py

@define
class TextSummaryTask(BaseTextInputTask):
    summary_engine: BaseSummaryEngine = field(default=Factory(lambda: PromptSummaryEngine()), kw_only=True)

    def run(self) -> TextArtifact:
        return TextArtifact(self.summary_engine.summarize_text(self.input.to_text(), rulesets=self.all_rulesets))

summary_engine: BaseSummaryEngine = field(default=Factory(lambda: PromptSummaryEngine()), kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/text_summary_task.py

def run(self) -> TextArtifact:
    return TextArtifact(self.summary_engine.summarize_text(self.input.to_text(), rulesets=self.all_rulesets))

TextToSpeechTask

Bases: BaseAudioGenerationTask

Source code in griptape/tasks/text_to_speech_task.py

@define
class TextToSpeechTask(BaseAudioGenerationTask):
    DEFAULT_INPUT_TEMPLATE = "{{ args[0] }}"

    _input: str | TextArtifact | Callable[[BaseTask], TextArtifact] = field(default=DEFAULT_INPUT_TEMPLATE)
    text_to_speech_engine: TextToSpeechEngine = field(default=Factory(lambda: TextToSpeechEngine()), kw_only=True)

    @property
    def input(self) -> TextArtifact:
        if isinstance(self._input, TextArtifact):
            return self._input
        elif isinstance(self._input, Callable):
            return self._input(self)
        else:
            return TextArtifact(J2().render_from_string(self._input, **self.full_context))

    @input.setter
    def input(self, value: TextArtifact) -> None:
        self._input = value

    def run(self) -> AudioArtifact:
        audio_artifact = self.text_to_speech_engine.run(prompts=[self.input.to_text()], rulesets=self.all_rulesets)

        if self.output_dir or self.output_file:
            self._write_to_file(audio_artifact)

        return audio_artifact

DEFAULT_INPUT_TEMPLATE = '{{ args[0] }}' class-attribute instance-attribute

input: TextArtifact property writable

text_to_speech_engine: TextToSpeechEngine = field(default=Factory(lambda: TextToSpeechEngine()), kw_only=True) class-attribute instance-attribute

run()

Source code in griptape/tasks/text_to_speech_task.py

def run(self) -> AudioArtifact:
    audio_artifact = self.text_to_speech_engine.run(prompts=[self.input.to_text()], rulesets=self.all_rulesets)

    if self.output_dir or self.output_file:
        self._write_to_file(audio_artifact)

    return audio_artifact

ToolTask

Bases: PromptTask, ActionsSubtaskOriginMixin

Source code in griptape/tasks/tool_task.py

@define
class ToolTask(PromptTask, ActionsSubtaskOriginMixin):
    ACTION_PATTERN = r"(?s)[^{]*({.*})"

    tool: BaseTool = field(kw_only=True)
    subtask: Optional[ActionsSubtask] = field(default=None, kw_only=True)
    task_memory: Optional[TaskMemory] = field(default=None, kw_only=True)

    @property
    def prompt_stack(self) -> PromptStack:
        stack = super().prompt_stack
        stack.tools = [self.tool]

        return stack

    def __attrs_post_init__(self) -> None:
        super().__attrs_post_init__()
        if self.task_memory is not None:
            self.set_default_tools_memory(self.task_memory)

    def preprocess(self, structure: Structure) -> ToolTask:
        super().preprocess(structure)

        if self.task_memory is None and structure.task_memory is not None:
            self.set_default_tools_memory(structure.task_memory)

        return self

    def default_system_template_generator(self, _: PromptTask) -> str:
        return J2("tasks/tool_task/system.j2").render(
            rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
            action_schema=utils.minify_json(json.dumps(self.tool.schema())),
            meta_memory=J2("memory/meta/meta_memory.j2").render(meta_memories=self.meta_memories),
            use_native_tools=self.prompt_driver.use_native_tools,
        )

    def actions_schema(self) -> Schema:
        return self._actions_schema_for_tools([self.tool])

    def run(self) -> BaseArtifact:
        result = self.prompt_driver.run(prompt_stack=self.prompt_stack)

        if self.prompt_driver.use_native_tools:
            subtask_input = result.to_artifact()
        else:
            action_matches = re.findall(self.ACTION_PATTERN, result.to_text(), re.DOTALL)

            if not action_matches:
                return ErrorArtifact("No action found in prompt output.")
            data = action_matches[-1]
            action_dict = json.loads(data)

            action_dict["tag"] = self.tool.name
            subtask_input = J2("tasks/tool_task/subtask.j2").render(action_json=json.dumps(action_dict))

        try:
            subtask = self.add_subtask(ActionsSubtask(subtask_input))

            subtask.before_run()
            subtask.run()
            subtask.after_run()

            if isinstance(subtask.output, ListArtifact):
                first_artifact = subtask.output[0]
                if isinstance(first_artifact, BaseArtifact):
                    self.output = first_artifact
                else:
                    raise ValueError(f"Output is not an Artifact: {type(first_artifact)}")
            else:
                self.output = InfoArtifact("No tool output")
        except Exception as e:
            self.output = ErrorArtifact(f"Error processing tool input: {e}", exception=e)
        return self.output

    def find_tool(self, tool_name: str) -> BaseTool:
        if self.tool.name == tool_name:
            return self.tool
        else:
            raise ValueError(f"Tool with name {tool_name} not found.")

    def find_memory(self, memory_name: str) -> TaskMemory:
        raise NotImplementedError("ToolTask does not support Task Memory.")

    def find_subtask(self, subtask_id: str) -> ActionsSubtask:
        if self.subtask and self.subtask.id == subtask_id:
            return self.subtask
        else:
            raise ValueError(f"Subtask with id {subtask_id} not found.")

    def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
        self.subtask = subtask
        self.subtask.attach_to(self)

        return self.subtask

    def set_default_tools_memory(self, memory: TaskMemory) -> None:
        self.task_memory = memory

        if self.task_memory:
            if self.tool.input_memory is None:
                self.tool.input_memory = [self.task_memory]
            if self.tool.output_memory is None and self.tool.off_prompt:
                self.tool.output_memory = {getattr(a, "name"): [self.task_memory] for a in self.tool.activities()}

ACTION_PATTERN = '(?s)[^{]*({.*})' class-attribute instance-attribute

prompt_stack: PromptStack property

subtask: Optional[ActionsSubtask] = field(default=None, kw_only=True) class-attribute instance-attribute

task_memory: Optional[TaskMemory] = field(default=None, kw_only=True) class-attribute instance-attribute

tool: BaseTool = field(kw_only=True) class-attribute instance-attribute

__attrs_post_init__()

Source code in griptape/tasks/tool_task.py

def __attrs_post_init__(self) -> None:
    super().__attrs_post_init__()
    if self.task_memory is not None:
        self.set_default_tools_memory(self.task_memory)

actions_schema()

Source code in griptape/tasks/tool_task.py

def actions_schema(self) -> Schema:
    return self._actions_schema_for_tools([self.tool])

add_subtask(subtask)

Source code in griptape/tasks/tool_task.py

def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
    self.subtask = subtask
    self.subtask.attach_to(self)

    return self.subtask

default_system_template_generator(_)

Source code in griptape/tasks/tool_task.py

def default_system_template_generator(self, _: PromptTask) -> str:
    return J2("tasks/tool_task/system.j2").render(
        rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
        action_schema=utils.minify_json(json.dumps(self.tool.schema())),
        meta_memory=J2("memory/meta/meta_memory.j2").render(meta_memories=self.meta_memories),
        use_native_tools=self.prompt_driver.use_native_tools,
    )

find_memory(memory_name)

Source code in griptape/tasks/tool_task.py

def find_memory(self, memory_name: str) -> TaskMemory:
    raise NotImplementedError("ToolTask does not support Task Memory.")

find_subtask(subtask_id)

Source code in griptape/tasks/tool_task.py

def find_subtask(self, subtask_id: str) -> ActionsSubtask:
    if self.subtask and self.subtask.id == subtask_id:
        return self.subtask
    else:
        raise ValueError(f"Subtask with id {subtask_id} not found.")

find_tool(tool_name)

Source code in griptape/tasks/tool_task.py

def find_tool(self, tool_name: str) -> BaseTool:
    if self.tool.name == tool_name:
        return self.tool
    else:
        raise ValueError(f"Tool with name {tool_name} not found.")

preprocess(structure)

Source code in griptape/tasks/tool_task.py

def preprocess(self, structure: Structure) -> ToolTask:
    super().preprocess(structure)

    if self.task_memory is None and structure.task_memory is not None:
        self.set_default_tools_memory(structure.task_memory)

    return self

run()

Source code in griptape/tasks/tool_task.py

def run(self) -> BaseArtifact:
    result = self.prompt_driver.run(prompt_stack=self.prompt_stack)

    if self.prompt_driver.use_native_tools:
        subtask_input = result.to_artifact()
    else:
        action_matches = re.findall(self.ACTION_PATTERN, result.to_text(), re.DOTALL)

        if not action_matches:
            return ErrorArtifact("No action found in prompt output.")
        data = action_matches[-1]
        action_dict = json.loads(data)

        action_dict["tag"] = self.tool.name
        subtask_input = J2("tasks/tool_task/subtask.j2").render(action_json=json.dumps(action_dict))

    try:
        subtask = self.add_subtask(ActionsSubtask(subtask_input))

        subtask.before_run()
        subtask.run()
        subtask.after_run()

        if isinstance(subtask.output, ListArtifact):
            first_artifact = subtask.output[0]
            if isinstance(first_artifact, BaseArtifact):
                self.output = first_artifact
            else:
                raise ValueError(f"Output is not an Artifact: {type(first_artifact)}")
        else:
            self.output = InfoArtifact("No tool output")
    except Exception as e:
        self.output = ErrorArtifact(f"Error processing tool input: {e}", exception=e)
    return self.output

set_default_tools_memory(memory)

Source code in griptape/tasks/tool_task.py

def set_default_tools_memory(self, memory: TaskMemory) -> None:
    self.task_memory = memory

    if self.task_memory:
        if self.tool.input_memory is None:
            self.tool.input_memory = [self.task_memory]
        if self.tool.output_memory is None and self.tool.off_prompt:
            self.tool.output_memory = {getattr(a, "name"): [self.task_memory] for a in self.tool.activities()}

ToolkitTask

Bases: PromptTask, ActionsSubtaskOriginMixin

Source code in griptape/tasks/toolkit_task.py

@define
class ToolkitTask(PromptTask, ActionsSubtaskOriginMixin):
    DEFAULT_MAX_STEPS = 20
    # Stop sequence for chain-of-thought in the framework. Using this "token-like" string to make it more unique,
    # so that it doesn't trigger on accident.
    RESPONSE_STOP_SEQUENCE = "<|Response|>"

    tools: list[BaseTool] = field(factory=list, kw_only=True)
    max_subtasks: int = field(default=DEFAULT_MAX_STEPS, kw_only=True)
    task_memory: Optional[TaskMemory] = field(default=None, kw_only=True)
    subtasks: list[ActionsSubtask] = field(factory=list)
    generate_assistant_subtask_template: Callable[[ActionsSubtask], str] = field(
        default=Factory(lambda self: self.default_assistant_subtask_template_generator, takes_self=True),
        kw_only=True,
    )
    generate_user_subtask_template: Callable[[ActionsSubtask], str] = field(
        default=Factory(lambda self: self.default_user_subtask_template_generator, takes_self=True),
        kw_only=True,
    )
    response_stop_sequence: str = field(default=RESPONSE_STOP_SEQUENCE, kw_only=True)

    def __attrs_post_init__(self) -> None:
        super().__attrs_post_init__()
        if self.task_memory:
            self.set_default_tools_memory(self.task_memory)

    @tools.validator  # pyright: ignore[reportAttributeAccessIssue]
    def validate_tools(self, _: Attribute, tools: list[BaseTool]) -> None:
        tool_names = [t.name for t in tools]

        if len(tool_names) > len(set(tool_names)):
            raise ValueError("tools names have to be unique in task")

    @property
    def tool_output_memory(self) -> list[TaskMemory]:
        unique_memory_dict = {}

        for memories in [tool.output_memory for tool in self.tools if tool.output_memory]:
            for memory_list in memories.values():
                for memory in memory_list:
                    if memory.name not in unique_memory_dict:
                        unique_memory_dict[memory.name] = memory

        return list(unique_memory_dict.values())

    @property
    def prompt_stack(self) -> PromptStack:
        stack = PromptStack(tools=self.tools)
        memory = self.structure.conversation_memory

        stack.add_system_message(self.generate_system_template(self))

        stack.add_user_message(self.input)

        if self.output:
            stack.add_assistant_message(self.output.to_text())
        else:
            for s in self.subtasks:
                if self.prompt_driver.use_native_tools:
                    action_calls = [
                        ToolAction(name=action.name, path=action.path, tag=action.tag, input=action.input)
                        for action in s.actions
                    ]
                    action_results = [
                        ToolAction(
                            name=action.name,
                            path=action.path,
                            tag=action.tag,
                            output=action.output if action.output is not None else s.output,
                        )
                        for action in s.actions
                    ]

                    stack.add_assistant_message(
                        ListArtifact(
                            [
                                *([TextArtifact(s.thought)] if s.thought else []),
                                *[ActionArtifact(a) for a in action_calls],
                            ],
                        ),
                    )
                    stack.add_user_message(
                        ListArtifact(
                            [
                                *[ActionArtifact(a) for a in action_results],
                                *([] if s.output else [TextArtifact("Please keep going")]),
                            ],
                        ),
                    )
                else:
                    stack.add_assistant_message(self.generate_assistant_subtask_template(s))
                    stack.add_user_message(self.generate_user_subtask_template(s))

        if memory:
            # inserting at index 1 to place memory right after system prompt
            memory.add_to_prompt_stack(self.prompt_driver, stack, 1)

        return stack