Skip to content

Task memory

TaskMemory

Bases: ActivityMixin

Source code in griptape/memory/task/task_memory.py 
 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
 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
@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(factory=dict, 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
    def validate_artifact_storages(self, _, 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]:
        find_storage = lambda a: 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

    def summarize_namespace(self, namespace: str) -> TextArtifact | InfoArtifact:
        storage = self.namespace_storage.get(namespace)

        if storage:
            return storage.summarize(namespace)
        else:
            return InfoArtifact("Can't find memory content")

    def query_namespace(self, namespace: str, query: str) -> TextArtifact | InfoArtifact:
        storage = self.namespace_storage.get(namespace)

        if storage:
            return storage.query(namespace=namespace, query=query, metadata=self.namespace_metadata.get(namespace))
        else:
            return InfoArtifact("Can't find memory content")

artifact_storages: dict[type, BaseArtifactStorage] = field(factory=dict, 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 
112
113
114
115
116
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 
30
31
32
33
34
35
36
37
38
39
def get_storage_for(self, artifact: BaseArtifact) -> Optional[BaseArtifactStorage]:
    find_storage = lambda a: 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 
104
105
106
107
108
109
110
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 
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
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")

query_namespace(namespace, query)

Source code in griptape/memory/task/task_memory.py 
126
127
128
129
130
131
132
def query_namespace(self, namespace: str, query: str) -> TextArtifact | InfoArtifact:
    storage = self.namespace_storage.get(namespace)

    if storage:
        return storage.query(namespace=namespace, query=query, metadata=self.namespace_metadata.get(namespace))
    else:
        return InfoArtifact("Can't find memory content")

store_artifact(namespace, artifact)

Source code in griptape/memory/task/task_memory.py 
 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
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")

summarize_namespace(namespace)

Source code in griptape/memory/task/task_memory.py 
118
119
120
121
122
123
124
def summarize_namespace(self, namespace: str) -> TextArtifact | InfoArtifact:
    storage = self.namespace_storage.get(namespace)

    if storage:
        return storage.summarize(namespace)
    else:
        return InfoArtifact("Can't find memory content")

validate_artifact_storages(_, artifact_storage)

Source code in griptape/memory/task/task_memory.py 
20
21
22
23
24
25
26
27
28
@artifact_storages.validator  # pyright: ignore
def validate_artifact_storages(self, _, 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))