Last month, I was building a document processing pipeline for a client who needed to extract data from insurance claims, cross-reference it against policy documents, and flag potential fraud risks. I started with a straightforward chain of LLM calls—extract, then validate, then assess. It worked fine for the happy path. But the moment I needed to loop back for missing information, or route different document types through different validation steps, my carefully constructed chain turned into a tangled mess of conditional logic and nested callbacks.
That's when I stumbled onto LlamaIndex Workflows. Unlike traditional chains that force you into a rigid A→B→C execution model, Workflows use an event-driven architecture. Steps emit events, and other steps listen for those events and react. Need a loop? Just have a step emit an event that triggers an earlier step. Need conditional branching? Emit different events based on your logic. It's a small conceptual shift that completely changes how you structure complex applications.
Let me walk you through how I actually got started with it, including the parts that tripped me up.
The Core Concept: Events and Steps
Before we write any code, you need to understand the two building blocks:
- Events are data containers. They carry information between steps.
- Steps are functions decorated with
@stepthat listen for specific events and emit other events.
Think of it like a radio broadcast system. A step tunes into a specific frequency (event type), processes what it hears, and broadcasts on another frequency. Multiple steps can listen to the same event, and a step can emit multiple different events.
Setting Up the Environment
First, install the core package. I'm using OpenAI here, but you can swap in any provider:
pip install "llama-index-core>=0.11.16" llama-index-utils-workflow
pip install llama-index-llms-openai
You'll also need your API key set up:
import os
os.environ["OPENAI_API_KEY"] = "your-key-here"
Building a Practical RAG Workflow with Reranking
Let's build something real: a RAG pipeline that retrieves documents, reranks them for relevance, and synthesizes an answer. I chose this because it's a workflow everyone understands, but it already benefits from the event-driven model—especially when you want to add retry logic or conditional routing later.
Step 1: Define Your Events
Events are just Pydantic models. This was actually one of the first things that confused me—I kept trying to pass raw dictionaries between steps. You must define explicit event classes:
from llama_index.core.workflow import StartEvent, StopEvent, Event
class RetrieveEvent(Event):
query: str
class RerankEvent(Event):
query: str
documents: list
class SynthesizeEvent(Event):
query: str
ranked_documents: list
StartEvent and StopEvent are built-in. StartEvent is what kicks everything off when you call workflow.run(), and StopEvent is what terminates the workflow and returns a result.
Step 2: Define Your Workflow Class and Steps
Now we create the workflow class and add steps. Each step uses type hints to declare which event it listens for and which events it can emit:
from llama_index.core.workflow import Workflow, step
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.postprocessor import SentenceTransformerRerank
class RAGWorkflow(Workflow):
@step
async def retrieve(self, ev: StartEvent) -> RetrieveEvent:
query = ev.get("query", "")
if not query:
raise ValueError("Query is required!")
# Load and index documents (in production, you'd pre-build this)
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
retriever = index.as_retriever(similarity_top_k=10)
nodes = retriever.retrieve(query)
documents = [node.node.get_content() for node in nodes]
return RetrieveEvent(query=query, documents=documents)
@step
async def rerank(self, ev: RetrieveEvent) -> RerankEvent:
# Use a reranker to narrow down results
reranker = SentenceTransformerRerank(
model="cross-encoder/ms-marco-MiniLM-L-2-v2",
top_n=3
)
# In a real implementation, you'd pass the actual NodeWithScore objects
# For simplicity, we're just passing the top documents forward
ranked_docs = ev.documents[:3] # Simplified for demo
return RerankEvent(query=ev.query, ranked_documents=ranked_docs)
@step
async def synthesize(self, ev: RerankEvent) -> StopEvent:
from llama_index.llms.openai import OpenAI
llm = OpenAI(model="gpt-4o")
context = "\n\n---\n\n".join(ev.ranked_documents)
prompt = f"""Based on the following context, answer the question.
Context:
{context}
Question: {ev.query}
Answer:"""
response = llm.complete(prompt)
return StopEvent(result=str(response))
Step 3: Run It
async def main():
workflow = RAGWorkflow(timeout=60, verbose=True)
result = await workflow.run(query="What are the key risk factors mentioned in the documents?")
print(result)
# In a Jupyter notebook:
import asyncio
asyncio.run(main())
The verbose=True flag is incredibly helpful during development. It prints out a visual trace of which events trigger which steps, so you can see the flow executing in real time.
Where I Got Stuck (And What I Learned)
Mistake #1: Forgetting to declare emitted event types. The @step decorator uses type hints to build a graph of your workflow. If your step can emit different events conditionally, you need to use a union type:
@step
async def validate(self, ev: RetrieveEvent) -> RerankEvent | StopEvent:
if not ev.documents:
return StopEvent(result="No documents found")
return RerankEvent(query=ev.query, documents=ev.documents)
I initially didn't do this and got cryptic errors about missing event handlers. The framework needs to know all possible emissions at parse time to validate the workflow graph.
Mistake #2: Trying to share state between steps with global variables. Workflows are designed to be stateless between steps. If you need to persist data across steps, use the Context object:
class MyWorkflow(Workflow):
@step
async def step_one(self, ev: StartEvent, ctx: Context) -> RetrieveEvent:
await ctx.set("original_query", ev.get("query"))
return RetrieveEvent(query=ev.get("query"))
@step
async def step_two(self, ev: RerankEvent, ctx: Context) -> StopEvent:
# Retrieve shared state
original_query = await ctx.get("original_query")
return StopEvent(result=f"Processed: {original_query}")
Notice the ctx: Context parameter. The framework injects it automatically. This pattern is essential for anything beyond trivial workflows.
Mistake #3: Not handling async properly. Workflows are inherently asynchronous. If you're running in a Jupyter notebook, you might need nest_asyncio to avoid event loop conflicts:
import nest_asyncio
nest_asyncio.apply()
Adding Loops: The Real Power
Here's where Workflows genuinely outshine linear chains. Let's say I want my RAG pipeline to retry retrieval if the initial results seem irrelevant:
class RetryEvent(Event):
query: str
attempt: int
class RAGWorkflowWithRetry(Workflow):
@step
async def retrieve(self, ev: StartEvent | RetryEvent, ctx: Context) -> RetrieveEvent | StopEvent:
query = ev.get("query", "") if isinstance(ev, StartEvent) else ev.query
attempt = await ctx.get("attempt", default=0) + 1
await ctx.set("attempt", attempt)
if attempt > 3:
return StopEvent(result="Could not find relevant documents after 3 attempts")
# ... retrieval logic ...
nodes = retriever.retrieve(query)
# Check relevance - if poor, loop back
if self._is_poor_retrieval(nodes):
return RetryEvent(query=query, attempt=attempt)
return RetrieveEvent(query=query, documents=[n.node.get_content() for n in nodes])
def _is_poor_retrieval(self, nodes):
# Simple heuristic: check if top scores are below threshold
return all(n.score < 0.5 for n in nodes) if nodes else True
Try implementing that loop cleanly with a linear chain. You'd end up with while loops, state management headaches, and error-prone control flow. With Workflows, it's just emitting a different event type.
Visualizing Your Workflow
One feature I didn't appreciate until later: you can generate a visual diagram of your workflow:
from llama_index.utils.workflow import draw_all_possible_flows
draw_all_possible_flows(RAGWorkflowWithRetry, filename="workflow.html")
This generates an HTML file with an interactive diagram showing all possible paths through your workflow. When I had a bug where a step wasn't being triggered, the diagram immediately showed me I'd forgotten to include an event type in a union return. Saved me an hour of debugging.
Practical Tips and Honest Limitations
Tips:
- Always start with
verbose=Trueduring development. The event trace is your best debugging tool. - Keep steps small and focused. A step should do one thing well, just like a good function.
- Use
Contextfor any data that needs to persist across steps—don't try to cheat with instance variables. - Name your events descriptively. You'll thank yourself when reading the trace output.
Limitations I've hit:
- The learning curve for the event-driven model is real, especially if you're used to imperative chains. Expect to spend a day or two thinking "this would be simpler as a regular function" before it clicks.
- Debugging can be tricky when steps emit multiple event types. The visualization helps, but tracing through complex branching logic still requires careful attention.
- The documentation, while improving, still has gaps around advanced patterns like fan-out/fan-in (multiple steps processing the same event in parallel). I had to read source code to figure out some of these patterns.
- For very simple pipelines (literally A→B→C with no branching), Workflows add overhead compared to a basic chain. Use them when you need the flexibility, not as a default.
LlamaIndex Workflows solved my document processing problem elegantly. The insurance claim pipeline that was becoming an unmaintainable chain of conditionals is now a clean set of self-documenting steps with clear data flow. The event-driven model isn't just a different syntax—it genuinely changes how you think about structuring complex LLM applications, and once it clicks, you won't want to go back to linear chains.