Skip to content

memory

__all__ = ['TaskMemory'] module-attribute

TaskMemory

Bases: ActivityMixin

Source code in griptape/memory/task/task_memory.py 
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@define
class TaskMemory(ActivityMixin):
    name: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True)
    artifact_storages: dict[type, BaseArtifactStorage] = field(
        default=Factory(
            lambda: {
                TextArtifact: TextArtifactStorage(),
                BlobArtifact: BlobArtifactStorage(),
            }
        ),
        kw_only=True,
    )
    namespace_storage: dict[str, BaseArtifactStorage] = field(factory=dict, kw_only=True)
    namespace_metadata: dict[str, Any] = field(factory=dict, kw_only=True)

    @artifact_storages.validator  # pyright: ignore[reportAttributeAccessIssue]
    def validate_artifact_storages(self, _: Attribute, artifact_storage: dict[type, BaseArtifactStorage]) -> None:
        seen_types = []

        for storage in artifact_storage.values():
            if type(storage) in seen_types:
                raise ValueError("can't have more than memory storage of the same type")

            seen_types.append(type(storage))

    def get_storage_for(self, artifact: BaseArtifact) -> Optional[BaseArtifactStorage]:
        def find_storage(a: BaseArtifact) -> Optional[BaseArtifactStorage]:
            return next((v for k, v in self.artifact_storages.items() if isinstance(a, k)), None)

        if isinstance(artifact, ListArtifact):
            if artifact.has_items():
                return find_storage(artifact.value[0])
            else:
                return None
        else:
            return find_storage(artifact)

    def process_output(
        self,
        tool_activity: Callable,
        subtask: ActionsSubtask,
        output_artifact: BaseArtifact,
    ) -> BaseArtifact:
        from griptape.utils import J2

        tool_name = getattr(getattr(tool_activity, "__self__"), "name")
        activity_name = getattr(tool_activity, "name")
        namespace = output_artifact.name

        if output_artifact:
            result = self.store_artifact(namespace, output_artifact)

            if result:
                return result
            else:
                self.namespace_metadata[namespace] = subtask.actions_to_json()

                output = J2("memory/tool.j2").render(
                    memory_name=self.name,
                    tool_name=tool_name,
                    activity_name=activity_name,
                    artifact_namespace=namespace,
                )

                if subtask.structure and subtask.structure.meta_memory:
                    subtask.structure.meta_memory.add_entry(
                        ActionSubtaskMetaEntry(
                            thought=subtask.thought,
                            actions=subtask.actions_to_json(),
                            answer=output,
                        ),
                    )

                return InfoArtifact(output, name=namespace)
        else:
            return InfoArtifact("tool output is empty")

    def store_artifact(self, namespace: str, artifact: BaseArtifact) -> Optional[BaseArtifact]:
        namespace_storage = self.namespace_storage.get(namespace)
        storage = self.get_storage_for(artifact)

        if not storage:
            return artifact
        elif namespace_storage and namespace_storage != storage:
            return ErrorArtifact("error storing tool output in memory")
        else:
            if storage:
                if isinstance(artifact, ListArtifact):
                    for a in artifact.value:
                        storage.store_artifact(namespace, a)

                    self.namespace_storage[namespace] = storage

                    return None
                elif isinstance(artifact, BaseArtifact):
                    storage.store_artifact(namespace, artifact)

                    self.namespace_storage[namespace] = storage

                    return None
                else:
                    return ErrorArtifact("error storing tool output in memory")
            else:
                return ErrorArtifact("error storing tool output in memory")

    def load_artifacts(self, namespace: str) -> ListArtifact:
        storage = self.namespace_storage.get(namespace)

        if storage:
            return storage.load_artifacts(namespace)
        else:
            return ListArtifact()

    def find_input_memory(self, memory_name: str) -> Optional[TaskMemory]:
        if memory_name == self.name:
            return self
        else:
            return None

artifact_storages: dict[type, BaseArtifactStorage] = field(default=Factory(lambda: {TextArtifact: TextArtifactStorage(), BlobArtifact: BlobArtifactStorage()}), kw_only=True) class-attribute instance-attribute

name: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True) class-attribute instance-attribute

namespace_metadata: dict[str, Any] = field(factory=dict, kw_only=True) class-attribute instance-attribute

namespace_storage: dict[str, BaseArtifactStorage] = field(factory=dict, kw_only=True) class-attribute instance-attribute

find_input_memory(memory_name)

Source code in griptape/memory/task/task_memory.py 
130
131
132
133
134
def find_input_memory(self, memory_name: str) -> Optional[TaskMemory]:
    if memory_name == self.name:
        return self
    else:
        return None

get_storage_for(artifact)

Source code in griptape/memory/task/task_memory.py 
42
43
44
45
46
47
48
49
50
51
52
def get_storage_for(self, artifact: BaseArtifact) -> Optional[BaseArtifactStorage]:
    def find_storage(a: BaseArtifact) -> Optional[BaseArtifactStorage]:
        return next((v for k, v in self.artifact_storages.items() if isinstance(a, k)), None)

    if isinstance(artifact, ListArtifact):
        if artifact.has_items():
            return find_storage(artifact.value[0])
        else:
            return None
    else:
        return find_storage(artifact)

load_artifacts(namespace)

Source code in griptape/memory/task/task_memory.py 
122
123
124
125
126
127
128
def load_artifacts(self, namespace: str) -> ListArtifact:
    storage = self.namespace_storage.get(namespace)

    if storage:
        return storage.load_artifacts(namespace)
    else:
        return ListArtifact()

process_output(tool_activity, subtask, output_artifact)

Source code in griptape/memory/task/task_memory.py 
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
def process_output(
    self,
    tool_activity: Callable,
    subtask: ActionsSubtask,
    output_artifact: BaseArtifact,
) -> BaseArtifact:
    from griptape.utils import J2

    tool_name = getattr(getattr(tool_activity, "__self__"), "name")
    activity_name = getattr(tool_activity, "name")
    namespace = output_artifact.name

    if output_artifact:
        result = self.store_artifact(namespace, output_artifact)

        if result:
            return result
        else:
            self.namespace_metadata[namespace] = subtask.actions_to_json()

            output = J2("memory/tool.j2").render(
                memory_name=self.name,
                tool_name=tool_name,
                activity_name=activity_name,
                artifact_namespace=namespace,
            )

            if subtask.structure and subtask.structure.meta_memory:
                subtask.structure.meta_memory.add_entry(
                    ActionSubtaskMetaEntry(
                        thought=subtask.thought,
                        actions=subtask.actions_to_json(),
                        answer=output,
                    ),
                )

            return InfoArtifact(output, name=namespace)
    else:
        return InfoArtifact("tool output is empty")

store_artifact(namespace, artifact)

Source code in griptape/memory/task/task_memory.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def store_artifact(self, namespace: str, artifact: BaseArtifact) -> Optional[BaseArtifact]:
    namespace_storage = self.namespace_storage.get(namespace)
    storage = self.get_storage_for(artifact)

    if not storage:
        return artifact
    elif namespace_storage and namespace_storage != storage:
        return ErrorArtifact("error storing tool output in memory")
    else:
        if storage:
            if isinstance(artifact, ListArtifact):
                for a in artifact.value:
                    storage.store_artifact(namespace, a)

                self.namespace_storage[namespace] = storage

                return None
            elif isinstance(artifact, BaseArtifact):
                storage.store_artifact(namespace, artifact)

                self.namespace_storage[namespace] = storage

                return None
            else:
                return ErrorArtifact("error storing tool output in memory")
        else:
            return ErrorArtifact("error storing tool output in memory")

validate_artifact_storages(_, artifact_storage)

Source code in griptape/memory/task/task_memory.py 
32
33
34
35
36
37
38
39
40
@artifact_storages.validator  # pyright: ignore[reportAttributeAccessIssue]
def validate_artifact_storages(self, _: Attribute, artifact_storage: dict[type, BaseArtifactStorage]) -> None:
    seen_types = []

    for storage in artifact_storage.values():
        if type(storage) in seen_types:
            raise ValueError("can't have more than memory storage of the same type")

        seen_types.append(type(storage))