Skip to content

Tool task

ToolTask

Bases: PromptTask, ActionsSubtaskOriginMixin

Source code in griptape/tasks/tool_task.py 
 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
@define
class ToolTask(PromptTask, ActionsSubtaskOriginMixin):
    ACTION_PATTERN = r"(?s)[^{]*({.*})"

    tool: BaseTool = field(kw_only=True)
    subtask: Optional[ActionsSubtask] = field(default=None, kw_only=True)
    task_memory: Optional[TaskMemory] = field(default=None, kw_only=True)

    def __attrs_post_init__(self) -> None:
        if self.task_memory is not None:
            self.set_default_tools_memory(self.task_memory)

    def preprocess(self, structure: Structure) -> ToolTask:
        super().preprocess(structure)

        if self.task_memory is None and structure.task_memory is not None:
            self.set_default_tools_memory(structure.task_memory)

        return self

    def default_system_template_generator(self, _: PromptTask) -> str:
        return J2("tasks/tool_task/system.j2").render(
            rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
            action_schema=utils.minify_json(json.dumps(self.tool.schema())),
            meta_memory=J2("memory/meta/meta_memory.j2").render(meta_memories=self.meta_memories),
        )

    def actions_schema(self) -> Schema:
        return self._actions_schema_for_tools([self.tool])

    def run(self) -> BaseArtifact:
        prompt_output = self.prompt_driver.run(prompt_stack=self.prompt_stack).to_text()
        action_matches = re.findall(self.ACTION_PATTERN, prompt_output, re.DOTALL)

        if action_matches:
            try:
                data = action_matches[-1]
                action_dict = json.loads(data)
                action_dict["tag"] = self.tool.name
                subtask_input = J2("tasks/tool_task/subtask.j2").render(action_json=json.dumps(action_dict))
                subtask = self.add_subtask(ActionsSubtask(subtask_input))

                subtask.before_run()
                subtask.run()
                subtask.after_run()

                if isinstance(subtask.output, ListArtifact):
                    self.output = subtask.output[0]
                else:
                    self.output = InfoArtifact("No tool output")
            except Exception as e:
                self.output = ErrorArtifact(f"Error processing tool input: {e}", exception=e)
            return self.output
        else:
            return ErrorArtifact("No action found in prompt output.")

    def find_tool(self, tool_name: str) -> BaseTool:
        if self.tool.name == tool_name:
            return self.tool
        else:
            raise ValueError(f"Tool with name {tool_name} not found.")

    def find_memory(self, memory_name: str) -> TaskMemory:
        raise NotImplementedError("ToolTask does not support Task Memory.")

    def find_subtask(self, subtask_id: str) -> ActionsSubtask:
        if self.subtask and self.subtask.id == subtask_id:
            return self.subtask
        else:
            raise ValueError(f"Subtask with id {subtask_id} not found.")

    def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
        self.subtask = subtask
        self.subtask.attach_to(self)

        return self.subtask

    def set_default_tools_memory(self, memory: TaskMemory) -> None:
        self.task_memory = memory

        if self.task_memory:
            if self.tool.input_memory is None:
                self.tool.input_memory = [self.task_memory]
            if self.tool.output_memory is None and self.tool.off_prompt:
                self.tool.output_memory = {getattr(a, "name"): [self.task_memory] for a in self.tool.activities()}

ACTION_PATTERN = '(?s)[^{]*({.*})' class-attribute instance-attribute

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

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

tool: BaseTool = field(kw_only=True) class-attribute instance-attribute

__attrs_post_init__()

Source code in griptape/tasks/tool_task.py 
28
29
30
def __attrs_post_init__(self) -> None:
    if self.task_memory is not None:
        self.set_default_tools_memory(self.task_memory)

actions_schema()

Source code in griptape/tasks/tool_task.py 
47
48
def actions_schema(self) -> Schema:
    return self._actions_schema_for_tools([self.tool])

add_subtask(subtask)

Source code in griptape/tasks/tool_task.py 
91
92
93
94
95
def add_subtask(self, subtask: ActionsSubtask) -> ActionsSubtask:
    self.subtask = subtask
    self.subtask.attach_to(self)

    return self.subtask

default_system_template_generator(_)

Source code in griptape/tasks/tool_task.py 
40
41
42
43
44
45
def default_system_template_generator(self, _: PromptTask) -> str:
    return J2("tasks/tool_task/system.j2").render(
        rulesets=J2("rulesets/rulesets.j2").render(rulesets=self.all_rulesets),
        action_schema=utils.minify_json(json.dumps(self.tool.schema())),
        meta_memory=J2("memory/meta/meta_memory.j2").render(meta_memories=self.meta_memories),
    )

find_memory(memory_name)

Source code in griptape/tasks/tool_task.py 
82
83
def find_memory(self, memory_name: str) -> TaskMemory:
    raise NotImplementedError("ToolTask does not support Task Memory.")

find_subtask(subtask_id)

Source code in griptape/tasks/tool_task.py 
85
86
87
88
89
def find_subtask(self, subtask_id: str) -> ActionsSubtask:
    if self.subtask and self.subtask.id == subtask_id:
        return self.subtask
    else:
        raise ValueError(f"Subtask with id {subtask_id} not found.")

find_tool(tool_name)

Source code in griptape/tasks/tool_task.py 
76
77
78
79
80
def find_tool(self, tool_name: str) -> BaseTool:
    if self.tool.name == tool_name:
        return self.tool
    else:
        raise ValueError(f"Tool with name {tool_name} not found.")

preprocess(structure)

Source code in griptape/tasks/tool_task.py 
32
33
34
35
36
37
38
def preprocess(self, structure: Structure) -> ToolTask:
    super().preprocess(structure)

    if self.task_memory is None and structure.task_memory is not None:
        self.set_default_tools_memory(structure.task_memory)

    return self

run()

Source code in griptape/tasks/tool_task.py 
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 run(self) -> BaseArtifact:
    prompt_output = self.prompt_driver.run(prompt_stack=self.prompt_stack).to_text()
    action_matches = re.findall(self.ACTION_PATTERN, prompt_output, re.DOTALL)

    if action_matches:
        try:
            data = action_matches[-1]
            action_dict = json.loads(data)
            action_dict["tag"] = self.tool.name
            subtask_input = J2("tasks/tool_task/subtask.j2").render(action_json=json.dumps(action_dict))
            subtask = self.add_subtask(ActionsSubtask(subtask_input))

            subtask.before_run()
            subtask.run()
            subtask.after_run()

            if isinstance(subtask.output, ListArtifact):
                self.output = subtask.output[0]
            else:
                self.output = InfoArtifact("No tool output")
        except Exception as e:
            self.output = ErrorArtifact(f"Error processing tool input: {e}", exception=e)
        return self.output
    else:
        return ErrorArtifact("No action found in prompt output.")

set_default_tools_memory(memory)

Source code in griptape/tasks/tool_task.py 
 97
 98
 99
100
101
102
103
104
def set_default_tools_memory(self, memory: TaskMemory) -> None:
    self.task_memory = memory

    if self.task_memory:
        if self.tool.input_memory is None:
            self.tool.input_memory = [self.task_memory]
        if self.tool.output_memory is None and self.tool.off_prompt:
            self.tool.output_memory = {getattr(a, "name"): [self.task_memory] for a in self.tool.activities()}