LlamaIndex Workflows, or an agent built from events

LlamaIndex Workflows, or an agent built from events

Most agent building libraries describe a flow as a graph: nodes, edges, and transition conditions. Workflows picks a different concept. Here there are steps listening for events of a given type and emitting events themselves, and execution consists of the engine delivering each event to whichever step subscribed to it.

The difference looks cosmetic and is not. In a graph you must draw every edge in advance. In an event driven arrangement, adding a step means writing a function listening for an existing event, without touching anything else.

The name deserves clarifying right away, since it misleads. Workflows grew out of the LlamaIndex library, known for document retrieval, but it is today a separate package with its own repository and its own development path. It can be used without the rest of that ecosystem.

Your first workflow

pip install llama-index-workflows

A step is a method with a decorator, taking an event of one type and returning an event of another. All the control logic sits in those types.

from workflows import Workflow, step, Context
from workflows.events import StartEvent, StopEvent, Event

class QuestionExpanded(Event):
    question: str

class ChunksFound(Event):
    chunks: list[str]

class Assistant(Workflow):
    @step
    async def expand(self, ev: StartEvent) -> QuestionExpanded:
        full = await rephrase(ev.question)
        return QuestionExpanded(question=full)

    @step
    async def search(self, ev: QuestionExpanded) -> ChunksFound:
        return ChunksFound(chunks=await store.search(ev.question))

    @step
    async def answer(self, ev: ChunksFound) -> StopEvent:
        return StopEvent(result=await model.ask(ev.chunks))

result = await Assistant(timeout=60).run(question="What is the notice period?")

Nowhere did you write that search follows question expansion. It follows from the types: the expanding step returns an event the searching step subscribed to. Order is a consequence of the declarations rather than a separate description that could drift out of sync with them.

That carries a practical advantage appreciated around the third week of work. You cannot build a flow with an edge leading nowhere, since an event with no receiver is visible immediately.

Context and state

Steps do not see each other's variables, so passing state outside event payloads uses a context attached to the run.

@step
async def gather(self, ctx: Context, ev: ChunksFound) -> StopEvent:
    count = await ctx.store.get("attempts", default=0)
    await ctx.store.set("attempts", count + 1)
    return StopEvent(result=...)

The split between events and context deserves thinking through at the start, since it is changed reluctantly later. The rule that works in practice reads like this: an event carries what the next step needs, and context carries what concerns the whole run.

A case number, a user identifier, and an attempt counter belong in context. A search result passed onward belongs in an event. Dumping everything into context produces a flow where you cannot see what depends on what, meaning exactly what this model was meant to prevent.

Context can be written out as a dictionary and restored again. That opens two possibilities: resuming an interrupted run, and holding state between HTTP requests without keeping a process in memory.

Serialisation copes with objects described by a data model, storing the full class name alongside the values, so restoring returns the right type rather than a bare dictionary. Keep data in the context rather than objects carrying connections or resource handles, though, since those will point at nothing meaningful once restored.

Checkpoints and resumption

There is no built in checkpointer here, which is better known upfront. The library supplies the raw material: after each completed step the engine emits an internal event on the stream carrying a state snapshot, the input event name, and the output event name, and persisting that snapshot is on you.

In practice that is a dozen or so lines of code. You listen to the run stream with internal events exposed, dump the context to a dictionary on every step completion, and put it wherever you want. Resumption means rebuilding the context from that dictionary and handing it to the next run of the flow, in a different process too.

The significance shows in the model bills. An eight step flow whose seventh step fails on an external API error costs six model calls per repair attempt absent checkpoints. With checkpoints it costs none, since you resume from the failure point.

There is a caveat to know before production, though. Resumption is at least once. Completed steps do not run again, since their output already sits in the restored state, while a step interrupted midway runs from the top, so its side effects have to be safe to repeat.

The second advantage concerns flows involving a person. A step can emit an event awaiting a decision, with state sitting saved for as long as needed. An agent prepares a proposed reply for a customer, a person approves it in a panel, the flow continues, and six hours plus a server restart pass in between.

Plan where that state goes before writing your first production flow. Process memory suffices for a prototype and fails at the first deployment, since runs started beforehand are lost.

Concurrency and branching

This is where the event model shows its advantage over a graph, since concurrency falls out naturally.

A step can emit several events at once, and the listening steps then start simultaneously. Collecting results means waiting for a given number of events of a type.

@step
async def fan_out(self, ctx: Context, ev: QuestionExpanded) -> None:
    for source in ["documents", "database", "web"]:
        ctx.send_event(SearchIn(source=source, question=ev.question))

@step
async def gather(self, ctx: Context, ev: PartialResult) -> StopEvent | None:
    results = ctx.collect_events(ev, [PartialResult] * 3)
    if results is None:
        return None
    return StopEvent(result=merge(results))

Returning an empty value means the step does not yet have a full set and is waiting for more events. The engine calls it on every incoming result until the set completes.

This pattern covers most real needs: searching several sources at once, having several reviewers judge an answer, making independent API calls. The time saved equals the difference between the sum of the durations and the slowest step, which across three model calls usually means two thirds.

There is one trap worth knowing. If one of the events never arrives, because the step emitting it failed, collection waits until the timeout expires. The timeout set on the flow is a safeguard here rather than a formality.

Loops, conditions, and returns

The event model handles branching differently from a graph, and the three most common constructions deserve seeing.

A condition is a step returning one of several possible events. You declare the return type as a union of types, and the engine delivers the emitted event to the right receiver.

@step
async def judge(self, ev: AnswerReady) -> StopEvent | ReviseAgain:
    if await score(ev.text) >= 0.8:
        return StopEvent(result=ev.text)
    return ReviseAgain(text=ev.text, notes=await collect_notes(ev.text))

A loop arises from the same mechanism: the revising step emits an event that returns to the judging step. There is no separate concept of a cycle, since a cycle is simply an event travelling back up the flow.

That simplicity carries a price you must pay knowingly. A loop without a counter turns until the timeout expires, and every turn costs model calls. An attempt counter kept in context and checked in the condition is three lines of code whose omission can cost dozens of dollars on one failed deployment.

The third construction is early termination. A step returning a stopping event ends the whole run regardless of what is happening in parallel branches. It helps with input validation and with detecting that further work is pointless.

Testing workflows

A step based structure carries an advantage easy to miss: every step is an ordinary function taking an object and returning an object, so it is tested without running anything else.

That means branching logic can be checked without calling a model. The step deciding whether an answer is good enough is tested by handing it a prepared event and checking which type it returned. A second instead of thirty and zero cost instead of a model call.

The second level is testing a whole flow with external calls substituted. Putting stubs in place of the model and the search, you check whether the steps line up in the expected order and whether conditions lead where they should.

The third level, the most valuable and the most often skipped, is a set of real cases with expected results, run against a real model. Only that answers whether a change to the system prompt improved anything. Without it, judgement reduces to an impression from a few manual attempts, and impressions mislead regularly.

Agents on top

The library also provides ready agent classes built on the same mechanism, so you need not write the tool loop from scratch.

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    tools=[check_stock, place_order],
    llm=model,
    system_prompt="You handle orders. Always check stock before placing one.",
)

response = await agent.run("Order two pairs of shoes in size 42.")

Since an agent is a workflow, it can be embedded as a step inside a larger one. That is the right way to combine both levels: describe a process with known steps as a flow, and hand a single step requiring judgement to an agent.

The reverse, handing a whole process to one agent with fifteen tools, produces something that works in a demo and resists diagnosis when it fails.

Workflows against the alternatives

The first criterion is mundane: if your document retrieval layer already runs on LlamaIndex, this choice saves translating between two libraries' concepts.

The second concerns how you think about the flow. A graph suits a process with an explicit state machine carrying loops and returns. Events suit a process that is a set of reactions to what happened, and one where much runs in parallel.

The third, as always, is whether it is necessary. Three model calls in sequence need neither of these libraries, and bolting orchestration onto something that fits inside a single function adds concepts to learn for no gain at all.

Running it in production

A flow can be exposed as a service, and that is the part worth thinking through before writing your second flow.

Execution is asynchronous, so the natural arrangement is accepting a job, returning an identifier, and serving the result once it completes. Streaming intermediate events lets you show the user progress, which on a forty second flow is sometimes the difference between a working product and an abandoned page.

Observability rests on every event being recordable. A log holding all events from one run, with timings and token usage, answers the question of what went wrong without a rerun. That is this model's greatest practical advantage and deserves using from day one rather than after the first incident.

Set timeouts and cost caps at the flow level. A single step calling a model in a loop can make dozens of requests before anyone notices, and a ceiling is the only mechanism here that works without your attention.

Think through what happens to a half finished run when you deploy a new version of the code. If steps were renamed or events changed fields, the saved state no longer matches the new flow and resumption fails. The simplest answer is waiting for in flight runs to finish before deploying, usually sufficient with short flows. With flows waiting many hours on a human decision, event shapes need versioning, the way a database schema does.

Common mistakes

The first is keeping all state in context instead of in events. The flow then stops showing dependencies between steps, and that was this model's main benefit.

The second is no timeout when collecting concurrent results. One event that never arrives suspends the whole run until the limit expires.

The third is keeping checkpoints in process memory. They vanish with a deployment restart, so production needs durable storage.

The fourth is handing a whole process to one agent with a dozen or more tools. Describe known steps as a flow, since you gain repeatability and cheaper diagnosis.

The fifth is skipping event recording. The event model gives you an audit trail for free, and not using it wastes this library's greatest advantage.

The sixth is assuming the library requires the rest of the ecosystem. The package is standalone and works without the document retrieval layer.

FAQ

Do I need the whole of LlamaIndex?

No. Workflows is a standalone package with its own repository and development path, installed separately. The ready agent classes come from the wider ecosystem, while the step and event mechanism itself works without it.

How does it differ from LangGraph?

In the concept describing the flow. LangGraph builds a graph of nodes and edges, here you have steps listening for events. A graph suits an explicit state machine, events suit heavy concurrency and a process growing by further reactions.

Can an interrupted flow be resumed?

Yes, though the writing is on you. There is no built in checkpointer: after each step the engine emits an internal event carrying a state snapshot, and you persist the context as a dictionary and rebuild it on resumption. Plan durable storage for those records, and remember that a step interrupted midway will run again.

Does it work outside Python?

The core library is available in a TypeScript variant too, though the ecosystem around Python is noticeably richer. On a project built entirely on browser technologies, check that variant's scope before deciding.

When is it not worth using?

For a process that is a sequence of two or three model calls, with no branching and no need for resumption. Ordinary code is then shorter, cheaper, and more readable than any orchestration library.

Documentation sits on the project site, and the package in the PyPI registry.