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) -> tuple[TextArtifact, ...]:
if all(isinstance(elem, TextArtifact) for elem in self._input):
return self._input # pyright: ignore
elif all(isinstance(elem, Callable) for elem in self._input):
return tuple([elem(self) for elem in self._input]) # pyright: ignore
elif isinstance(self._input, tuple):
return tuple(
[
TextArtifact(J2().render_from_string(input_template, **self.full_context)) # pyright: ignore
for input_template in self._input
]
)
else:
return tuple([TextArtifact(J2().render_from_string(self._input, **self.full_context))])
@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: tuple[TextArtifact, ...]
property
writable
after_run()
Source code in griptape/tasks/base_multi_text_input_task.py
| 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
| 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}")
|