rag_task

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")