Bases: BaseArtifactStorage
Source code in griptape/memory/task/storage/text_artifact_storage.py
| @define
class TextArtifactStorage(BaseArtifactStorage):
query_engine: VectorQueryEngine = field(kw_only=True)
summary_engine: BaseSummaryEngine = field(kw_only=True)
csv_extraction_engine: CsvExtractionEngine = field(kw_only=True)
json_extraction_engine: JsonExtractionEngine = field(kw_only=True)
def can_store(self, artifact: BaseArtifact) -> bool:
return isinstance(artifact, TextArtifact)
def store_artifact(self, namespace: str, artifact: BaseArtifact) -> None:
if isinstance(artifact, TextArtifact):
self.query_engine.upsert_text_artifact(artifact, namespace)
else:
raise ValueError("Artifact must be of instance TextArtifact")
def load_artifacts(self, namespace: str) -> ListArtifact:
return self.query_engine.load_artifacts(namespace)
def summarize(self, namespace: str) -> TextArtifact:
return self.summary_engine.summarize_artifacts(self.load_artifacts(namespace))
def query(self, namespace: str, query: str, metadata: Any = None) -> TextArtifact:
return self.query_engine.query(namespace=namespace, query=query, metadata=str(metadata) if metadata else None)
|
query_engine: VectorQueryEngine = field(kw_only=True)
class-attribute
instance-attribute
summary_engine: BaseSummaryEngine = field(kw_only=True)
class-attribute
instance-attribute
can_store(artifact)
Source code in griptape/memory/task/storage/text_artifact_storage.py
| def can_store(self, artifact: BaseArtifact) -> bool:
return isinstance(artifact, TextArtifact)
|
load_artifacts(namespace)
Source code in griptape/memory/task/storage/text_artifact_storage.py
| def load_artifacts(self, namespace: str) -> ListArtifact:
return self.query_engine.load_artifacts(namespace)
|
query(namespace, query, metadata=None)
Source code in griptape/memory/task/storage/text_artifact_storage.py
| def query(self, namespace: str, query: str, metadata: Any = None) -> TextArtifact:
return self.query_engine.query(namespace=namespace, query=query, metadata=str(metadata) if metadata else None)
|
store_artifact(namespace, artifact)
Source code in griptape/memory/task/storage/text_artifact_storage.py
| def store_artifact(self, namespace: str, artifact: BaseArtifact) -> None:
if isinstance(artifact, TextArtifact):
self.query_engine.upsert_text_artifact(artifact, namespace)
else:
raise ValueError("Artifact must be of instance TextArtifact")
|
summarize(namespace)
Source code in griptape/memory/task/storage/text_artifact_storage.py
| def summarize(self, namespace: str) -> TextArtifact:
return self.summary_engine.summarize_artifacts(self.load_artifacts(namespace))
|