Workflows: steps and events

Workflows are how LlamaIndex expresses any multi-step AI process, including its own agents, with full control.

A Workflow is a class whose async methods, marked with @step, react to events. An event is a small typed message, a Pydantic class you define by subclassing Event.

The type hints define the wiring:

@step
async def draft(self, ev: StartEvent) -> DraftEvent: ...
@step
async def review(self, ev: DraftEvent) -> StopEvent: ...

StartEvent carries the keyword arguments given to run(...): ev.topic, or ev.get("topic") when it may be missing.
A step returning DraftEvent triggers every step that accepts DraftEvent.
Returning StopEvent(result=...) ends the run; await workflow.run(...) gives you that result.

LlamaIndex checks the wiring before running, so a step whose event nothing produces is caught early.

Patterns:

Branching: return different event types from one step (annotate the return as EventA | EventB).
Loops: return an event handled by an earlier step, for example a retry after a failed check. Keep a counter.
Parallel work: ctx.send_event(...) several times, and collect results with ctx.collect_events(ev, [ResultEvent] * n).
Shared state: add ctx: Context to a step and use await ctx.store.get/set.
Progress: ctx.write_event_to_stream(ev) sends events to handler.stream_events().

Set timeout= in the constructor (seconds) to stop runaway runs. Workflows are also published as the standalone llama-index-workflows package (imported as workflows); inside LlamaIndex use llama_index.core.workflow.

Use a workflow when the steps are known (classify, retrieve, check, answer); use an agent when the LLM should choose the steps. Agents themselves are workflows, so they can be called from a step.

Example

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

class DraftEvent(Event):
    text: str
class RetryEvent(Event): pass
class SloganFlow(Workflow):
    @step
    async def write(self, ctx: Context, ev: StartEvent | RetryEvent) -> DraftEvent:
        tries = await ctx.store.get("tries", default=0) + 1
        await ctx.store.set("tries", tries)
        product = ev.get("product") or await ctx.store.get("product")
        await ctx.store.set("product", product)
        # In real code: await llm.acomplete(...)
        return DraftEvent(text=f"{product}: try #{tries}" + ("!" if tries > 1 else ""))

    @step
    async def check(self, ctx: Context, ev: DraftEvent) -> RetryEvent | StopEvent:
        if not ev.text.endswith("!") and await ctx.store.get("tries") < 3:
            return RetryEvent()                       # loop back to write
        return StopEvent(result=ev.text)

async def main():
    print(await SloganFlow(timeout=30).run(product="SunMug"))   # SunMug: try #2!
asyncio.run(main())

Remember

@step methods consume and return typed Events; type hints define the wiring
StartEvent holds run() arguments; StopEvent(result=...) ends the run
Use ctx.store for state and timeout= to bound runs

Watch out for

Returning an event type that no step accepts, which the validation reports before the run starts.

Check yourself

What starts a step annotated as async def review(self, ev: DraftEvent) -> StopEvent?

Official docs: developers.llamaindex.ai/python/llamaagents/workflows/

Example

Remember

Watch out for

Check yourself

Builds on

Unlocks