tool#
Bases: BaseTool “)
Attributes#
Name |
Type |
Description |
|---|---|---|
<a href="tool.mdx#griptape.tools.rag.tool.RagTool.description" title="description = field() class-attribute instance-attribute (griptape.tools.rag.tool.RagTool.description)"><code class="language-python">description</code></a> |
str |
LLM-friendly RAG engine description.<br/> |
<a href="tool.mdx#griptape.tools.rag.tool.RagTool.rag_engine" title="rag_engine = field() class-attribute instance-attribute (griptape.tools.rag.tool.RagTool.rag_engine)"><code class="language-python">rag_engine</code></a> |
<a href="../../engines/rag#griptape.engines.rag.RagEngine" title="RagEngine (griptape.engines.rag.RagEngine)"><code class="language-python">RagEngine</code></a> |
RagEngine.<br/> |
Source Code in griptape/tools/rag/tool.py
@define(kw_only=True)
class RagTool(BaseTool):
"""Tool for querying a RAG engine.
Attributes:
description: LLM-friendly RAG engine description.
rag_engine: `RagEngine`.
"""
description: str = field()
rag_engine: RagEngine = field()
@activity(
config={
"description": "{{ _self.description }}",
"schema": Schema({Literal("query", description="A natural language search query"): str}),
},
)
def search(self, params: dict) -> ListArtifact | ErrorArtifact:
query = params["values"]["query"]
try:
artifacts = self.rag_engine.process_query(query).outputs
outputs = []
for artifact in artifacts:
if isinstance(artifact, ListArtifact):
outputs.extend(artifact.value)
else:
outputs.append(artifact)
if len(outputs) > 0:
return ListArtifact(outputs)
return ErrorArtifact("query output is empty")
except Exception as e:
return ErrorArtifact(f"error querying: {e}")
description = field()class-attribute instance-attributerag_engine = field()class-attribute instance-attribute
search(params)#
Source Code in griptape/tools/rag/tool.py
@activity(
config={
"description": "{{ _self.description }}",
"schema": Schema({Literal("query", description="A natural language search query"): str}),
},
)
def search(self, params: dict) -> ListArtifact | ErrorArtifact:
query = params["values"]["query"]
try:
artifacts = self.rag_engine.process_query(query).outputs
outputs = []
for artifact in artifacts:
if isinstance(artifact, ListArtifact):
outputs.extend(artifact.value)
else:
outputs.append(artifact)
if len(outputs) > 0:
return ListArtifact(outputs)
return ErrorArtifact("query output is empty")
except Exception as e:
return ErrorArtifact(f"error querying: {e}")