Skip to content

base_multi_text_input_task

logger = logging.getLogger(Defaults.logging_config.logger_name) module-attribute

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)