LlamaIndex Workflows and human intervention or human-in-the-loop

[anirbanbasu]

Is there a recommended or standard way of implementing human-in-the-loop with LlamaIndex Workflows? I guess that stepping through the workflow with a mechanism to change the context that the workflow (including nested ones) see is one way of going about it. However, stepping through the workflow is not the same as planned (or, perhaps even dynamic) intervention.

In comparison, LangGraph has human-in-the-loop. Is this a feature that LlamaIndex Workflows will support in the future?

[dosubot[bot]]

I found a similar open issue that might be relevant to your question: [Feature Request]: Human-In-The-Loop Multistep Query, which is closed and was last updated on February 06, 2024 ^[1].

Currently, LlamaIndex Workflows support dynamic intervention through event handling, allowing events to be sent to specific steps or broadcasted to all steps, enabling dynamic adjustments and interventions during the workflow execution. The Context object in LlamaIndex Workflows can store data across iterations and multiple workflow runs, offering both global and private data storage, which can be dynamically altered ^[2][3].

For implementing human-in-the-loop mechanisms, there isn't specific documentation or examples available in the repository. However, you can create custom event handlers to manage dynamic interventions. Here is an example of how to create a custom event handler:

from typing import Dict
from llama_index.core.instrumentation.event_handlers import BaseEventHandler
from llama_index.packs.diff_private_simple_dataset.events import (
    SyntheticExampleStartEvent,
    SyntheticExampleEndEvent,
    LLMEmptyResponseEvent,
    EmptyIntersectionEvent,
)
import json

class TooManyProblemsEncounteredError(Exception):
    pass

class DiffPrivacyEventHandler(BaseEventHandler):
    num_splits: int
    t_max: int
    synthetic_example_starts: int = 0
    synthetic_example_ends: int = 0
    llm_empty_responses: int = 0
    empty_intersections: int = 0
    critical_threshold: int = 0.025  # ~2.5% error rate with OpenAI API calls

    @classmethod
    def class_name(cls) -> str:
        """Class name."""
        return "MyEventHandler"

    def compute_approximate_error_rate(self):
        """Returns an approximate error rate."""
        return (self.llm_empty_responses + self.empty_intersections) / (
            self.synthetic_example_starts * self.t_max * (self.num_splits + 1)
        )

    def handle(self, event) -> None:
        """Logic for handling event."""

        if isinstance(event, SyntheticExampleStartEvent):
            self.synthetic_example_starts += 1
        elif isinstance(event, SyntheticExampleEndEvent):
            self.synthetic_example_ends += 1
        elif isinstance(event, LLMEmptyResponseEvent):
            self.llm_empty_responses += 1

            with open("error_report.json", "w") as f:
                json.dump(self.dict(), f)

            if self.compute_approximate_error_rate() > self.critical_threshold:
                raise TooManyProblemsEncounteredError(
                    "There were too many errors encountered."
                )
        elif isinstance(event, EmptyIntersectionEvent):
            self.empty_intersections += 1

            with open("error_report.json", "w") as f:
                json.dump(self.dict(), f)

            if self.compute_approximate_error_rate() > self.critical_threshold:
                raise TooManyProblemsEncounteredError(
                    "There were too many errors encountered."
                )

For more detailed examples and documentation, you can refer to the LlamaIndex documentation on workflows. Specifically, the Reflection Workflow example provides a detailed walkthrough of implementing a workflow that dynamically validates and retries structured outputs ^[4][5][6][7].

Regarding future support for human-in-the-loop features similar to LangGraph, I couldn't find specific information in the repository. It might be worth keeping an eye on the repository for any updates or new feature announcements.

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

1. To ensure that the stream_events loop continues until a StopEvent is encountered, you can explicitly check for the StopEvent within the loop and break out of it when it is received. This ensures that the loop doesn't exit prematurely:

   async for event in handler.stream_events():
       if isinstance(event, WorkflowStatusEvent):
           print(event.msg)
       elif isinstance(event, InputRequiredEvent):
           print(f"Waiting for human response to {event.prefix}: {event.msg}.")
       elif isinstance(event, StopEvent):
           print("StopEvent received, ending event stream.")
           break

2. If you don't see anything printed from the pending_tasks loop and the asyncio.wait call doesn't timeout, it suggests that there are no pending tasks and the workflow should have completed. Ensure that the done_tasks are being checked correctly and that the handler's result is being accessed properly:

   if done_tasks:
       result = handler.result()
       print(result)

3. Checking the received human response in intercept_human_intervention is a good step. To verify if the workflow handler is actively waiting for events, ensure that the send_event method is called while the workflow is in a state to process incoming events. The fact that the handler prints "Step intercept_human_intervention produced event StopEvent" does indicate that it is processing the human response correctly.

4. If all steps are being executed as expected, but the second WorkflowStatusEvent is not received, ensure that the event is being correctly written to the stream and that there are no issues with the event queue. Double-check the logic in ctx.write_event_to_stream to ensure it is functioning as intended.

5. Since no exceptions are caught, and the workflow seems to be processing events, consider adding additional logging around the event dispatching and handling to trace the exact flow. This might help identify where the process might be getting stuck or if there are any overlooked issues.

To continue talking to Dosu, mention @dosu.

[anirbanbasu]

@dosu, thanks for your response.

The weird thing is that adding the StopEvent check like you suggested as well as another catch-all else block, neither the "StopEvent received, ending event stream." is printed nor the "Some other event [...]". This means that the async for loop is somehow exiting printing "Waiting for human response to hitl_test: I need a human to proceed!" from the line print(f"Waiting for human response to {event.prefix}: {event.msg}.").

Interestingly, though the "Done with the event stream." message is not printed, which suggests that the async for never exits and yet is fails to catch the events after the InputRequireEvent.

async for event in handler.stream_events():
    if isinstance(event, WorkflowStatusEvent):
        print(event.msg)
    elif isinstance(event, InputRequiredEvent):
        print(f"Waiting for human response to {event.prefix}: {event.msg}.")
    elif isinstance(event, StopEvent):
        print("StopEvent received, ending event stream.")
        break
    else:
        print(f"Some other event: {event}")
print("Done with the event stream.")

[dosubot[bot]]

It seems like the async for loop is not processing events after the InputRequiredEvent, which might indicate an issue with how events are being queued or processed. Here are a few things to check and try:

1. Event Queue: Ensure that the event queue is not getting blocked or stuck. Verify that the streaming_queue in the context (ctx) is functioning correctly and that events are being added to it as expected.

2. Event Handling Logic: Double-check the logic in ctx.write_event_to_stream and ctx.send_event to ensure they are correctly handling and dispatching events. There might be an issue with how events are being enqueued or dequeued.

3. Concurrency Issues: Consider potential concurrency issues that might be affecting the event processing. Ensure that the workflow and event handling are properly synchronized, especially if there are multiple threads or asynchronous tasks involved.

4. Debugging: Add more detailed logging around the event queue operations and the stream_events method to trace the flow of events. This can help identify where the process might be getting stuck or if events are not being processed as expected.

5. Simplify the Workflow: As a debugging step, try simplifying the workflow to isolate the issue. For example, remove any additional complexity and focus on the basic event handling to see if the problem persists.

These steps should help you identify where the issue might be occurring and ensure that the event stream is processed correctly.

To continue talking to Dosu, mention @dosu.

[anirbanbasu]

@dosu, thanks! Starting with the event queue, running the following from a different thread where ic is the icecream debugger.

q = self._running_workflow_handler.ctx.streaming_queue
ic(q.qsize(), q.maxsize, q)

I get:

ic| q.qsize(): 0                                                                                                               
    q.maxsize: 0
    q: <Queue at 0x16fd666f0 maxsize=0 tasks=4>

Why is the queue max size and size both 0? I have a feeling that this is a concurrency issue since the system I am working with is more complex. I will try to simplify it as you have suggested.

[dosubot[bot]]

The streaming_queue having a maxsize of 0 typically indicates that the queue is unbounded, meaning it can grow as needed without a predefined limit. However, the qsize() being 0 suggests that there are no events currently in the queue, which might be why your event stream isn't processing further events.

Here are a few things to consider:

1. Unbounded Queue: An unbounded queue (maxsize=0) should not inherently cause issues with event processing, but it does mean that the queue can grow indefinitely. Ensure that events are being added to the queue correctly and that there are no logic errors preventing events from being enqueued.

2. Concurrency and Synchronization: Since you suspect a concurrency issue, ensure that all access to the queue is properly synchronized. If multiple threads or asynchronous tasks are interacting with the queue, make sure they are not interfering with each other.

3. Event Production: Verify that events are being produced and added to the queue as expected. If the queue is empty, it might indicate that events are not being generated or enqueued correctly after the InputRequiredEvent.

4. Simplification: As you mentioned, simplifying the system can help isolate the issue. Start with a minimal setup to ensure that basic event handling works, then gradually reintroduce complexity to identify where the problem arises.

By focusing on these areas, you should be able to determine why the event stream isn't processing as expected and address any concurrency issues that might be affecting the workflow.

To continue talking to Dosu, mention @dosu.