Skip to content

Text query task

TextQueryTask

Bases: BaseTextInputTask

Source code in griptape/tasks/text_query_task.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@define
class TextQueryTask(BaseTextInputTask):
    _query_engine: BaseQueryEngine = field(kw_only=True, default=None, alias="query_engine")
    loader: TextLoader = field(default=Factory(lambda: TextLoader()), kw_only=True)
    namespace: Optional[str] = field(default=None, kw_only=True)
    top_n: Optional[int] = field(default=None, kw_only=True)

    @property
    def query_engine(self) -> BaseQueryEngine:
        if self._query_engine is None:
            if self.structure is not None:
                self._query_engine = VectorQueryEngine(
                    prompt_driver=self.structure.config.global_drivers.prompt_driver,
                    vector_store_driver=self.structure.config.global_drivers.vector_store_driver,
                )
            else:
                raise ValueError("Query Engine is not set.")
        return self._query_engine

    @query_engine.setter
    def query_engine(self, value: BaseQueryEngine) -> None:
        self._query_engine = value

    def run(self) -> TextArtifact:
        return self.query_engine.query(
            self.input.to_text(), namespace=self.namespace, rulesets=self.all_rulesets, top_n=self.top_n
        )

loader: TextLoader = field(default=Factory(lambda: TextLoader()), kw_only=True) class-attribute instance-attribute

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

query_engine: BaseQueryEngine property writable

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

run()

Source code in griptape/tasks/text_query_task.py 
32
33
34
35
def run(self) -> TextArtifact:
    return self.query_engine.query(
        self.input.to_text(), namespace=self.namespace, rulesets=self.all_rulesets, top_n=self.top_n
    )