Bases: SerializableMixin
, ExponentialBackoffMixin
, ABC
Source code in griptape/drivers/image_query/base_image_query_driver.py
| @define
class BaseImageQueryDriver(SerializableMixin, ExponentialBackoffMixin, ABC):
structure: Optional[Structure] = field(default=None, kw_only=True)
max_tokens: int = field(default=256, kw_only=True, metadata={"serializable": True})
def before_run(self, query: str, images: list[ImageArtifact]) -> None:
if self.structure:
self.structure.publish_event(
StartImageQueryEvent(query=query, images_info=[image.to_text() for image in images])
)
def after_run(self, result: str) -> None:
if self.structure:
self.structure.publish_event(FinishImageQueryEvent(result=result))
def query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
for attempt in self.retrying():
with attempt:
self.before_run(query, images)
result = self.try_query(query, images)
self.after_run(result.value)
return result
else:
raise Exception("image query driver failed after all retry attempts")
@abstractmethod
def try_query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
...
|
max_tokens: int = field(default=256, kw_only=True, metadata={'serializable': True})
class-attribute
instance-attribute
structure: Optional[Structure] = field(default=None, kw_only=True)
class-attribute
instance-attribute
after_run(result)
Source code in griptape/drivers/image_query/base_image_query_driver.py
| def after_run(self, result: str) -> None:
if self.structure:
self.structure.publish_event(FinishImageQueryEvent(result=result))
|
before_run(query, images)
Source code in griptape/drivers/image_query/base_image_query_driver.py
| def before_run(self, query: str, images: list[ImageArtifact]) -> None:
if self.structure:
self.structure.publish_event(
StartImageQueryEvent(query=query, images_info=[image.to_text() for image in images])
)
|
query(query, images)
Source code in griptape/drivers/image_query/base_image_query_driver.py
| def query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
for attempt in self.retrying():
with attempt:
self.before_run(query, images)
result = self.try_query(query, images)
self.after_run(result.value)
return result
else:
raise Exception("image query driver failed after all retry attempts")
|
try_query(query, images)
abstractmethod
Source code in griptape/drivers/image_query/base_image_query_driver.py
| @abstractmethod
def try_query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
...
|