Are events "consumed" or "deleted" in workflows?

[InAnYan]

Hi! I've been using LlamaIndex workflows recently, and I'm puzzled about how it works internally.

The thing I don't understand -- what its mathematical model is? Docs says it has event-driven architecture, but this doesn't explain much.

Take for example this code:

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 Test(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 from_a_b(self, ctx: Context, evs: EventA | EventB) -> Optional[StopEvent]:
        got: Optional[Tuple[EventA, EventB]] = ctx.collect_events(evs, [EventA, EventB])  # type: ignore
        
        if not got:
            return None
        
        a, b = got
        
        return StopEvent()


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


asyncio.run(main())

Which produces:

```plain Running step start Step start produced event EventA Running step from_a_b Step from_a_b produced no event Running step make_b Step make_b produced event EventB Running step from_a_b Step from_a_b produced event StopEvent ```

So, it's like EventA is consumed in make_b and from_a_b steps. How does LlamaIndex workflow
understand that when make_b is called, the EventA must not be "consumed" or "deleted" fully, but rather saved for later from_a_b call? This is very interesting to me.

[dosubot[bot]]

In LlamaIndex workflows, events are not "deleted" after being consumed; instead, they are managed through a mechanism that allows them to be retained for future use. This is achieved using the ctx.collect_events() method, which collects and stores events until all required events have been gathered. This method allows the workflow to wait for a specified number of events of certain types before proceeding with the next step, enabling the workflow to handle multiple events and process them together ^[1].

In your example, the from_a_b step uses ctx.collect_events(evs, [EventA, EventB]) to wait for both EventA and EventB to be available. This ensures that EventA is not fully consumed or deleted after the make_b step, allowing it to be used again in the from_a_b step. The workflow's event-driven architecture allows for this kind of event retention and coordination between steps, ensuring that events are processed in the correct order and context.

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}

[InAnYan]

Another thing that concerns me -- won't that introduce memory leaks?

I'm sorry if I dig up too much or my ideas are not relevant or too far-fetched

[InAnYan]

I continue to torture workflows, and I've found in sources that wait_for_event actually consumes events! I immediately thought that if someone would not know this property of wait_for_event function, this might lead to bugs.

Take this code for example:

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


class EventA(Event):
    pass


class EventB(Event):
    pass


class Test(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 finish(self, ctx: Context, ev: EventA | EventB) -> StopEvent:
        b = await ctx.wait_for_event(EventB)
        print("Received b")
        a = await ctx.wait_for_event(EventA)
        print("Received a")
        return StopEvent()


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


asyncio.run(main())

On my machine this produces:

Running step start
Step start produced event EventA
Running step finish
Running step make_b
Step make_b produced event EventB
Received b
Running step finish

And it ends in timeout error. My guess why this happened: first wait_for_event call waited for EventB, before EventB was produced, there was EventA generated, which wait_for_event deleted.

I've added some print calls inside Context.wait_for_event, and got this:

Running step start
Step start produced event EventA
Running step finish
Got <class 'llama_index.core.workflow.events.StartEvent'>
Got <class '__main__.EventA'>
Running step make_b
Step make_b produced event EventB
Got <class '__main__.EventB'>
Received b
Running step finish

Q. E. D.

[logan-markewich]

Wait for event is definitely a little buggy, I implemented it badly (whoops)

Trying to fix it here, need more time to look at it though. If you have any ideas for an approach or pr, feel free to help out
#18226

[logan-markewich]

There is no "mathematical model" -- saying that its event driven is correct

Steps only run when their required events are emitted, and the system keeps track of events in queues and whatnot, and handles emitting events that steps return