Skip to content

Base multi text input task

BaseMultiTextInputTask

Bases: RuleMixin, BaseTask, ABC

Source code in griptape/tasks/base_multi_text_input_task.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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(self) for callable in self._input if isinstance(callable, 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([input.to_text() for input in self.input])
        self.structure.logger.info(f"{self.__class__.__name__} {self.id}\nInput: {joined_input}")

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

        self.structure.logger.info(f"{self.__class__.__name__} {self.id}\nOutput: {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 
49
50
51
52
def after_run(self) -> None:
    super().after_run()

    self.structure.logger.info(f"{self.__class__.__name__} {self.id}\nOutput: {self.output.to_text()}")

before_run()

Source code in griptape/tasks/base_multi_text_input_task.py 
43
44
45
46
47
def before_run(self) -> None:
    super().before_run()

    joined_input = "\n".join([input.to_text() for input in self.input])
    self.structure.logger.info(f"{self.__class__.__name__} {self.id}\nInput: {joined_input}")