How to do human-in-the-loop with Nested Workflow

[main-2983]

I try to replicate this example for Human-in-the-loop, but using nested workflow. I got TimeOut: llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds

First I rewrite the HumanInTheLoopWorkflow to accept another workflow

class HumanInTheLoopWorkflow2(Workflow):
    @step
    async def step0(self, ev: StartEvent) -> Event1:
        return Event1()

    @step
    async def step1(self, ev: Event1, wf: HITLWorkflow) -> StopEvent:
        result = await wf.run()

        return StopEvent(result)

The actually Human in the Loop sub-workflow

class HITLWorkflow(Workflow):
    @step
    async def step1(self, ev: StartEvent) -> InputRequiredEvent:
        return InputRequiredEvent(prefix="Enter a number: ")

    @step
    async def fin(self, ev: HumanResponseEvent) -> StopEvent:
        return StopEvent(ev.response)

Then I run the workflow using the code snippet

async def main():
    # workflow should work with streaming
    workflow = HumanInTheLoopWorkflow2()
    sub_wf = HITLWorkflow()

    workflow.add_workflows(wf=sub_wf)

    handler = workflow.run()

    async for event in handler.stream_events():
        print(event)
        if isinstance(event, InputRequiredEvent):
            # here, we can handle human input however you want
            # this means using input(), websockets, accessing async state, etc.
            # here, we just use input()
            response = input(event.prefix)
            handler.ctx.send_event(HumanResponseEvent(response=response))

    final_result = await handler

    print(final_result)


if __name__ == '__main__':
    import asyncio

    asyncio.run(main())

I was told to expose the event of the inner workflow but I don't know how to do it? Can somebody help?

[dosubot[bot]]

To expose events from the inner (nested) workflow in a human-in-the-loop setup, you need to ensure that the events produced by the sub-workflow (like InputRequiredEvent) are surfaced to the outer workflow's event stream, so your main event loop can handle them (e.g., prompt the user for input).

In your current setup, the outer workflow (HumanInTheLoopWorkflow2) awaits the result of the inner workflow (HITLWorkflow) by calling await wf.run(). However, this call does not automatically propagate the inner workflow's events to the outer workflow's event stream. As a result, your main loop never sees the InputRequiredEvent from the inner workflow, leading to a timeout because the expected human input never arrives.

How to expose inner workflow events:

Instead of awaiting wf.run() directly inside a step, you should "forward" the events from the inner workflow to the outer workflow's handler. This can be done by iterating over the inner workflow's event stream and yielding those events from the outer workflow. Here’s a pattern you can use:

class HumanInTheLoopWorkflow2(Workflow):
    @step
    async def step0(self, ev: StartEvent) -> Event1:
        return Event1()

    @step
    async def step1(self, ev: Event1, wf: HITLWorkflow, ctx: Context) -> StopEvent:
        handler = wf.run(ctx=ctx)
        async for event in handler.stream_events():
            ctx.send_event(event)  # Expose inner workflow events to the outer context
            if isinstance(event, InputRequiredEvent):
                # Wait for human input event to be sent to the context
                response_event = await ctx.wait_for_event(HumanResponseEvent)
                handler.ctx.send_event(response_event)
        result = await handler
        return StopEvent(result)

This approach ensures that any InputRequiredEvent from the inner workflow is visible to your main event loop, allowing you to handle human input as needed. The key is to use the same context (ctx) for both workflows and to forward events explicitly.

Why you got a WorkflowTimeoutError:

The timeout occurs because the inner workflow emits an InputRequiredEvent, but since it is not exposed to the main event loop, no human input is ever received, and the workflow waits indefinitely until it times out. This is a common issue when nesting workflows without proper event propagation. For more details on nested workflows and event handling, see the nested workflow documentation.

Other tips to avoid deadlocks and timeouts:

• Always ensure that all events requiring external input are surfaced to the main event loop.
• Use while not handler.is_done() or similar patterns to step through workflows and synchronize event handling, as recommended in workflow timeout discussions.
• Avoid directly awaiting wf.run() inside a step if you need to interact with its events.

If you need a more concrete example or run into further issues, let me know!

To reply, just mention my name @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}   

[main-2983]

Is the main function still the same as this:

async def main():
    # workflow should work with streaming
    workflow = HumanInTheLoopWorkflow2()
    sub_wf = HITLWorkflow()

    workflow.add_workflows(wf=sub_wf)

    handler = workflow.run()

    async for event in handler.stream_events():
        print(event)
        if isinstance(event, InputRequiredEvent):
            # here, we can handle human input however you want
            # this means using input(), websockets, accessing async state, etc.
            # here, we just use input()
            response = input(event.prefix)
            handler.ctx.send_event(HumanResponseEvent(response=response))

    final_result = await handler

    print(final_result)

Using the code you provide above I still got WorkflowTimeoutError @dosubot

[dosubot[bot]]

Your main function is structurally correct for streaming events and handling human input, but the root cause of your WorkflowTimeoutError is likely in how the nested workflow's events are exposed and handled in your HumanInTheLoopWorkflow2 class.

In your current implementation, the outer workflow simply does result = await wf.run(), which does not propagate the inner workflow's InputRequiredEvent to the main event loop. As a result, your main loop never sees the event and cannot provide the required human input, so the workflow times out after 10 seconds (the default) reference.

To fix this, you need to explicitly forward the inner workflow's events to the outer context. Here’s how you should modify your HumanInTheLoopWorkflow2:

class HumanInTheLoopWorkflow2(Workflow):
    @step
    async def step0(self, ev: StartEvent) -> Event1:
        return Event1()

    @step
    async def step1(self, ev: Event1, wf: HITLWorkflow, ctx: Context) -> StopEvent:
        handler = wf.run(ctx=ctx)
        async for event in handler.stream_events():
            ctx.send_event(event)  # Forward inner events to the outer context
            if isinstance(event, InputRequiredEvent):
                response_event = await ctx.wait_for_event(HumanResponseEvent)
                handler.ctx.send_event(response_event)
        result = await handler
        return StopEvent(result)

This ensures that InputRequiredEvent from the inner workflow is visible to your main event loop, so your main function can handle it as expected.

Other important notes:

• If you still get a timeout, try setting timeout=None in your workflow constructors to disable the default 10-second timeout reference.
• Do not share the Context object between workflows unless you are intentionally forwarding events as above reference.
• If you are running in Jupyter or another async environment, ensure your event loop is properly managed (e.g., with nest_asyncio).

Your main function can remain as you wrote it, but the key is that the outer workflow must forward the inner workflow’s events. If you make the above change to HumanInTheLoopWorkflow2, your main function should work as intended.

If you still encounter issues, double-check that your step signatures include ctx: Context and that you are not accidentally blocking or missing any events in your workflow steps.

To reply, just mention my name @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}