CLIEventHandler

dbally.audit.CLIEventHandler

CLIEventHandler()

Bases: EventHandler

This handler displays all interactions between LLM and user happening during Collection.ask execution inside the terminal.

Usage

    import dbally
    from dbally.audit.event_handlers.cli_event_handler import CLIEventHandler

    dbally.event_handlers = [CLIEventHandler()]
    my_collection = dbally.create_collection("my_collection", llm)

After using CLIEventHandler, during every Collection.ask execution you will see output similar to the one below:

Source code in src/dbally/audit/event_handlers/cli_event_handler.py

def __init__(self) -> None:
    super().__init__()
    self._console = Console(record=True) if RICH_OUTPUT else None

request_start async

request_start(user_request: RequestStart) -> None

Displays information about event starting to the terminal.

PARAMETER	DESCRIPTION
user_request	Object containing name of collection and asked query TYPE: RequestStart

Source code in src/dbally/audit/event_handlers/cli_event_handler.py

async def request_start(self, user_request: RequestStart) -> None:
    """
    Displays information about event starting to the terminal.

    Args:
        user_request: Object containing name of collection and asked query
    """
    self._print_syntax(f"[orange3 bold]Request starts... \n[orange3 bold]MESSAGE: [grey53]{user_request.question}")
    self._print_syntax("[grey53]\n=======================================")
    self._print_syntax("[grey53]=======================================\n")

event_start async

event_start(event: Event, request_context: None) -> None

Displays information that event has started, then all messages inside the prompt

PARAMETER	DESCRIPTION
event	db-ally event to be logged with all the details. TYPE: Event
request_context	Optional context passed from request_start method TYPE: None

Source code in src/dbally/audit/event_handlers/cli_event_handler.py

async def event_start(self, event: Event, request_context: None) -> None:
    """
    Displays information that event has started, then all messages inside the prompt


    Args:
        event: db-ally event to be logged with all the details.
        request_context: Optional context passed from request_start method
    """

    if isinstance(event, LLMEvent):
        self._print_syntax(
            f"[cyan bold]LLM event starts... \n[cyan bold]LLM EVENT PROMPT TYPE: [grey53]{event.type}"
        )

        if isinstance(event.prompt, tuple):
            for msg in event.prompt:
                self._print_syntax(f"\n[orange3]{msg['role']}")
                self._print_syntax(msg["content"], "text")
        else:
            self._print_syntax(f"{event.prompt}", "text")
    elif isinstance(event, SimilarityEvent):
        self._print_syntax(
            f"[cyan bold]Similarity event starts... \n"
            f"[cyan bold]INPUT: [grey53]{event.input_value}\n"
            f"[cyan bold]STORE: [grey53]{event.store}\n"
            f"[cyan bold]FETCHER: [grey53]{event.fetcher}\n"
        )
    elif isinstance(event, FallbackEvent):
        self._print_syntax(
            f"[grey53]\n=======================================\n"
            "[grey53]=======================================\n"
            f"[orange bold]Fallback event starts \n"
            f"[orange bold]Triggering collection: [grey53]{event.triggering_collection_name}\n"
            f"[orange bold]Triggering view name: [grey53]{event.triggering_view_name}\n"
            f"[orange bold]Error description: [grey53]{event.error_description}\n"
            f"[orange bold]Fallback collection name: [grey53]{event.fallback_collection_name}\n"
            "[grey53]=======================================\n"
            "[grey53]=======================================\n"
        )

event_end async

event_end(event: Optional[Event], request_context: None, event_context: None) -> None

Displays the response from the LLM.

PARAMETER	DESCRIPTION
event	db-ally event to be logged with all the details. TYPE: Optional[Event]
request_context	Optional context passed from request_start method TYPE: None
event_context	Optional context passed from event_start method TYPE: None

Source code in src/dbally/audit/event_handlers/cli_event_handler.py

async def event_end(self, event: Optional[Event], request_context: None, event_context: None) -> None:
    """
    Displays the response from the LLM.

    Args:
        event: db-ally event to be logged with all the details.
        request_context: Optional context passed from request_start method
        event_context: Optional context passed from event_start method
    """
    if isinstance(event, LLMEvent):
        self._print_syntax(f"\n[green bold]RESPONSE: {event.response}")
        self._print_syntax("[grey53]\n=======================================")
        self._print_syntax("[grey53]=======================================\n")
    elif isinstance(event, SimilarityEvent):
        self._print_syntax(f"[green bold]OUTPUT: {event.output_value}")
        self._print_syntax("[grey53]\n=======================================")
        self._print_syntax("[grey53]=======================================\n")

request_end async

request_end(output: RequestEnd, request_context: Optional[dict] = None) -> None

Displays the output of the request, namely the results and the context

PARAMETER	DESCRIPTION
output	The output of the request. TYPE: RequestEnd
request_context	Optional context passed from request_start method TYPE: Optional[dict] DEFAULT: None

Source code in src/dbally/audit/event_handlers/cli_event_handler.py

async def request_end(self, output: RequestEnd, request_context: Optional[dict] = None) -> None:
    """
    Displays the output of the request, namely the `results` and the `context`

    Args:
        output: The output of the request.
        request_context: Optional context passed from request_start method
    """
    if output.result:
        self._print_syntax("[green bold]REQUEST OUTPUT:")
        self._print_syntax(f"Number of rows: {len(output.result.results)}")

        if "sql" in output.result.context:
            self._print_syntax(f"{output.result.context['sql']}", "psql")
    else:
        self._print_syntax("[red bold]No results found")