LlamaIndex Workflows: steps, events and multi-agent systems

This guide is for readers who already write Python functions and want several steps, or several agents, to work together in a fixed order. By the end you will be able to build single-step and multi-step workflows with Workflow, understand how events and state pass between steps, and assemble a multi-agent workflow that shares state.

What a workflow is

A workflow in LlamaIndex organizes code into sequential, manageable steps. You define Steps, each triggered by Events, and each step can emit new Events that trigger the next step^[1].

Workflows offer clear organization of code into discrete steps, an event-driven architecture for flexible control flow, type-safe communication between steps, built-in state management, and support for both simple and complex agent interactions^[1].

Setup

Install the Workflow package with pip install llama-index-utils-workflow^[1]. A notebook that runs on Google Colab is available for following the code step by step^[1].

Step 1: A single-step workflow

A single-step workflow is a class that inherits from Workflow, with functions decorated with @step. StartEvent and StopEvent are special events that mark the start and the end of the workflow^[1]:

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

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        # do something here
        return StopEvent(result="Hello, world!")


w = MyWorkflow(timeout=10, verbose=False)
result = await w.run()

Call w.run() to run the workflow^[1]. The example creates the workflow object with timeout=10 and verbose=False^[1].

Step 2: Pass data between steps with events

To connect several steps, you create custom events that carry data. An Event subclass passes the output of the first step to the second. In the example, ProcessingEvent does this job^[1]:

from llama_index.core.workflow import Event

class ProcessingEvent(Event):
    intermediate_result: str

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> ProcessingEvent:
        # Process initial data
        return ProcessingEvent(intermediate_result="Step 1 complete")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

w = MultiStepWorkflow(timeout=10, verbose=False)
result = await w.run()
result

The type hints matter. They ensure the workflow runs in the right order, and the same mechanism lets you build branches, loops and joins^[1].

Step 3: Loops

Using the union operator | in a type hint creates a loop. In the example below, the input of step_one can be StartEvent or LoopEvent, and its output can be ProcessingEvent or LoopEvent. The example uses a random number to simulate success or failure: on failure the step returns a LoopEvent and the flow goes back to the first step; on success it moves on to the second step^[1]:

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent | LoopEvent) -> ProcessingEvent | LoopEvent:
        if random.randint(0, 1) == 0:
            print("Bad thing happened")
            return LoopEvent(loop_output="Back to step one.")
        else:
            print("Good thing happened")
            return ProcessingEvent(intermediate_result="First step complete.")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

Only the ProcessingEvent path reaches step_two. The LoopEvent path sends the flow back to step_one, so each run either finishes or tries again^[1].

Step 4: Draw the workflow

draw_all_possible_flows draws a workflow and saves it as an HTML file. The example writes to flow.html^[1]:

from llama_index.utils.workflow import draw_all_possible_flows

w = ... # as defined in the previous section
draw_all_possible_flows(w, "flow.html")

Sharing state between steps

When every step needs to read and write the same data, use state management. Add the Context type hint to a parameter of the step function, then store and read values with ctx.store.set and ctx.store.get^[1]:

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


@step
async def query(self, ctx: Context, ev: StartEvent) -> StopEvent:
    # store query in the context
    await ctx.store.set("query", "What is the capital of France?")

    # do something with context and event
    val = ...

    # retrieve query from the context
    query = await ctx.store.get("query")

    return StopEvent(result=val)

The LlamaIndex documentation covers the more complex nuances of workflows^[1].

Multi-agent workflows with AgentWorkflow

Instead of writing each step by hand, you can use the AgentWorkflow class to create a multi-agent workflow. Agents collaborate according to their specialised capabilities and can hand tasks off to one another^[1]. Import the agent classes from llama_index.core.agent.workflow, not from llama_index.core.agent^[1].

You must name one agent as the root agent in the AgentWorkflow constructor. A user message is routed to the root agent first. The root agent can handle the request with its own tools, hand it off to a better-suited agent, or return a response to the user^[1].

The example defines two ReActAgents, one for multiplication and one for addition. Each agent is given a name, a description of what it can do, a system_prompt, its tools and an llm^[1]. The tools are plain functions. You do not need to wrap them in FunctionTool, because LlamaIndex reads the function name and docstring for the name and description^[1]:

from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent
from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

# Define some tools
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")

# we can pass functions directly without FunctionTool -- the fn/docstring are parsed for the name/description
multiply_agent = ReActAgent(
    name="multiply_agent",
    description="Is able to multiply two integers",
    system_prompt="A helpful assistant that can use a tool to multiply numbers.",
    tools=[multiply],
    llm=llm,
)

addition_agent = ReActAgent(
    name="add_agent",
    description="Is able to add two integers",
    system_prompt="A helpful assistant that can use a tool to add numbers.",
    tools=[add],
    llm=llm,
)

# Create the workflow
workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
)

# Run the system
response = await workflow.run(user_msg="Can you add 5 and 3?")

Adding shared state to agents

AgentWorkflow accepts an initial_state dict, which is available to all agents. The state is stored under the state key of the workflow context, and it is injected into the state_prompt, which augments each new user message^[1].

The next example adds a counter to both tools. Each call increases num_fn_calls by one:

from llama_index.core.workflow import Context

# Define some tools
async def add(ctx: Context, a: int, b: int) -> int:
    """Add two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a + b

async def multiply(ctx: Context, a: int, b: int) -> int:
    """Multiply two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a * b

workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
    initial_state={"num_fn_calls": 0},
    state_prompt="Current state: {state}. User message: {msg}",
)

# run the workflow with context
ctx = Context(workflow)
response = await workflow.run(user_msg="Can you add 5 and 3?", ctx=ctx)

# pull out and inspect the state
state = await ctx.store.get("state")
print(state["num_fn_calls"])

After the run, reading the state through the same Context shows how many function calls were made^[1]. Because the initial state is available to all agents, a counter that one tool updates is visible to the whole workflow^[1].

Next steps

Start with a single-step workflow, then add events, branches and state, and finally hand the agents to AgentWorkflow. For finer control, the LlamaIndex documentation on workflows is the next place to read^[1].

Adapted from unit 2 of Hugging Face's open-source course *Agents Course*, which covers workflows in LlamaIndex^[1], released under the Apache-2.0 License.