#Workflows

LlamaIndex Workflows are event-driven and step-based: each step receives a typed Event, does work, and returns another Event. The runtime routes that event to the next @step whose annotations accept it. Loops are “return an event an earlier step handles.” Branches are ordinary ifs that return different event types. Official docs argue this feels closer to plain Python than encoding loops on DAG edges.

Canonical pages: Workflows introduction and Workflows API. If signatures disagree with this chapter, those pages win—do not invent parameters.

With llama-index / llama-index-core, the stable import path is llama_index.core.workflow. You can also pip install llama-index-workflows and import the standalone workflows package (newer official samples sometimes do). FunctionAgent runs on the same runtime, so Context in Quick Start is the same state object.

#Mental model

flowchart LR
  S[StartEvent] --> A["@step generate"]
  A --> E[Custom Event]
  E --> B["@step handle"]
  B --> X[StopEvent]

Two special events:

• StartEvent — keyword arguments to workflow.run(...) become start fields (topic= → ev.topic)
• StopEvent — returning it ends the run; result is what await w.run(...) yields

Custom events are Pydantic models. @step input/output types are validated before run: missing producers/consumers or no StopEvent usually fail at startup.

#Smallest useful shape (official JokeFlow idea)

The introduction page’s “generate → process → stop” pattern, with gpt-4o-mini. If your install wants from workflows import Workflow, step, follow that page—do not mix undocumented class names.

from llama_index.core.workflow import (
    Workflow,
    step,
    Event,
    StartEvent,
    StopEvent,
)
from llama_index.llms.openai import OpenAI

class JokeEvent(Event):
    joke: str

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        prompt = f"Write your best joke about {ev.topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        prompt = f"Critique this joke:\n{ev.joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

async def main():
    w = JokeFlow(timeout=60, verbose=False)
    print(await w.run(topic="pirates"))

To stream intermediates, keep the handler: handler = w.run(...), async for ev in handler.stream_events(), then await handler.

#Connecting a query pipeline

RAG inside a workflow is usually StartEvent → retrieve step (custom Retrieved event) → synthesize → StopEvent. The retrieve step calls your retriever.aretrieve or query_engine.aquery and puts text on the event. Official notebooks cover RAG + rerank, corrective RAG, citation engines, and more—copy parameter lists from that Examples section, not from memory.

Shared state, HITL, and fan-out live in the official table: ctx.store, ctx.send_event, returning list[Event], and so on. Those Context APIs skip some static graph checks if you overuse them.

#vs LangGraph / CrewAI

A fixed retrieve → generate path is a QueryEngine. Reach for Workflows when you need loops, branches, or parallel sub-queries. For a standard tool agent, start with FunctionAgent; for cross-framework orchestration, LangGraph or CrewAI.

#Next steps

• Vector Stores
• Practical Examples
• Official: Workflows introduction