Skip to content

Events

Overview

You can use EventListeners to listen for events during a Structure's execution.

Specific Event Types

You can listen to specific event types:

from griptape.structures import Agent
from griptape.events import (
    BaseEvent,
    StartTaskEvent,
    FinishTaskEvent,
    StartActionsSubtaskEvent,
    FinishActionsSubtaskEvent,
    StartPromptEvent,
    FinishPromptEvent,
    EventListener,
)
from griptape.drivers import LocalEventListenerDriver


def handler(event: BaseEvent):
    print(event.__class__)


agent = Agent(
    event_listeners=[
        EventListener(
            event_types=[
                StartTaskEvent,
                FinishTaskEvent,
                StartActionsSubtaskEvent,
                FinishActionsSubtaskEvent,
                StartPromptEvent,
                FinishPromptEvent,
            ],
            driver=LocalEventListenerDriver(handler=handler),
        )
    ]
)

agent.run("tell me about griptape")
<class 'griptape.events.start_task_event.StartTaskEvent'>
[09/08/23 10:51:16] INFO     PromptTask a20c236d1d86480fb14ae976e6cf8983
                             Input: tell me about griptape
<class 'griptape.events.start_prompt_event.StartPromptEvent'>
<class 'griptape.events.finish_prompt_event.FinishPromptEvent'>
[09/08/23 10:51:27] INFO     PromptTask a20c236d1d86480fb14ae976e6cf8983
                             Output: Griptape is a gritty, sandpaper-like material that is applied to the top of a skateboard deck. It
                             provides traction and grip, allowing skateboarders to keep their feet firmly on the board and perform tricks more
                             easily. Griptape comes in different colors and designs, but the most common type is black. It's adhesive on one
                             side so it can stick to the skateboard. Over time, griptape can wear down and need to be replaced to maintain
                             optimal performance. It's an essential component for skateboarding and other similar sports.
<class 'griptape.events.finish_task_event.FinishTaskEvent'>

All Event Types

Or listen to all events:

from griptape.structures import Agent
from griptape.events import BaseEvent, EventListener
from griptape.drivers import LocalEventListenerDriver


def handler1(event: BaseEvent):
    print("Handler 1", event.__class__)


def handler2(event: BaseEvent):
    print("Handler 2", event.__class__)


agent = Agent(
    event_listeners=[
        EventListener(driver=LocalEventListenerDriver(handler=handler1)),
        EventListener(driver=LocalEventListenerDriver(handler=handler1)),
    ]
)

agent.run("tell me about griptape")
Handler 1 <class 'griptape.events.start_structure_run_event.StartStructureRunEvent'>
Handler 2 <class 'griptape.events.start_structure_run_event.StartStructureRunEvent'>
Handler 1 <class 'griptape.events.start_task_event.StartTaskEvent'>
Handler 2 <class 'griptape.events.start_task_event.StartTaskEvent'>
[10/26/23 11:49:29] INFO     PromptTask 20e3ef1f8856453ebabc84863ac36784
                             Input: tell me about griptape
Handler 1 <class 'griptape.events.start_prompt_event.StartPromptEvent'>
Handler 2 <class 'griptape.events.start_prompt_event.StartPromptEvent'>
Handler 1 <class 'griptape.events.finish_prompt_event.FinishPromptEvent'>
Handler 2 <class 'griptape.events.finish_prompt_event.FinishPromptEvent'>
[10/26/23 11:49:55] INFO     PromptTask 20e3ef1f8856453ebabc84863ac36784
                             Output: Griptape is a gritty, sandpaper-like material that is applied to the top of
                             a skateboard, longboard, or scooter deck to provide traction between the rider's
                             feet and the deck. It is an essential component for performing tricks and
                             maintaining control of the board or scooter.

                             Griptape is typically black, but it comes in a variety of colors and designs. It is
                             made by embedding an abrasive material (similar to sand) into a tough,
                             weather-resistant adhesive. The adhesive side is then applied to the deck, while
                             the abrasive side faces up to provide grip.

                             The grip provided by the griptape allows riders to keep their footing on the board,
                             especially during tricks where the board is flipped or spun. It also helps riders
                             control the board better during downhill rides or sharp turns.

                             Over time, griptape can wear down and lose its effectiveness, at which point it can
                             be removed and replaced. It's an essential part of skateboarding equipment and
                             plays a significant role in the sport's safety and performance.
Handler 1 <class 'griptape.events.finish_task_event.FinishTaskEvent'>
Handler 2 <class 'griptape.events.finish_task_event.FinishTaskEvent'>
Handler 1 <class 'griptape.events.finish_structure_run_event.FinishStructureRunEvent'>
Handler 2 <class 'griptape.events.finish_structure_run_event.FinishStructureRunEvent'>

Streaming

You can use the CompletionChunkEvent to stream the completion results from Prompt Drivers.

from griptape.events import CompletionChunkEvent, EventListener
from griptape.tasks import ToolkitTask
from griptape.structures import Pipeline
from griptape.tools import WebScraper, TaskMemoryClient
from griptape.drivers import LocalEventListenerDriver


pipeline = Pipeline(
    event_listeners=[
        EventListener(
            driver=LocalEventListenerDriver(handler=lambda e: print(e.token, end="", flush=True)),
            event_types=[CompletionChunkEvent],
        )
    ]
)

pipeline.add_tasks(
    ToolkitTask(
        "Based on https://griptape.ai, tell me what griptape is.",
        tools=[WebScraper(), TaskMemoryClient(off_prompt=False)],
    )
)

pipeline.run()

You can also use the Stream utility to automatically wrap CompletionChunkEvents in a Python iterator.

from griptape.utils import Stream
from griptape.tasks import ToolkitTask
from griptape.structures import Pipeline
from griptape.tools import WebScraper


pipeline = Pipeline()
pipeline.config.global_drivers.prompt_driver.stream = True
pipeline.add_tasks(ToolkitTask("Based on https://griptape.ai, tell me what griptape is.", tools=[WebScraper()]))

for artifact in Stream(pipeline).run():
    print(artifact.value, end="", flush=True),

Counting Tokens

To count tokens, you can use Event Listeners and the TokenCounter util:

from griptape import utils
from griptape.events import BaseEvent, StartPromptEvent, FinishPromptEvent, EventListener
from griptape.structures import Agent
from griptape.drivers import LocalEventListenerDriver


token_counter = utils.TokenCounter()


def count_tokens(e: BaseEvent):
    if isinstance(e, StartPromptEvent) or isinstance(e, FinishPromptEvent):
        token_counter.add_tokens(e.token_count)


agent = Agent(
    event_listeners=[
        EventListener(
            driver=LocalEventListenerDriver(handler=lambda e: count_tokens(e)),
            event_types=[StartPromptEvent, FinishPromptEvent],
        )
    ]
)

agent.run("tell me about large language models")

print(f"total tokens: {token_counter.tokens}")
[09/25/23 16:32:41] INFO     PromptTask c93569eb1d264675b52bef184b269621
                             Input: tell me about large language models
[09/25/23 16:33:01] INFO     PromptTask c93569eb1d264675b52bef184b269621
                             Output: Large language models are a type of artificial intelligence model that are trained on
                             a vast amount of text data. They are designed to generate human-like text based on the input
                             they are given. These models can answer questions, write essays, summarize texts, translate
                             languages, and even generate creative content like poetry or stories.

                             One of the most well-known large language models is GPT-3, developed by OpenAI. It has 175
                             billion machine learning parameters and was trained on hundreds of gigabytes of text.

                             These models work by predicting the probability of a word given the previous words used in
                             the text. They don't understand text in the way humans do, but they can generate coherent and
                             contextually relevant sentences by learning patterns in the data they were trained on.

                             However, they also have limitations. For instance, they can sometimes generate incorrect or
                             nonsensical responses. They can also be sensitive to slight changes in input phrasing, and
                             they don't have the ability to fact-check information or access real-time data, so they can't
                             provide up-to-date information or verify the truthfulness of their outputs. They also don't
                             have a sense of ethics or morality, so they rely on guidelines and safety measures put in
                             place by their developers.
total tokens: 273

Inspecting Payloads

You can use the StartPromptEvent to inspect the Prompt Stack and final prompt string before it is sent to the LLM.

from griptape.structures import Agent
from griptape.events import BaseEvent, StartPromptEvent, EventListener
from griptape.drivers import LocalEventListenerDriver


def handler(event: BaseEvent):
    if isinstance(event, StartPromptEvent):
        print("Prompt Stack Inputs:")
        for input in event.prompt_stack.inputs:
            print(f"{input.role}: {input.content}")
        print("Final Prompt String:")
        print(event.prompt)


agent = Agent(
    event_listeners=[EventListener(driver=LocalEventListenerDriver(handler=handler), event_types=[StartPromptEvent])]
)

agent.run("Write me a poem.")
...
Prompt Stack Inputs:
system:
user: Write me a poem.
Final Prompt String:


User: Write me a poem.

Assistant:
...