LlamaIndex Workflows explained: event-driven agents

LlamaIndex workflows let you model agent logic as event-driven steps instead of a monolithic chain. Unlike a linear chain, llamaindex workflows treat each unit of work as a step that emits and consumes typed events, giving you explicit control over branching, concurrency, and retries. This guide walks through building a practical event-driven agent with the Workflow API, from skeleton to production concerns.

Why event-driven agents beat linear chains

A typical ReAct loop jams planning, tool calls, and synthesis into one async function with a while loop and a state dict. That works for a demo, but it breaks down when you need parallel retrieval, human-in-the-loop approval, or partial failure recovery. The moment you add a second tool type or a conditional branch, the loop becomes a nest of flags and counters that no one wants to debug.

The workflow abstraction fixes this by making every transition explicit. With llamaindex workflows, each @step is an independent coroutine. The engine routes events based on type. You keep state in a Context object, not in fragile closure variables. This makes the agent graph inspectable, unit-testable, and easy to reason about when something stalls in production.

Event-driven design also maps cleanly to distributed systems. Each step can be a separate service if you later swap the in-memory runner for a queue. The event schema is the contract.

Install and scaffold

Use the core package:

pip install llama-index-core

A minimal workflow looks like this:

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

class HelloWorkflow(Workflow):
    @step
    async def run(self, ctx: Context, ev: StartEvent) -> StopEvent:
        return StopEvent(result="hello")

w = HelloWorkflow()
result = await w.run(query="ignored")
print(result)

StartEvent is injected by the engine when you call run(). StopEvent terminates the run and carries the final payload. Anything else is your own Event subclass. The query kwarg passed to run() is attached to StartEvent as an attribute, so you can read ev.query inside the step.

Define custom events and steps

Real agents need domain events. Subclass Event:

from llama_index.core.workflow import Event

class SubQueryEvent(Event):
    subquery: str

class SubResultEvent(Event):
    subquery: str
    answer: str

A step that only dispatches work should call ctx.send_event and return None:

class ResearchWorkflow(Workflow):
    @step
    async def decompose(self, ctx: Context, ev: StartEvent) -> None:
        # assume a list of subqueries from an LLM call
        subqueries = ["what is X", "how does Y work"]
        ctx.data["expected"] = len(subqueries)
        for q in subqueries:
            ctx.send_event(SubQueryEvent(subquery=q))

Note ctx.data is a dict persisted for the workflow instance. Use it to track counts or intermediate state. Do not treat it as a free-for-all; prefer passing explicit fields in events when possible.

Build a parallel research agent

The next step consumes SubQueryEvent, performs a retrieval or LLM call, and emits SubResultEvent:

    @step
    async def research(self, ctx: Context, ev: SubQueryEvent) -> SubResultEvent:
        # placeholder for retrieval + generation
        answer = f"answer for {ev.subquery}"
        return SubResultEvent(subquery=ev.subquery, answer=answer)

Now aggregate. A step can block on incoming events with ctx.wait_for_event:

    @step
    async def aggregate(self, ctx: Context, ev: StartEvent) -> StopEvent:
        expected = ctx.data.get("expected", 0)
        answers = []
        for _ in range(expected):
            res = await ctx.wait_for_event(SubResultEvent)
            answers.append(res.answer)
        return StopEvent(result=answers)

aggregate also listens on StartEvent, which means it runs concurrently with decompose. That is intentional: it waits while research steps process. This pattern is the core of event-driven agents—multiple steps subscribe to different events and the engine schedules them as tasks on the same loop.

Pitfall: deadlocks from missing sends

If expected is wrong or a research step throws, wait_for_event hangs forever. Always wrap external calls in try/except and send a failure event or decrement the expected count. A hung workflow consumes a coroutine and a connection until timeout at the infrastructure layer.

Error handling and retries

Steps are just coroutines. Use standard Python exception handling:

    @step
    async def research(self, ctx: Context, ev: SubQueryEvent) -> SubResultEvent:
        try:
            answer = await call_llm(ev.subquery)
        except TimeoutError:
            ctx.send_event(SubQueryEvent(subquery=ev.subquery))  # retry once
            return None
        return SubResultEvent(subquery=ev.subquery, answer=answer)

Tradeoff: naive retry can duplicate work if the original succeeded but the event was delayed. For production, make steps idempotent and tag events with a UUID. Store seen IDs in ctx.data and drop duplicates.

Wire LLM calls and model routing

Inside a step you can use any LlamaIndex LLM. If you route through an OpenRouter-class gateway like n4n.ai, you get automatic fallback across 240+ models when a provider is rate-limited, which pairs well with long-running llamaindex workflows that can’t afford a hard failure mid-stream. The code is unchanged:

from llama_index.llms.openai import OpenAI

llm = OpenAI(
    api_key="your-key",
    base_url="https://api.n4n.ai/v1",  # OpenAI-compatible
    model="anthropic/claude-3.5-sonnet",
)

Honor provider cache-control hints by passing extra_headers if your gateway forwards them; this cuts cost on repeated subqueries. The workflow does not care which model answered, only that a string came back.

Running the workflow end-to-end

Instantiate and run:

wf = ResearchWorkflow()
final = await wf.run(query="explain transformers")
print(final)

The engine starts decompose and aggregate on StartEvent, then routes SubQueryEvent to research. Because research is pure async, both subqueries execute concurrently on the event loop. For CPU-bound work, offload to a thread pool with asyncio.to_thread.

Testing workflows

Unit-test steps by constructing fake events:

async def test_research():
    wf = ResearchWorkflow()
    ctx = Context(wf)
    ev = SubQueryEvent(subquery="test")
    out = await wf.research(ctx, ev)
    assert isinstance(out, SubResultEvent)

Do not call wf.run() in every test; it spins the full event loop. Test steps in isolation, then write one integration test for the happy path. Use pytest-asyncio and mark coroutines with @pytest.mark.asyncio.

Pitfall: forgetting async

All @step methods must be async. A missing await on an LLM call returns a coroutine, which the engine will not resolve. The workflow will appear to silently stall. Run a linter that flags unawaited coroutines in step files.

Productionizing: observability and limits

Add a max-steps guard. The workflow engine does not inherently limit total events. Store a counter in ctx.data and bail out after a threshold:

ctx.data["steps"] = ctx.data.get("steps", 0) + 1
if ctx.data["steps"] > 50:
    return StopEvent(result="limit exceeded")

Log every event type in a @step that subscribes to the base Event class if you need a trace. For distributed execution, serialize Context.data to Redis; the default in-memory store dies with the process. If you need human approval, send a ApprovalEvent and pause the workflow by not sending the continuation event until a webhook calls ctx.send_event.

Common pitfalls and tradeoffs

• Over-granular steps: too many tiny steps inflate latency from event scheduling overhead. Batch where possible.
• Hidden state: ctx.data is mutable global state. Prefer passing explicit fields in events.
• Event type explosions: resist creating a new event per branch. Use a single event with a kind enum.
• Sync blocking: never call time.sleep or sync HTTP inside a step; you block the event loop and stall all parallel branches.
• Unbounded waits: wait_for_event without a timeout is a liability. Wrap it in asyncio.wait_for.
• Duplicate side effects: because retries resend events, any external write must be idempotent.

LlamaIndex workflows are not a silver bullet. For a single sequential call, a plain function is simpler. Use the workflow engine when you have genuine branching, concurrency, or long-running recovery needs. When those appear, the explicit event graph will save you from spaghetti agent code.

Where to go next

Read the LlamaIndex workflow docs for Context.stream_events and human-in-the-loop patterns. The patterns above—typed events, explicit dispatch, guarded aggregation—are the foundation for any serious event-driven agent built on llamaindex workflows. Start with one parallel branch, add a retry policy, then instrument the context before you ship.