Workflows#

Overview#

A Workflow is a non-sequential DAG that can be used for complex concurrent scenarios with tasks having multiple inputs.

You can access the final output of the Workflow by using the Structured Output attribute.

Context#

Workflows have access to the following RAG Context variables in addition to the base context :

task_outputs : dictionary containing mapping of all task IDs to their outputs.
parent_outputs : dictionary containing mapping of parent task IDs to their outputs.
parents_output_text : string containing the concatenated outputs of all parent tasks.
parents : dictionary containing mapping of parent task IDs to their task objects.
children : dictionary containing mapping of child task IDs to their task objects.

Workflow#

Let’s build a simple workflow. Let’s say, we want to write a story in a fantasy world with some unique characters. We could setup a workflow that generates a world based on some keywords. Then we pass the world description to any number of child tasks that create characters. Finally, the last task pulls in information from all parent tasks and writes up a short story.

from griptape.structures import Workflow
from griptape.tasks import PromptTask
from griptape.utils import StructureVisualizer

world_task = PromptTask(
    "Create a fictional world based on the following key words {{ keywords|join(, ) }}",
    context={"keywords": ["fantasy", "ocean", "tidal lock"]},
    id="world",
)

def character_task(task_id: str, character_name: str) -> PromptTask:
    return PromptTask(
        "Based on the following world description create a character named {{ name }}:\n{{ parent_outputs[world] }}",
        context={"name": character_name},
        id=task_id,
        parent_ids=["world"],
    )

scotty_task = character_task("scotty", "Scotty")
annie_task = character_task("annie", "Annie")

story_task = PromptTask(
    "Based on the following description of the world and characters, write a short story:\n{{ parent_outputs[world] }}\n{{ parent_outputs[scotty] }}\n{{ parent_outputs[annie] }}",
    id="story",
    parent_ids=["world", "scotty", "annie"],
)

workflow = Workflow(tasks=[world_task, story_task, scotty_task, annie_task, story_task])

print(StructureVisualizer(workflow).to_url())

workflow.run()

Note that we use the StructureVisualizer to get a visual representation of the workflow. If we visit the printed url, it should look like this:

Declarative vs Imperative Syntax#

The above example showed how to create a workflow using the declarative syntax via the parent_ids init param, but there are a number of declarative and imperative options for you to choose between. There is no functional difference, they merely exist to allow you to structure your code as is most readable for your use case. Possibilities are illustrated below.

Declaratively specify parents (same as above example):

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

workflow = Workflow(
    tasks=[
        PromptTask("Name an animal", id="animal"),
        PromptTask("Describe {{ parent_outputs[animal] }} with an adjective", id="adjective", parent_ids=["animal"]),
        PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal", parent_ids=["adjective"]),
    ],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Declaratively specify children:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

workflow = Workflow(
    tasks=[
        PromptTask("Name an animal", id="animal", child_ids=["adjective"]),
        PromptTask(
            "Describe {{ parent_outputs[animal] }} with an adjective", id="adjective", child_ids=["new-animal"]
        ),
        PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal"),
    ],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Declaratively specifying a mix of parents and children:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

workflow = Workflow(
    tasks=[
        PromptTask("Name an animal", id="animal"),
        PromptTask(
            "Describe {{ parent_outputs[animal] }} with an adjective",
            id="adjective",
            parent_ids=["animal"],
            child_ids=["new-animal"],
        ),
        PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal"),
    ],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Imperatively specify parents:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

animal_task = PromptTask("Name an animal", id="animal")
adjective_task = PromptTask("Describe {{ parent_outputs[animal] }} with an adjective", id="adjective")
new_animal_task = PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal")

adjective_task.add_parent(animal_task)
new_animal_task.add_parent(adjective_task)

workflow = Workflow(
    tasks=[animal_task, adjective_task, new_animal_task],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Imperatively specify children:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

animal_task = PromptTask("Name an animal", id="animal")
adjective_task = PromptTask("Describe {{ parent_outputs[animal] }} with an adjective", id="adjective")
new_animal_task = PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal")

animal_task.add_child(adjective_task)
adjective_task.add_child(new_animal_task)

workflow = Workflow(
    tasks=[animal_task, adjective_task, new_animal_task],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Imperatively specify a mix of parents and children:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

workflow = Workflow(
    rules=[Rule("output a single lowercase word")],
)

animal_task = PromptTask("Name an animal", id="animal", structure=workflow)
adjective_task = PromptTask(
    "Describe {{ parent_outputs[animal] }} with an adjective", id="adjective", structure=workflow
)
new_animal_task = PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal", structure=workflow)

adjective_task.add_parent(animal_task)
adjective_task.add_child(new_animal_task)

workflow.run()

Or even mix imperative and declarative:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

animal_task = PromptTask("Name an animal", id="animal")
adjective_task = PromptTask(
    "Describe {{ parent_outputs[animal] }} with an adjective", id="adjective", parent_ids=["animal"]
)

new_animal_task = PromptTask("Name a {{ parent_outputs[adjective] }} animal", id="new-animal")
new_animal_task.add_parent(adjective_task)

workflow = Workflow(
    tasks=[animal_task, adjective_task, new_animal_task],
    rules=[Rule("output a single lowercase word")],
)

workflow.run()

Insert Parallel Tasks#

Workflow.insert_tasks() provides a convenient way to insert parallel tasks between parents and children.

Imperatively insert parallel tasks between a parent and child:

from griptape.rules import Rule
from griptape.structures import Workflow
from griptape.tasks import PromptTask

workflow = Workflow(
    rules=[Rule("output a single lowercase word")],
)

animal_task = PromptTask("Name an animal", id="animal")
adjective_task = PromptTask("Describe {{ parent_outputs[animal] }} with an adjective", id="adjective")
color_task = PromptTask("Describe {{ parent_outputs[animal] }} with a color", id="color")
new_animal_task = PromptTask("Name an animal described as: \n{{ parents_output_text }}", id="new-animal")

#  The following workflow runs animal_task, then (adjective_task, and color_task)

#  in parallel, then finally new_animal_task.

#
#  In other words, the output of animal_task is passed to both adjective_task and color_task

#  and the outputs of adjective_task and color_task are then passed to new_animal_task.

workflow.add_task(animal_task)
workflow.add_task(new_animal_task)
workflow.insert_tasks(animal_task, [adjective_task, color_task], new_animal_task)

workflow.run()

Bitshift Composition#

Task relationships can also be set up with the Python bitshift operators >> and << . The following statements are all functionally equivalent:

task1 >> task2
task1.add_child(task2)

task2 << task1
task2.add_parent(task1)

task3 >> [task4, task5]
task3.add_children([task4, task5])

When using the bitshift to compose operators, the relationship is set in the direction that the bitshift operator points. For example, task1 >> task2 means that task1 runs first and task2 runs second. Multiple operators can be composed – keep in mind the chain is executed left-to-right and the rightmost object is always returned. For example:

task1 >> task2 >> task3 << task4

is equivalent to:

task1.add_child(task2)
task2.add_child(task3)
task3.add_parent(task4)

Workflows

Contents