Can several steps accept one event in a workflow?

[InAnYan]

For example, I have a step that produces event A. Can I have two other steps (step1, step2) which accept A event and process it independently?

Because I always thought that whenever step is called, it always consumes event. Like step1 accepts A, step2 won't get that A. Would that introduce a conflict?

Because in the docs there were examples on:

1. Accepting 1 event and returning 1 event.
2. Accepting several events and returning 1 event.
3. Accepting 1 event and returning several events.
4. (It implies) accepting several events and returning several events.

But all of those steps were fully consumed events.

[dosubot[bot]]

Yes, multiple steps in a workflow can independently process the same event without conflict. The Workflow class allows for defining steps that can handle specific event types, and there are no restrictions on multiple steps accepting and processing the same event type. Therefore, both step1 and step2 can accept and process event A independently, as long as they are configured to accept that event type. The workflow's event-driven architecture supports this kind of parallel processing of events by different steps ^[1].

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]

Yes, you can do this without problems.

Here is an example:

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


class ProcessEvent(Event):
    start: StartEvent


class ResultA(Event):
    process: ProcessEvent


class ResultB(Event):
    process: ProcessEvent


class MyStopEvent(StopEvent):
    a: ResultA
    b: ResultB
    

class MyWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> ProcessEvent:
        return ProcessEvent(start=ev)

    @step
    async def make_result_a(self, ev: ProcessEvent) -> ResultA:
        return ResultA(process=ev)

    @step
    async def make_result_b(self, ev: ProcessEvent) -> ResultB:
        return ResultB(process=ev)

    @step
    async def finish(self, ctx: Context, ev: ResultA | ResultB) -> Optional[MyStopEvent]:
        events = ctx.collect_events(ev, [ResultA, ResultB])
        if not events:
            return None

        result_a, result_b = events

        return MyStopEvent(a=result_a, b=result_b)


if __name__ == '__main__':
    async def main():
        workflow = MyWorkflow(verbose=True)
        result = await workflow.run()
        pprint(result)

    asyncio.run(main())

This will produce output:

Running step start
Step start produced event ProcessEvent
Running step make_result_a
Step make_result_a produced event ResultA
Running step make_result_b
Step make_result_b produced event ResultB
Running step finish
Step finish produced no event
Running step finish
Step finish produced event MyStopEvent
MyStopEvent(a=ResultA(process=ProcessEvent(start=StartEvent())), b=ResultB(process=ProcessEvent(start=StartEvent())))

As you can see make_result_a and make_result_b successfully accepted ProcessEvent and made their Results. finish step could successfully receive results and produce a MyStopEvent

[InAnYan]

I hope maintainers will find okay for "self-answering" culture, like on StackOverflow

[InAnYan]

BTW, this even works if these two steps accept a StartEvent, not only some intermediate event:

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

class ResultA(Event):
    process: StartEvent


class ResultB(Event):
    process: StartEvent


class MyStopEvent(StopEvent):
    a: ResultA
    b: ResultB
    

class MyWorkflow(Workflow):
    @step
    async def make_result_a(self, ev: StartEvent) -> ResultA:
        return ResultA(process=ev)

    @step
    async def make_result_b(self, ev: StartEvent) -> ResultB:
        return ResultB(process=ev)

    @step
    async def finish(self, ctx: Context, ev: ResultA | ResultB) -> Optional[MyStopEvent]:
        events = ctx.collect_events(ev, [ResultA, ResultB])
        if not events:
            return None

        result_a, result_b = events

        return MyStopEvent(a=result_a, b=result_b)


if __name__ == '__main__':
    async def main():
        workflow = MyWorkflow(verbose=True)
        result = await workflow.run()
        pprint(result)

    asyncio.run(main())

Perfectly produces:

Running step make_result_a
Step make_result_a produced event ResultA
Running step make_result_b
Step make_result_b produced event ResultB
Running step finish
Step finish produced no event
Running step finish
Step finish produced event MyStopEvent
MyStopEvent(a=ResultA(process=StartEvent()), b=ResultB(process=StartEvent()))