[Bug]: Workflow doesn't work

[InAnYan]

Bug Description

Triggered by #18245 (comment). I thought that LlamaIndex did some magic.

But I made a Workflow where an event is consumed, which blocks the execution.

I don't think that you can do anything here. But I think it needs an explanation for users.

(My inner Knuth is now happy)

Sorry for the strange issue title.

Version

0.12.25

Steps to Reproduce

Diagram:

import asyncio
from typing import Optional
from llama_index.core.workflow import Context, Event, StartEvent, StopEvent, Workflow, step


class EventA(Event):
    pass


class EventB(Event):
    pass


class EventC(Event):
    pass


class TestWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> EventA:
        return EventA()

    @step
    async def make_b(self, ev: EventA) -> EventB:
        return EventB()

    @step
    async def make_c(self, ev: EventB) -> EventC:
        return EventC()

    @step
    async def finish(self, ctx: Context, evs: EventA | EventC) -> Optional[StopEvent]:
        got = ctx.collect_events(evs, [EventA, EventB])
        
        if not got:
            return None
        
        return StopEvent()


async def main():
    w = TestWorkflow(verbose=True)
    await w.run()


asyncio.run(main())

Relevant Logs/Tracbacks

Running step start
Step start produced event EventA
Running step finish
Step finish produced no event
Running step make_b
Step make_b produced event EventB
Running step make_c
Step make_c produced event EventC
Running step finish
Step finish produced no event
Exception in callback Dispatcher.span.<locals>.wrapper.<locals>.handle_future_result()() at [REDACTED]/.venv/lib/python3.13/site-packages/llama_index/core/instrumentation/dispatcher.py:274
handle: <Handle Dispatcher.span.<locals>.wrapper.<locals>.handle_future_result()() at [REDACTED]/.venv/lib/python3.13/site-packages/llama_index/core/instrumentation/dispatcher.py:274>
Traceback (most recent call last):
  File "/usr/lib/python3.13/asyncio/events.py", line 89, in _run
    self._context.run(self._callback, *self._args)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[REDACTED]/.venv/lib/python3.13/site-packages/llama_index/core/instrumentation/dispatcher.py", line 286, in handle_future_result
    raise exception
  File "[REDACTED]/.venv/lib/python3.13/site-packages/llama_index/core/workflow/workflow.py", line 401, in _run_workflow
    raise WorkflowTimeoutError(msg)
llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds
Traceback (most recent call last):
  File "[REDACTED]/test.py", line 47, in <module>
    asyncio.run(main())
    ~~~~~~~~~~~^^^^^^^^
  File "/usr/lib/python3.13/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "/usr/lib/python3.13/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "/usr/lib/python3.13/asyncio/base_events.py", line 725, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "[REDACTED]/test.py", line 44, in main
    await w.run()
  File "[REDACTED]/.venv/lib/python3.13/site-packages/llama_index/core/workflow/workflow.py", line 401, in _run_workflow                                                                                        
    raise WorkflowTimeoutError(msg)
llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds

[dosubot]

There are two similar discussions related to event consumption and handling in LlamaIndex workflows:

1. Are events "consumed" or "deleted" in workflows? - This discussion addresses issues with event consumption and the need for better documentation on how events are managed within workflows ^[1].

2. Workflow errors in collect_events - This discussion involves errors related to the collect_events method, which might be relevant to your issue with the finish step producing no event ^[2].

Regarding your issue, the collect_events method in LlamaIndex captures and buffers events, returning None until all required events are collected. It seems that the finish step is not receiving all the expected events (EventA and EventB), which is why it sometimes produces no event and leads to a WorkflowTimeoutError ^[3][4]. You might want to ensure that all necessary events are being correctly produced and collected before the finish step is executed.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[logan-markewich]

Im confused -- why should this workflow ever work?

The finish step does not accept EventB, so it will wait forever

[InAnYan]

I'm terribly sorry, yes, you are right

[InAnYan]

It still works. LlamaIndex Workflows are magical

[InAnYan]

BTW, this is also shows the power of static typing (when it's used).

If @step could understand when several event types are expected, then it would eliminate call to ctx.collect_events, and this problem would never occur

[logan-markewich]

Collect events is moreso for map-reduce type patterns (at least thats how I always use it)