Human In The Loop

[collindutter]

Griptape needs a clear story for how to implement real-world human in the loop. Many of the demos online rely on either:

1. Sticking an input() in a Tool call.
2. Using a finite state machine.

Option 1 is not realistic for real-world usage, option 2 is complex and requires significant code changes. Can we provide a middle ground example that uses what we've got today?

[collindutter]

Adding a relatively simple method to EventBus and we can do:

def publish_and_wait(self, event: BaseEvent, response_event_type: type[T]) -> T:
        future = Future()

        def on_event(event: BaseEvent) -> None:
            future.set_result(event)

        EventBus.add_event_listener(EventListener(on_event=on_event, event_types=[response_event_type]))
        EventBus.publish_event(event)
        return future.result()

import logging
import random

from attrs import define
from griptape.artifacts import InfoArtifact
from griptape.events import BaseEvent, EventBus, EventListener
from griptape.structures import Agent
from griptape.tasks import PromptTask
from griptape.tools import BaseTool
from griptape.utils import Chat
from griptape.utils.decorators import activity
from schema import Schema


@define()
class NeedConfirmationEvent(BaseEvent):
    hotel_name: str


@define()
class ProvideConfirmationEvent(BaseEvent):
    confirmed: bool


@define()
class HotelBookingTool(BaseTool):
    @activity(
        {
            "description": "Can be used to book a hotel.",
            "schema": Schema({"hotel_name": str}),
        }
    )
    def book_hotel(self, hotel_name: str) -> InfoArtifact:
        available = self._is_hotel_available(hotel_name)

        if available:
            confirmed = self._confirm_hotel_booking(hotel_name)
            if confirmed:
                result = f"Hotel {hotel_name} has been booked."
            else:
                result = f"Hotel {hotel_name} booking has been canceled."
        else:
            result = f"Hotel {hotel_name} is not available."

        return InfoArtifact(result)

    def _confirm_hotel_booking(self, hotel_name: str) -> bool:
        provide_confirmation_event = EventBus.publish_and_wait(
            NeedConfirmationEvent(hotel_name),
            response_event_type=ProvideConfirmationEvent,
        )

        return provide_confirmation_event.confirmed

    def _is_hotel_available(self, _: str) -> bool:
        return random.random() > 0.1


def human_confirmation(event: NeedConfirmationEvent) -> None:
    response = input(f"Yay or nay {event.hotel_name}: ")
    EventBus.publish_event(
        ProvideConfirmationEvent(confirmed=response.strip().lower() == "yay")
    )


EventBus.add_event_listener(
    EventListener(on_event=human_confirmation, event_types=[NeedConfirmationEvent])
)


agent = Agent(tasks=[PromptTask(tools=[HotelBookingTool()])])
Chat(agent, logger_level=logging.INFO).start()

[collindutter]

Same pattern applied to a decorator for blocking Tool uses.

import logging
import random
from collections.abc import Callable
from functools import wraps
from typing import Any

from attrs import define
from griptape.artifacts import InfoArtifact
from griptape.events import BaseEvent, EventBus, EventListener
from griptape.structures import Agent
from griptape.tasks import PromptTask
from griptape.tools import BaseTool
from griptape.utils import Chat
from griptape.utils.decorators import activity
from schema import Schema


@define()
class ToolRequest(BaseEvent):
    tool_name: str


@define()
class ToolResponse(BaseEvent):
    approved: bool = False
    feedback: str = ""


def requires_hitl(func: Callable) -> Callable:
    @wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        response = EventBus.publish_and_wait(
            ToolRequest(func.__name__),
            response_event_type=ToolResponse,
        )
        if response.approved:
            return func(*args, **kwargs)
        else:
            return InfoArtifact(response.feedback)

    return wrapper


@define()
class HotelBookingTool(BaseTool):
    @activity(
        {
            "description": "Can be used to book a hotel.",
            "schema": Schema({"hotel_name": str}),
        }
    )
    @requires_hitl
    def book_hotel(self, hotel_name: str) -> InfoArtifact:
        available = self._is_hotel_available(hotel_name)

        if available:
            result = f"Hotel {hotel_name} is available."
        else:
            result = f"Hotel {hotel_name} is not available."

        return InfoArtifact(result)

    def _is_hotel_available(self, _: str) -> bool:
        return random.random() > 0.1


def human_confirmation(event: ToolRequest) -> None:
    response = input(f"Yay or nay {event.tool_name}: ")
    parts = [part.strip().lower() for part in response.split("/")]
    approved = parts[0] == "yay"
    feedback = parts[1] if len(parts) > 1 else ""

    EventBus.publish_event(ToolResponse(approved=approved, feedback=feedback))


EventBus.add_event_listener(
    EventListener(on_event=human_confirmation, event_types=[ToolRequest])
)


agent = Agent(tasks=[PromptTask(tools=[HotelBookingTool()])])
Chat(agent, logger_level=logging.INFO).start()

User: Book the hilton
Thinking...
[03/05/25 17:07:01] INFO     PromptTask 2b65313678bb4e39bcbd41a2ec073270
                             Input: Book the hilton
[03/05/25 17:07:03] INFO     Subtask f6a4826da2f14d1091b963a35f5ec068
                             Actions: [
                               {
                                 "tag": "call_9HxNsWtIRLKCxpiHi4NO4fcK",
                                 "name": "HotelBookingTool",
                                 "path": "book_hotel",
                                 "input": {
                                   "values": {
                                     "hotel_name": "Hilton"
                                   }
                                 }
                               }
                             ]
Yay or nay book_hotel: nay/suggest the westin
[03/05/25 17:07:08] INFO     Subtask f6a4826da2f14d1091b963a35f5ec068
                             Response: suggest the westin
[03/05/25 17:07:09] INFO     PromptTask 2b65313678bb4e39bcbd41a2ec073270
                             Output: It seems there was an issue with booking the Hilton. Would you like me to book The
                             Westin instead?
Assistant: It seems there was an issue with booking the Hilton. Would you like me to book The Westin instead?
User: Yeah sure
Thinking...
[03/05/25 17:07:30] INFO     PromptTask 2b65313678bb4e39bcbd41a2ec073270
                             Input: Yeah sure
[03/05/25 17:07:31] INFO     Subtask 0600046ada4f4473a033956d582e3e69
                             Actions: [
                               {
                                 "tag": "call_V8hCn3AIJ4zRIre3nKAFpmYe",
                                 "name": "HotelBookingTool",
                                 "path": "book_hotel",
                                 "input": {
                                   "values": {
                                     "hotel_name": "The Westin"
                                   }
                                 }
                               }
                             ]