Tasks#

Overview#

A add_tasks(*tasks) is a purpose-built abstraction for the Large Language Model (LLM). Gen AI Builder offers various types of Tasks, each suitable for specific use cases.

Context#

Tasks that take input have a field run(prompt_input) which lets you define the Task objective. Within the run(prompt_input) , you can access the following RAG Context variables:

args : an array of arguments passed to the .run() method.
structure : the structure that the task belongs to.
user defined context variables

Additional RAG Context variables may be added based on the Structure running the task.

from griptape.structures import Agent
from griptape.tasks import PromptTask

agent = Agent()
agent.add_task(
    PromptTask(
        "Respond to the users following question {{ args[0] }} in the language {{preferred_language}} and tone {{tone}}.",
        context={"preferred_language": "ENGLISH", "tone": "PLAYFUL"},
    )
)

agent.run("How do I bake a cake?")

Hooks#

All Tasks implement RunnableMixin which provides on_before_run and on_after_run hooks for the Task lifecycle.

These hooks can be used to perform actions before and after the Task is run. For example, you can mask sensitive information before running the Task, and transform the output after the Task is run.

import json
import re

from griptape.structures import Agent
from griptape.tasks import PromptTask
from griptape.tasks.base_task import BaseTask

SSN_PATTERN = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")

original_input = None

def on_before_run(task: BaseTask) -> None:
    global original_input  # noqa: PLW0603

    original_input = task.input.value

    if isinstance(task, PromptTask):
        task.input = SSN_PATTERN.sub("xxx-xx-xxxx", task.input.value)

def on_after_run(task: BaseTask) -> None:
    if task.output is not None:
        task.output.value = json.dumps(
            {"original_input": original_input, "masked_input": task.input.value, "output": task.output.value}, indent=2
        )

agent = Agent(
    tasks=[
        PromptTask(
            "Respond to this user: {{ args[0] }}",
            on_before_run=on_before_run,
            on_after_run=on_after_run,
        )
    ]
)
agent.run("Hello! My favorite color is blue, and my social security number is 123-45-6789.")

Prompt Task#

For general-purpose interaction with LLMs, use the PromptTask :

from griptape.structures import Agent
from griptape.tasks import PromptTask

agent = Agent()
agent.add_task(
    # take the first argument from the agent `run` method
    PromptTask("Respond to the following request: {{ args[0] }}"),
)

agent.run("Write me a haiku")

Tools#

You can pass in one or more Tools which the LLM will decide to use through Chain of Thought (CoT) reasoning. Because tool execution uses CoT, it is recommended to only use with very capable models.

from griptape.structures import Agent
from griptape.tasks import PromptTask
from griptape.tools import FileManagerTool, PromptSummaryTool, WebScraperTool

agent = Agent()
agent.add_task(
    PromptTask(
        "Load https://www.griptape.ai, summarize it, and store it in a file called griptape.txt",
        tools=[WebScraperTool(off_prompt=True), FileManagerTool(off_prompt=True), PromptSummaryTool(off_prompt=True)],
    ),
)

agent.run()

Reflect On Tool Use#

By default, Gen AI Builder will pass the results of Tool runs back to the LLM for reflection. This enables the LLM to reason about the results and potentially use additional tools.

However, there may be times where you may want the LLM to give you back the results directly, without reflection. You can disable this behavior by setting reflect_on_tool_use to False .

from griptape.artifacts.list_artifact import ListArtifact
from griptape.drivers.web_search.duck_duck_go import DuckDuckGoWebSearchDriver
from griptape.tasks import PromptTask
from griptape.tools import WebScraperTool, WebSearchTool

search_task = PromptTask(
    tools=[WebSearchTool(web_search_driver=DuckDuckGoWebSearchDriver())],
    reflect_on_tool_use=False,
)
search_results = search_task.run("Do two searches, one for vim and one for emacs.")

#  When disabling `reflect_on_tool_use`, the Tasks results will be returned as a ListArtifact.

#  Each item in the ListArtifact will be the result of a single tool execution.

if isinstance(search_results, ListArtifact):
    for result in search_results:
        print(result)

scrape_task = PromptTask(
    tools=[WebScraperTool()],
    reflect_on_tool_use=True,
)

#  If we dont care about the individual results, we can join them back together before passing to the next task.

answer = scrape_task.run(["Compare and contrast vim and emacs: ", search_results.to_text()])

Important

Disabling reflection will prevent the LLM from using one Tool to inform the use of another Tool. Instead, you must coordinate the Tool uses yourself.

Images#

If the model supports it, you can also pass image inputs:

from griptape.loaders import ImageLoader
from griptape.structures import Agent

agent = Agent()

image_artifact = ImageLoader().load("tests/resources/mountain.jpg")

agent.run([image_artifact, "Whats in this image?"])

Tool Task#

Warning

ToolTask is deprecated and will be removed in a future version. Use Prompt Task with reflect_on_tool_use set to False instead.

Another way to use Gen AI Builder Tools , is with a Tool Task . This Task takes in a single Tool which the LLM will use without Chain of Thought (CoT) reasoning. Because this Task does not use CoT, it is better suited for less capable models.

from griptape.structures import Agent
from griptape.tasks import ToolTask
from griptape.tools import CalculatorTool

#  Initialize the agent and add a task

agent = Agent()
agent.add_task(ToolTask(tool=CalculatorTool()))

#  Run the agent with a prompt

agent.run("Give me the answer for 5*4.")

Extraction Task#

To extract information from text, use an ExtractionTask . This Task takes an Extraction Engines , and a set of arguments specific to the Engine.

CSV Extraction#

from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.engines import CsvExtractionEngine
from griptape.structures import Agent
from griptape.tasks import ExtractionTask

#  Instantiate the CSV extraction engine

csv_extraction_engine = CsvExtractionEngine(
    prompt_driver=OpenAiChatPromptDriver(model="gpt-3.5-turbo"), column_names=["Name", "Age", "Address"]
)

#  Define some unstructured data and columns

csv_data = """
Alice, 28, lives in New York.
Bob, 35 lives in California.
Charlie is 40 and lives in Texas.
"""

#  Create an agent and add the ExtractionTask to it

agent = Agent()
agent.add_task(
    ExtractionTask(
        extraction_engine=csv_extraction_engine,
    )
)

#  Run the agent

agent.run(csv_data)

JSON Extraction#

from schema import Schema

from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.engines import JsonExtractionEngine
from griptape.structures import Agent
from griptape.tasks import ExtractionTask

#  Instantiate the json extraction engine

json_extraction_engine = JsonExtractionEngine(
    prompt_driver=OpenAiChatPromptDriver(model="gpt-3.5-turbo"),
    template_schema=Schema({"users": [{"name": str, "age": int, "location": str}]}).json_schema("UserSchema"),
)

#  Define some unstructured data and a schema

json_data = """
Alice (Age 28) lives in New York.
Bob (Age 35) lives in California.
"""

agent = Agent()
agent.add_task(
    ExtractionTask(
        extraction_engine=json_extraction_engine,
    )
)

#  Run the agent

agent.run(json_data)

Text Summary Task#

To summarize a text, use the TextSummaryTask . This Task takes an Summarization Engine , and a set of arguments to the engine.

from griptape.structures import Agent
from griptape.tasks import TextSummaryTask

#  Create a new agent

agent = Agent()

#  Add the TextSummaryTask to the agent

agent.add_task(TextSummaryTask())

#  Run the agent

agent.run(
    "Artificial Intelligence (AI) is a branch of computer science that deals with "
    "creating machines capable of thinking and learning. It encompasses various fields "
    "such as machine learning, neural networks, and deep learning. AI has the potential "
    "to revolutionize many sectors, including healthcare, finance, and transportation. "
    "Our life in this modern age depends largely on computers. It is almost impossible "
    "to think about life without computers. We need computers in everything that we use "
    "in our daily lives. So it becomes very important to make computers intelligent so "
    "that our lives become easy. Artificial Intelligence is the theory and development "
    "of computers, which imitates the human intelligence and senses, such as visual "
    "perception, speech recognition, decision-making, and translation between languages."
    " Artificial Intelligence has brought a revolution in the world of technology. "
)

RAG Task#

To query text, use the RagTask . This task takes a RAG Engine , and a set of arguments specific to the engine.

from griptape.artifacts import TextArtifact
from griptape.drivers.embedding.openai import OpenAiEmbeddingDriver
from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.drivers.vector.local import LocalVectorStoreDriver
from griptape.engines.rag import RagEngine
from griptape.engines.rag.modules import PromptResponseRagModule, VectorStoreRetrievalRagModule
from griptape.engines.rag.stages import ResponseRagStage, RetrievalRagStage
from griptape.structures import Agent
from griptape.tasks import RagTask

#  Initialize Embedding Driver and Vector Store Driver

vector_store_driver = LocalVectorStoreDriver(embedding_driver=OpenAiEmbeddingDriver())

artifacts = [
    TextArtifact("Griptape builds AI-powered applications that connect securely to your enterprise data and APIs."),
    TextArtifact("Griptape Agents provide incredible power and flexibility when working with large language models."),
]
vector_store_driver.upsert_collection({"griptape": artifacts})

#  Instantiate the agent and add RagTask with the RagEngine

agent = Agent()
agent.add_task(
    RagTask(
        "Respond to the following query: {{ args[0] }}",
        rag_engine=RagEngine(
            retrieval_stage=RetrievalRagStage(
                retrieval_modules=[
                    VectorStoreRetrievalRagModule(
                        vector_store_driver=vector_store_driver, query_params={"namespace": "griptape", "top_n": 20}
                    )
                ]
            ),
            response_stage=ResponseRagStage(
                response_modules=[PromptResponseRagModule(prompt_driver=OpenAiChatPromptDriver(model="gpt-4.1"))]
            ),
        ),
    )
)

#  Run the agent with a query string

agent.run("Give me information about Griptape")

Code Execution Task#

To execute an arbitrary Python function, use the CodeExecutionTask . This task takes a python function, and authors can elect to return a custom artifact.

In this example, the generate_title function combines a hero’s name and setting from previous tasks with a random title, returning a TextArtifact that contains the generated fantasy title. The output of this task can then be referenced by subsequent tasks using the parent_outputs templating variable, as shown in the final PromptTask .

import random

from griptape.artifacts import BaseArtifact, TextArtifact
from griptape.structures import Workflow
from griptape.tasks import CodeExecutionTask, PromptTask
from griptape.utils import StructureVisualizer

#  Define a function that will generate a fantasy title by combining inputs

def generate_title(task: CodeExecutionTask) -> BaseArtifact:
    # Make sure this is part of a valid structure
    structure = task.structure
    if structure is None:
        raise ValueError("Task must be part of a structure to run")

    # Retrieve outputs from previous tasks using their task IDs
    hero = structure.find_task("hero").output  # Gets the hero name from first prompt
    setting = structure.find_task("setting").output  # Gets the setting from second prompt

    # List of epic titles to randomly choose from
    titles = ["the Brave", "the Shadow", "the Flame", "the Eternal", "the Cursed"]
    # Combine the hero name, random title, and setting into a full title
    generated_title = f"{hero} {random.choice(titles)} of {setting}"

    # Return the generated title as a TextArtifact that can be used by other tasks
    return TextArtifact(generated_title)

#  Create a new workflow to organize our tasks

workflow = Workflow()

#  Add a sequence of connected tasks to the workflow

workflow.add_tasks(
    # First task: Prompt for a hero name
    # The child_ids parameter connects this to the generate_title task
    PromptTask("Name a fantasy hero (e.g., Eldric)", id="hero", child_ids=["generate_title"]),
    # Second task: Prompt for a setting
    # Also connected to generate_title task
    PromptTask(
        "Describe a mystical setting in a couple sentences (e.g., the Shattered Isles)",
        id="setting",
        child_ids=["generate_title"],
    ),
    # Third task: Execute our custom code to generate the title
    # The on_run parameter specifies which function to execute
    CodeExecutionTask(on_run=generate_title, id="generate_title", child_ids=["story"]),
    # Fourth task: Prompt for a story using the generated title
    # {{ parent_outputs[generate_title] }} references the output from the previous task
    PromptTask(
        "Write a brief, yet heroic tale about {{ parent_outputs[generate_title] }}. ",
        id="story",
    ),
)

#  Visualize the workflow structure (helpful for debugging)

print(StructureVisualizer(workflow).to_url())

#  Execute the entire workflow

workflow.run()

Branch Task#

By default, a Workflow will only run a Task when all the Tasks it depends on have finished.

You can make use of BranchTask in order to tell the Workflow not to run all dependent tasks, but instead to pick and choose one or more paths to go down.

The BranchTask ’s python_runner function can return one of three things:

An InfoArtifact containing the id of the Task to run next.
A ListArtifact of InfoArtifact s containing the id s of the Tasks to run next.
An empty ListArtifact to indicate that no Tasks should be run next.

from griptape.artifacts import InfoArtifact
from griptape.loaders import ImageLoader
from griptape.structures import Workflow
from griptape.tasks import BranchTask, PromptTask

def on_run(task: BranchTask) -> InfoArtifact:
    if "hot dog" in task.input.value:
        return InfoArtifact("hot_dog")
    return InfoArtifact("not_hot_dog")

image_artifact = ImageLoader().load("tests/resources/mountain.png")

workflow = Workflow(
    tasks=[
        PromptTask(["What is in this image?", image_artifact], child_ids=["branch"]),
        BranchTask(
            "{{ parents_output_text }}",
            on_run=on_run,
            id="branch",
            child_ids=["hot_dog", "not_hot_dog"],
        ),
        PromptTask("Tell me about hot dogs", id="hot_dog"),
        PromptTask("Tell me about anything but hotdogs", id="not_hot_dog"),
    ]
)

workflow.run()

Image Generation Tasks#

To generate an image, use one of the following add_tasks(*tasks) . All Image Generation Tasks accept an Image Generation Drivers .

All successful Image Generation Tasks will always output an Image Artifact . Each task can be configured to additionally write the generated image to disk by providing either the output_file or output_dir field. The output_file field supports file names in the current directory (my_image.png ), relative directory prefixes (images/my_image.png ), or absolute paths (/usr/var/my_image.png ). By setting output_dir , the task will generate a file name and place the image in the requested directory.

Prompt Image Generation Task#

The Prompt Image Generation Task generates an image from a text prompt.

from griptape.drivers.image_generation.openai import OpenAiImageGenerationDriver
from griptape.structures import Pipeline
from griptape.tasks import PromptImageGenerationTask

#  Create a driver configured to use OpenAIs DALL-E 3 model.

driver = OpenAiImageGenerationDriver(
    model="dall-e-3",
    quality="hd",
    style="natural",
)

#  Instantiate a pipeline.

pipeline = Pipeline()

#  Add a PromptImageGenerationTask to the pipeline.

pipeline.add_tasks(
    PromptImageGenerationTask(
        input="{{ args[0] }}",
        image_generation_driver=driver,
        output_dir="images/",
    )
)

pipeline.run("An image of a mountain on a summer day")

Variation Image Generation Task#

The Variation Image Generation Task generates an image using an input image and a text prompt. The input image is used as a basis for generating a new image as requested by the text prompt.

from griptape.drivers.image_generation.amazon_bedrock import AmazonBedrockImageGenerationDriver
from griptape.drivers.image_generation_model.bedrock_stable_diffusion import (
    BedrockStableDiffusionImageGenerationModelDriver,
)
from griptape.loaders import ImageLoader
from griptape.structures import Pipeline
from griptape.tasks import VariationImageGenerationTask

#  Create a driver configured to use Stable Diffusion via Bedrock.

driver = AmazonBedrockImageGenerationDriver(
    image_generation_model_driver=BedrockStableDiffusionImageGenerationModelDriver(),
    model="stability.stable-diffusion-xl-v1",
)

#  Load input image artifact.

image_artifact = ImageLoader().load("tests/resources/mountain.png")

#  Instantiate a pipeline.

pipeline = Pipeline()

#  Add a VariationImageGenerationTask to the pipeline.

pipeline.add_task(
    VariationImageGenerationTask(
        input=("{{ args[0] }}", image_artifact),
        image_generation_driver=driver,
        output_dir="images/",
    )
)

pipeline.run("An image of a mountain landscape on a snowy winter day")

Inpainting Image Generation Task#

The Inpainting Image Generation Task generates an image using an input image, a mask image, and a text prompt. The input image will be modified within the bounds of the mask image as requested by the text prompt.

from griptape.drivers.image_generation.amazon_bedrock import AmazonBedrockImageGenerationDriver
from griptape.drivers.image_generation_model.bedrock_stable_diffusion import (
    BedrockStableDiffusionImageGenerationModelDriver,
)
from griptape.loaders import ImageLoader
from griptape.structures import Pipeline
from griptape.tasks import InpaintingImageGenerationTask

#  Create a driver configured to use Stable Diffusion via Bedrock.

driver = AmazonBedrockImageGenerationDriver(
    image_generation_model_driver=BedrockStableDiffusionImageGenerationModelDriver(),
    model="stability.stable-diffusion-xl-v1",
)

#  Load input image artifacts.

image_artifact = ImageLoader().load("tests/resources/mountain.png")

mask_artifact = ImageLoader().load("tests/resources/mountain-mask.png")

#  Instantiate a pipeline.

pipeline = Pipeline()

#  Add an InpaintingImageGenerationTask to the pipeline.

pipeline.add_task(
    InpaintingImageGenerationTask(
        input=("{{ args[0] }}", image_artifact, mask_artifact), image_generation_driver=driver, output_dir="images/"
    )
)

pipeline.run("An image of a castle built into the side of a mountain")

Outpainting Image Generation Task#

The Outpainting Image Generation Task generates an image using an input image, a mask image, and a text prompt. The input image will be modified outside the bounds of a mask image as requested by the text prompt.

from griptape.drivers.image_generation.amazon_bedrock import AmazonBedrockImageGenerationDriver
from griptape.drivers.image_generation_model.bedrock_stable_diffusion import (
    BedrockStableDiffusionImageGenerationModelDriver,
)
from griptape.loaders import ImageLoader
from griptape.structures import Pipeline
from griptape.tasks import OutpaintingImageGenerationTask

#  Create a driver configured to use Stable Diffusion via Bedrock.

driver = AmazonBedrockImageGenerationDriver(
    image_generation_model_driver=BedrockStableDiffusionImageGenerationModelDriver(),
    model="stability.stable-diffusion-xl-v1",
)

#  Load input image artifacts.

image_artifact = ImageLoader().load("tests/resources/mountain.png")

mask_artifact = ImageLoader().load("tests/resources/mountain-mask.png")

#  Instantiate a pipeline.

pipeline = Pipeline()

#  Add an OutpaintingImageGenerationTask to the pipeline.

pipeline.add_task(
    OutpaintingImageGenerationTask(
        input=("{{ args[0] }}", image_artifact, mask_artifact),
        image_generation_driver=driver,
        output_dir="images/",
    )
)

pipeline.run("An image of a mountain shrouded by clouds")

Structure Run Task#

The Structure Run Task runs another Structure with a given input. This Task is useful for orchestrating multiple specialized Structures in a single run. Note that the input to the Task is a tuple of arguments that will be passed to the Structure.

import os

from griptape.drivers.structure_run.local import LocalStructureRunDriver
from griptape.drivers.web_search.google import GoogleWebSearchDriver
from griptape.rules import Rule, Ruleset
from griptape.structures import Agent, Pipeline
from griptape.tasks import StructureRunTask
from griptape.tools import (
    PromptSummaryTool,
    WebScraperTool,
    WebSearchTool,
)

def build_researcher() -> Agent:
    return Agent(
        tools=[
            WebSearchTool(
                web_search_driver=GoogleWebSearchDriver(
                    api_key=os.environ["GOOGLE_API_KEY"],
                    search_id=os.environ["GOOGLE_API_SEARCH_ID"],
                ),
            ),
            WebScraperTool(
                off_prompt=True,
            ),
            PromptSummaryTool(off_prompt=False),
        ],
        rulesets=[
            Ruleset(
                name="Position",
                rules=[
                    Rule(
                        value="Senior Research Analyst",
                    )
                ],
            ),
            Ruleset(
                name="Objective",
                rules=[
                    Rule(
                        value="Uncover cutting-edge developments in AI and data science",
                    )
                ],
            ),
            Ruleset(
                name="Background",
                rules=[
                    Rule(
                        value="""You work at a leading tech think tank.,
                        Your expertise lies in identifying emerging trends.
                        You have a knack for dissecting complex data and presenting actionable insights."""
                    )
                ],
            ),
            Ruleset(
                name="Desired Outcome",
                rules=[
                    Rule(
                        value="Full analysis report in bullet points",
                    )
                ],
            ),
        ],
    )

def build_writer() -> Agent:
    return Agent(
        input="Instructions: {{args[0]}}\nContext: {{args[1]}}",
        rulesets=[
            Ruleset(
                name="Position",
                rules=[
                    Rule(
                        value="Tech Content Strategist",
                    )
                ],
            ),
            Ruleset(
                name="Objective",
                rules=[
                    Rule(
                        value="Craft compelling content on tech advancements",
                    )
                ],
            ),
            Ruleset(
                name="Backstory",
                rules=[
                    Rule(
                        value="""You are a renowned Content Strategist, known for your insightful and engaging articles.
                        You transform complex concepts into compelling narratives."""
                    )
                ],
            ),
            Ruleset(
                name="Desired Outcome",
                rules=[
                    Rule(
                        value="Full blog post of at least 4 paragraphs",
                    )
                ],
            ),
        ],
    )

team = Pipeline(
    tasks=[
        StructureRunTask(
            (
                """Perform a detailed examination of the newest developments in AI as of 2024.
                Pinpoint major trends, breakthroughs, and their implications for various industries.""",
            ),
            structure_run_driver=LocalStructureRunDriver(create_structure=build_researcher),
        ),
        StructureRunTask(
            (
                """Utilize the gathered insights to craft a captivating blog
                article showcasing the key AI innovations.
                Ensure the content is engaging yet straightforward, appealing to a tech-aware readership.
                Keep the tone appealing and use simple language to make it less technical.""",
                "{{parent_output}}",
            ),
            structure_run_driver=LocalStructureRunDriver(create_structure=build_writer),
        ),
    ],
)

team.run()

Assistant Task#

The Assistant Task enables Structures to interact with various “assistant” services using Assistant Drivers .

import os

from griptape.drivers.assistant.griptape_cloud import GriptapeCloudAssistantDriver
from griptape.structures import Pipeline
from griptape.tasks import AssistantTask

pipeline = Pipeline(
    tasks=[
        AssistantTask(
            assistant_driver=GriptapeCloudAssistantDriver(
                assistant_id=os.environ["GT_CLOUD_ASSISTANT_ID"],
            ),
        ),
    ]
)

pipeline.run("Hello!")

Text to Speech Task#

This Task enables Structures to synthesize speech from text using Text to Speech Drivers .

import os

from griptape.drivers.text_to_speech.elevenlabs import ElevenLabsTextToSpeechDriver
from griptape.structures import Pipeline
from griptape.tasks import TextToSpeechTask

driver = ElevenLabsTextToSpeechDriver(
    api_key=os.environ["ELEVEN_LABS_API_KEY"],
    model="eleven_multilingual_v2",
    voice="Matilda",
)

task = TextToSpeechTask(
    text_to_speech_driver=driver,
)

Pipeline(tasks=[task]).run("Generate audio from this text: Hello, world!")

Audio Transcription Task#

This Task enables Structures to transcribe speech from text using Audio Transcription Drivers .

from griptape.drivers.audio_transcription.openai import OpenAiAudioTranscriptionDriver
from griptape.loaders import AudioLoader
from griptape.structures import Pipeline
from griptape.tasks import AudioTranscriptionTask

driver = OpenAiAudioTranscriptionDriver(model="whisper-1")

task = AudioTranscriptionTask(
    input=lambda _: AudioLoader().load("tests/resources/sentences2.wav"),
    audio_transcription_driver=driver,
)

Pipeline(tasks=[task]).run()

Tasks

Contents

Tasks#

Overview#

Context#

Hooks#

Prompt Task#

Tools#

Reflect On Tool Use#

Images#

Tool Task#

Extraction Task#

CSV Extraction#

JSON Extraction#

Text Summary Task#

RAG Task#

Code Execution Task#

Branch Task#

Image Generation Tasks#

Prompt Image Generation Task#

Variation Image Generation Task#

Inpainting Image Generation Task#

Outpainting Image Generation Task#

Structure Run Task#

Assistant Task#

Text to Speech Task#

Audio Transcription Task#