Mixins

__all__ = ['ActivityMixin', 'ExponentialBackoffMixin', 'ActionsSubtaskOriginMixin', 'RuleMixin', 'ImageArtifactFileOutputMixin', 'SerializableMixin'] module-attribute

ActionsSubtaskOriginMixin

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@define(slots=False)
class ActionsSubtaskOriginMixin:
    @abstractmethod
    def find_tool(self, tool_name: str) -> BaseTool:
        ...

    @abstractmethod
    def find_memory(self, memory_name: str) -> TaskMemory:
        ...

    @abstractmethod
    def find_subtask(self, subtask_id: str) -> ActionsSubtask:
        ...

    @abstractmethod
    def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
        ...

    @abstractmethod
    def actions_schema(self) -> Schema:
        ...

    def _actions_schema_for_tools(self, tools: list[BaseTool]) -> Schema:
        action_schemas = []

        for tool in tools:
            for activity_schema in tool.activity_schemas():
                action_schema = activity_schema.schema
                tag_key = Literal("tag", description="Unique tag name for action execution.")

                action_schema[tag_key] = str

                action_schemas.append(action_schema)

        return Schema(description="JSON schema for an array of actions.", schema=action_schemas)

actions_schema() abstractmethod

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@abstractmethod
def actions_schema(self) -> Schema:
    ...

add_subtask(subtask) abstractmethod

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@abstractmethod
def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
    ...

find_memory(memory_name) abstractmethod

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@abstractmethod
def find_memory(self, memory_name: str) -> TaskMemory:
    ...

find_subtask(subtask_id) abstractmethod

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@abstractmethod
def find_subtask(self, subtask_id: str) -> ActionsSubtask:
    ...

find_tool(tool_name) abstractmethod

Source code in griptape/mixins/actions_subtask_origin_mixin.py

@abstractmethod
def find_tool(self, tool_name: str) -> BaseTool:
    ...

ActivityMixin

Source code in griptape/mixins/activity_mixin.py

@define(slots=False)
class ActivityMixin:
    allowlist: Optional[list[str]] = field(default=None, kw_only=True)
    denylist: Optional[list[str]] = field(default=None, kw_only=True)

    @allowlist.validator  # pyright: ignore
    def validate_allowlist(self, _, allowlist: Optional[list[str]]) -> None:
        if allowlist is None:
            return

        if self.denylist is not None:
            raise ValueError("can't have both allowlist and denylist specified")

        for activity_name in allowlist:
            self._validate_tool_activity(activity_name)

    @denylist.validator  # pyright: ignore
    def validate_denylist(self, _, denylist: Optional[list[str]]) -> None:
        if denylist is None:
            return

        if self.allowlist is not None:
            raise ValueError("can't have both allowlist and denylist specified")

        for activity_name in denylist:
            self._validate_tool_activity(activity_name)

    def enable_activities(self) -> None:
        self.allowlist = None
        self.denylist = None

    def disable_activities(self) -> None:
        self.allowlist = []
        self.denylist = None

    # This method has to remain a method and can't be decorated with @property because
    # of the max depth recursion issue in `inspect.getmembers`.
    def activities(self) -> list[Callable]:
        methods = []

        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            allowlist_condition = self.allowlist is None or name in self.allowlist
            denylist_condition = self.denylist is None or name not in self.denylist

            if getattr(method, "is_activity", False) and allowlist_condition and denylist_condition:
                methods.append(method)

        return methods

    def find_activity(self, name: str) -> Optional[Callable]:
        for activity in self.activities():
            if getattr(activity, "is_activity", False) and getattr(activity, "name") == name:
                return activity

        return None

    def activity_name(self, activity: Callable) -> str:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        else:
            return getattr(activity, "name")

    def activity_description(self, activity: Callable) -> str:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        else:
            return Template(getattr(activity, "config")["description"]).render({"_self": self})

    def activity_schema(self, activity: Callable) -> Optional[Schema]:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        elif getattr(activity, "config")["schema"]:
            full_schema = {
                "values": getattr(activity, "config")["schema"].schema if getattr(activity, "config")["schema"] else {}
            }

            return Schema(full_schema)
        else:
            return None

    def activity_to_input(self, activity: Callable) -> dict:
        if self.activity_schema(activity):
            return {Literal("input"): {"values": getattr(activity, "config")["schema"]}}
        else:
            return {}

    def _validate_tool_activity(self, activity_name):
        tool = self.__class__

        activity = getattr(tool, activity_name, None)

        if not activity or not getattr(activity, "is_activity", False):
            raise ValueError(f"activity {activity_name} is not a valid activity for {tool}")

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

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

activities()

Source code in griptape/mixins/activity_mixin.py

def activities(self) -> list[Callable]:
    methods = []

    for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
        allowlist_condition = self.allowlist is None or name in self.allowlist
        denylist_condition = self.denylist is None or name not in self.denylist

        if getattr(method, "is_activity", False) and allowlist_condition and denylist_condition:
            methods.append(method)

    return methods

activity_description(activity)

Source code in griptape/mixins/activity_mixin.py

def activity_description(self, activity: Callable) -> str:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    else:
        return Template(getattr(activity, "config")["description"]).render({"_self": self})

activity_name(activity)

Source code in griptape/mixins/activity_mixin.py

def activity_name(self, activity: Callable) -> str:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    else:
        return getattr(activity, "name")

activity_schema(activity)

Source code in griptape/mixins/activity_mixin.py

def activity_schema(self, activity: Callable) -> Optional[Schema]:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    elif getattr(activity, "config")["schema"]:
        full_schema = {
            "values": getattr(activity, "config")["schema"].schema if getattr(activity, "config")["schema"] else {}
        }

        return Schema(full_schema)
    else:
        return None

activity_to_input(activity)

Source code in griptape/mixins/activity_mixin.py

def activity_to_input(self, activity: Callable) -> dict:
    if self.activity_schema(activity):
        return {Literal("input"): {"values": getattr(activity, "config")["schema"]}}
    else:
        return {}

disable_activities()

Source code in griptape/mixins/activity_mixin.py

def disable_activities(self) -> None:
    self.allowlist = []
    self.denylist = None

enable_activities()

Source code in griptape/mixins/activity_mixin.py

def enable_activities(self) -> None:
    self.allowlist = None
    self.denylist = None

find_activity(name)

Source code in griptape/mixins/activity_mixin.py

def find_activity(self, name: str) -> Optional[Callable]:
    for activity in self.activities():
        if getattr(activity, "is_activity", False) and getattr(activity, "name") == name:
            return activity

    return None

validate_allowlist(_, allowlist)

Source code in griptape/mixins/activity_mixin.py

@allowlist.validator  # pyright: ignore
def validate_allowlist(self, _, allowlist: Optional[list[str]]) -> None:
    if allowlist is None:
        return

    if self.denylist is not None:
        raise ValueError("can't have both allowlist and denylist specified")

    for activity_name in allowlist:
        self._validate_tool_activity(activity_name)

validate_denylist(_, denylist)

Source code in griptape/mixins/activity_mixin.py

@denylist.validator  # pyright: ignore
def validate_denylist(self, _, denylist: Optional[list[str]]) -> None:
    if denylist is None:
        return

    if self.allowlist is not None:
        raise ValueError("can't have both allowlist and denylist specified")

    for activity_name in denylist:
        self._validate_tool_activity(activity_name)

ExponentialBackoffMixin

Bases: ABC

Source code in griptape/mixins/exponential_backoff_mixin.py

@define(slots=False)
class ExponentialBackoffMixin(ABC):
    min_retry_delay: float = field(default=2, kw_only=True)
    max_retry_delay: float = field(default=10, kw_only=True)
    max_attempts: int = field(default=10, kw_only=True)
    after_hook: Callable = field(default=lambda s: logging.warning(s), kw_only=True)
    ignored_exception_types: Tuple[Type[Exception], ...] = field(factory=tuple, kw_only=True)

    def retrying(self) -> Retrying:
        return Retrying(
            wait=wait_exponential(min=self.min_retry_delay, max=self.max_retry_delay),
            retry=retry_if_not_exception_type(self.ignored_exception_types),
            stop=stop_after_attempt(self.max_attempts),
            reraise=True,
            after=self.after_hook,
        )

after_hook: Callable = field(default=lambda s: logging.warning(s), kw_only=True) class-attribute instance-attribute

ignored_exception_types: Tuple[Type[Exception], ...] = field(factory=tuple, kw_only=True) class-attribute instance-attribute

max_attempts: int = field(default=10, kw_only=True) class-attribute instance-attribute

max_retry_delay: float = field(default=10, kw_only=True) class-attribute instance-attribute

min_retry_delay: float = field(default=2, kw_only=True) class-attribute instance-attribute

retrying()

Source code in griptape/mixins/exponential_backoff_mixin.py

def retrying(self) -> Retrying:
    return Retrying(
        wait=wait_exponential(min=self.min_retry_delay, max=self.max_retry_delay),
        retry=retry_if_not_exception_type(self.ignored_exception_types),
        stop=stop_after_attempt(self.max_attempts),
        reraise=True,
        after=self.after_hook,
    )

ImageArtifactFileOutputMixin

Source code in griptape/mixins/image_artifact_file_output_mixin.py

@define(slots=False)
class ImageArtifactFileOutputMixin:
    output_dir: Optional[str] = field(default=None, kw_only=True)
    output_file: Optional[str] = field(default=None, kw_only=True)

    @output_dir.validator  # pyright: ignore
    def validate_output_dir(self, _, output_dir: str) -> None:
        if not output_dir:
            return

        if self.output_file:
            raise ValueError("Can't have both output_dir and output_file specified.")

    @output_file.validator  # pyright: ignore
    def validate_output_file(self, _, output_file: str) -> None:
        if not output_file:
            return

        if self.output_dir:
            raise ValueError("Can't have both output_dir and output_file specified.")

    def _write_to_file(self, artifact: ImageArtifact) -> None:
        if self.output_file:
            outfile = self.output_file
        elif self.output_dir:
            outfile = os.path.join(self.output_dir, artifact.name)
        else:
            raise ValueError("No output_file or output_dir specified.")

        if os.path.dirname(outfile):
            os.makedirs(os.path.dirname(outfile), exist_ok=True)

        with open(outfile, "wb") as f:
            f.write(artifact.value)

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

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

validate_output_dir(_, output_dir)

Source code in griptape/mixins/image_artifact_file_output_mixin.py

@output_dir.validator  # pyright: ignore
def validate_output_dir(self, _, output_dir: str) -> None:
    if not output_dir:
        return

    if self.output_file:
        raise ValueError("Can't have both output_dir and output_file specified.")

validate_output_file(_, output_file)

Source code in griptape/mixins/image_artifact_file_output_mixin.py

@output_file.validator  # pyright: ignore
def validate_output_file(self, _, output_file: str) -> None:
    if not output_file:
        return

    if self.output_dir:
        raise ValueError("Can't have both output_dir and output_file specified.")

RuleMixin

Source code in griptape/mixins/rule_mixin.py

@define(slots=False)
class RuleMixin:
    DEFAULT_RULESET_NAME = "Default Ruleset"
    ADDITIONAL_RULESET_NAME = "Additional Ruleset"

    rulesets: list[Ruleset] = field(factory=list, kw_only=True)
    rules: list[Rule] = field(factory=list, kw_only=True)
    structure: Optional[Structure] = field(default=None, kw_only=True)

    @rulesets.validator  # pyright: ignore
    def validate_rulesets(self, _, rulesets: list[Ruleset]) -> None:
        if not rulesets:
            return

        if self.rules:
            raise ValueError("Can't have both rulesets and rules specified.")

    @rules.validator  # pyright: ignore
    def validate_rules(self, _, rules: list[Rule]) -> None:
        if not rules:
            return

        if self.rulesets:
            raise ValueError("Can't have both rules and rulesets specified.")

    @property
    def all_rulesets(self) -> list[Ruleset]:
        structure_rulesets = []

        if self.structure:
            if self.structure.rulesets:
                structure_rulesets = self.structure.rulesets
            elif self.structure.rules:
                structure_rulesets = [Ruleset(name=self.DEFAULT_RULESET_NAME, rules=self.structure.rules)]

        task_rulesets = []
        if self.rulesets:
            task_rulesets = self.rulesets
        elif self.rules:
            if structure_rulesets:
                task_ruleset_name = self.ADDITIONAL_RULESET_NAME
            else:
                task_ruleset_name = self.DEFAULT_RULESET_NAME

            task_rulesets = [Ruleset(name=task_ruleset_name, rules=self.rules)]

        return structure_rulesets + task_rulesets

ADDITIONAL_RULESET_NAME = 'Additional Ruleset' class-attribute instance-attribute

DEFAULT_RULESET_NAME = 'Default Ruleset' class-attribute instance-attribute

all_rulesets: list[Ruleset] property

rules: list[Rule] = field(factory=list, kw_only=True) class-attribute instance-attribute

rulesets: list[Ruleset] = field(factory=list, kw_only=True) class-attribute instance-attribute

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

validate_rules(_, rules)

Source code in griptape/mixins/rule_mixin.py

@rules.validator  # pyright: ignore
def validate_rules(self, _, rules: list[Rule]) -> None:
    if not rules:
        return

    if self.rulesets:
        raise ValueError("Can't have both rules and rulesets specified.")

validate_rulesets(_, rulesets)

Source code in griptape/mixins/rule_mixin.py

@rulesets.validator  # pyright: ignore
def validate_rulesets(self, _, rulesets: list[Ruleset]) -> None:
    if not rulesets:
        return

    if self.rules:
        raise ValueError("Can't have both rulesets and rules specified.")

SerializableMixin

Bases: Generic[T]

Source code in griptape/mixins/serializable_mixin.py

@define(slots=False)
class SerializableMixin(Generic[T]):
    type: str = field(
        default=Factory(lambda self: self.__class__.__name__, takes_self=True),
        kw_only=True,
        metadata={"serializable": True},
    )

    @classmethod
    def get_schema(cls: type[T], subclass_name: Optional[str] = None) -> Schema:
        """Generates a Marshmallow schema for the class.

        Args:
            subclass_name: An optional subclass name. Required if the class is abstract.
        """
        if ABC in cls.__bases__:
            if subclass_name is None:
                raise ValueError(f"Type field is required for abstract class: {cls.__name__}")

            subclass_cls = cls._import_cls_rec(cls.__module__, subclass_name)

            schema_class = BaseSchema.from_attrs_cls(subclass_cls)
        else:
            schema_class = BaseSchema.from_attrs_cls(cls)

        return schema_class()

    @classmethod
    def from_dict(cls: type[T], data: dict) -> T:
        return cast(T, cls.get_schema(subclass_name=data["type"] if "type" in data else None).load(data))

    @classmethod
    def from_json(cls: type[T], data: str) -> T:
        return cls.from_dict(json.loads(data))

    def __str__(self) -> str:
        return json.dumps(self.to_dict())

    def to_json(self) -> str:
        return json.dumps(self.to_dict())

    def to_dict(self) -> dict:
        schema = BaseSchema.from_attrs_cls(self.__class__)

        return dict(schema().dump(self))

    @classmethod
    def _import_cls_rec(cls, module_name: str, class_name: str) -> type:
        """Imports a class given a module name and class name.
        Will recursively traverse up the module's path until it finds a
        package that it can import `class_name` from.

        Args:
            module_name: The module name.
            class_name: The class name.

        Returns:
            The imported class if found. Raises `ValueError` if not found.
        """
        try:
            module = import_module(module_name)
            test = getattr(module, class_name, None)
        except ModuleNotFoundError:
            test = None

        if test is None:
            module_dirs = module_name.split(".")[:-1]
            module_name = ".".join(module_dirs)

            if not len(module_dirs):
                raise ValueError(f"Unable to import class: {class_name}")
            return cls._import_cls_rec(module_name, class_name)
        else:
            return test

type: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

__str__()

Source code in griptape/mixins/serializable_mixin.py

def __str__(self) -> str:
    return json.dumps(self.to_dict())

from_dict(data) classmethod

Source code in griptape/mixins/serializable_mixin.py

@classmethod
def from_dict(cls: type[T], data: dict) -> T:
    return cast(T, cls.get_schema(subclass_name=data["type"] if "type" in data else None).load(data))

from_json(data) classmethod

Source code in griptape/mixins/serializable_mixin.py

@classmethod
def from_json(cls: type[T], data: str) -> T:
    return cls.from_dict(json.loads(data))

get_schema(subclass_name=None) classmethod

Generates a Marshmallow schema for the class.

Parameters:

Name	Type	Description	Default
subclass_name	Optional[str]	An optional subclass name. Required if the class is abstract.	None

Source code in griptape/mixins/serializable_mixin.py

@classmethod
def get_schema(cls: type[T], subclass_name: Optional[str] = None) -> Schema:
    """Generates a Marshmallow schema for the class.

    Args:
        subclass_name: An optional subclass name. Required if the class is abstract.
    """
    if ABC in cls.__bases__:
        if subclass_name is None:
            raise ValueError(f"Type field is required for abstract class: {cls.__name__}")

        subclass_cls = cls._import_cls_rec(cls.__module__, subclass_name)

        schema_class = BaseSchema.from_attrs_cls(subclass_cls)
    else:
        schema_class = BaseSchema.from_attrs_cls(cls)

    return schema_class()

to_dict()

Source code in griptape/mixins/serializable_mixin.py

def to_dict(self) -> dict:
    schema = BaseSchema.from_attrs_cls(self.__class__)

    return dict(schema().dump(self))

to_json()

Source code in griptape/mixins/serializable_mixin.py

def to_json(self) -> str:
    return json.dumps(self.to_dict())