Skip to content

Mixins

__all__ = ['ActivityMixin', 'ExponentialBackoffMixin', 'ActionSubtaskOriginMixin'] module-attribute

ActionSubtaskOriginMixin

Source code in griptape/griptape/mixins/action_subtask_origin_mixin.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@define(slots=False)
class ActionSubtaskOriginMixin:
    @abstractmethod
    def find_tool(self, tool_name: str) -> Optional[BaseTool]:
        ...

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

    @abstractmethod
    def find_subtask(self, subtask_id: str) -> Optional[ActionSubtask]:
        ...

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

add_subtask(subtask) abstractmethod

Source code in griptape/griptape/mixins/action_subtask_origin_mixin.py 
26
27
28
@abstractmethod
def add_subtask(self, subtask: ActionSubtask) -> ActionSubtask:
    ...

find_memory(memory_name) abstractmethod

Source code in griptape/griptape/mixins/action_subtask_origin_mixin.py 
18
19
20
@abstractmethod
def find_memory(self, memory_name: str) -> Optional[TaskMemory]:
    ...

find_subtask(subtask_id) abstractmethod

Source code in griptape/griptape/mixins/action_subtask_origin_mixin.py 
22
23
24
@abstractmethod
def find_subtask(self, subtask_id: str) -> Optional[ActionSubtask]:
    ...

find_tool(tool_name) abstractmethod

Source code in griptape/griptape/mixins/action_subtask_origin_mixin.py 
14
15
16
@abstractmethod
def find_tool(self, tool_name: str) -> Optional[BaseTool]:
    ...

ActivityMixin

Source code in griptape/griptape/mixins/activity_mixin.py 
 8
 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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@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
    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
    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 method in self.activities():
            if getattr(method, "is_activity", False) and method.name == name:
                return method

        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 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(activity.config["description"]).render({"_self": self})

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

            return Schema(full_schema).json_schema("InputSchema")
        else:
            return None

    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/griptape/mixins/activity_mixin.py 
45
46
47
48
49
50
51
52
53
54
55
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/griptape/mixins/activity_mixin.py 
70
71
72
73
74
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(activity.config["description"]).render({"_self": self})

activity_name(activity)

Source code in griptape/griptape/mixins/activity_mixin.py 
64
65
66
67
68
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 activity.name

activity_schema(activity)

Source code in griptape/griptape/mixins/activity_mixin.py 
76
77
78
79
80
81
82
83
84
def activity_schema(self, activity: Callable) -> Optional[dict]:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    elif activity.config["schema"]:
        full_schema = {"values": activity.config["schema"].schema if activity.config["schema"] else {}}

        return Schema(full_schema).json_schema("InputSchema")
    else:
        return None

disable_activities()

Source code in griptape/griptape/mixins/activity_mixin.py 
39
40
41
def disable_activities(self) -> None:
    self.allowlist = []
    self.denylist = None

enable_activities()

Source code in griptape/griptape/mixins/activity_mixin.py 
35
36
37
def enable_activities(self) -> None:
    self.allowlist = None
    self.denylist = None

find_activity(name)

Source code in griptape/griptape/mixins/activity_mixin.py 
57
58
59
60
61
62
def find_activity(self, name: str) -> Optional[Callable]:
    for method in self.activities():
        if getattr(method, "is_activity", False) and method.name == name:
            return method

    return None

validate_allowlist(_, allowlist)

Source code in griptape/griptape/mixins/activity_mixin.py 
13
14
15
16
17
18
19
20
21
22
@allowlist.validator
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/griptape/mixins/activity_mixin.py 
24
25
26
27
28
29
30
31
32
33
@denylist.validator
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/griptape/mixins/exponential_backoff_mixin.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@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 : 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/griptape/mixins/exponential_backoff_mixin.py 
16
17
18
19
20
21
22
23
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,
    )