LlamaIndex Workflows — Event-Driven Agentic RAG

Core Concepts

Foundational components of event-driven query and document orchestration.

Event-Driven Control

Runs on an event bus pattern. Steps yield event objects that automatically trigger subsequent tasks, allowing non-linear flows and concurrency loops.

Shared Context memory

Use the workflow context object (`ctx`) to read or write keys (`ctx.set`/`ctx.get`) across separate step executions safely.

Parallel Execution

Spawn multiple steps in parallel simply by mapping decorators to the same input Event. The workflow runs them concurrently.

Custom Event Streams

Stream intermediate milestones from steps using `ctx.send_event(Event)`. Useful for displaying retrieval progress to frontends.

Human-in-the-Loop Gating

Pause step executions to wait for external approval using `await ctx.get_input("key")`, continuing logic once payload is submitted.

Error Recovery Loops

If retrieved document context is graded insufficient by a step, the workflow automatically fires a query-refinement event to search again.

Setup & Installation

Configure LlamaIndex workflows for event-driven search routines.

Install Core packages

                    # Install the core llama-index library containing Workflows

                    pip install llama-index-core

                    # Install OpenAI model integrations and vector store clients

                    pip install llama-index-llms-openai llama-index-embeddings-openai

Event-Driven Workflow Code

Defining custom Events, context variables, parallel step runs, streaming progress, and launching the thread.

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

                    from llama_index.llms.openai import OpenAI

                    import asyncio

                    # Define custom event messaging structures

                    class QueryEvent(Event):

                        query: str

                    class ProgressEvent(Event):

                        message: str

                    class AgenticRAGWorkflow(Workflow):

                        def __init__(self, *args, **kwargs):

                            super().__init__(*args, **kwargs)

                            self.llm = OpenAI(model="gpt-4o")

                        # Step 1: Initialize context memory and trigger downstream parallel steps

                        @step

                        async def init_state(self, ctx: Context, ev: StartEvent) -> QueryEvent:

                            user_query = ev.get("query")

                            await ctx.set("original_query", user_query)

                            ctx.write_event_to_stream(ProgressEvent(message="Initializing query state..."))

                            return QueryEvent(query=user_query)

                        # Step 2: Runs in parallel with retrieve_vector (simulated)

                        @step

                        async def retrieve_sql(self, ctx: Context, ev: QueryEvent) -> Event:

                            ctx.write_event_to_stream(ProgressEvent(message="Scanning relational logs..."))

                            await asyncio.sleep(0.5) # Non-blocking async sleep

                            await ctx.set("sql_context", "task tx_992 failure duplicate primary key constraint.")

                            return Event() # Empty event notifying execution progress

                        # Step 3: Synthesis aggregates shared context variables

                        @step

                        async def synthesize(self, ctx: Context, ev: Event) -> StopEvent:

                            # Wait until SQL context has been loaded by concurrent step

                            sql_context = await ctx.get("sql_context", default=None)

                            if not sql_context:

                                ctx.write_event_to_stream(ProgressEvent(message="Waiting on parallel retrieve steps..."))

                                return None # Halts current thread, waits for next event trigger

                            query = await ctx.get("original_query")

                            prompt = f"Query: {query}\nContext: {sql_context}\nSummarize failure reasons."

                            response = await self.llm.acomplete(prompt)

                            return StopEvent(result=str(response))

                    async def main():

                        w = AgenticRAGWorkflow(timeout=10, verbose=False)

                        # Start runner and listen to custom progress stream events

                        handler = w.run(query="Why did deploy task tx_992 fail?")

                        async def read_events():

                            async for event in handler.stream_events():

                                if isinstance(event, ProgressEvent):

                                    print(f"[*] Progress: {event.message}")

                        _, result = await asyncio.gather(read_events(), handler)

                        print(f"Final Output: {result}")

Frequently Asked Questions

What are the key benefits of LlamaIndex Workflows over QueryEngines?▶

Standard QueryEngines are linear (Retrieve -> Post-process -> Synthesize). Workflows allow you to inject conditional routing, loop cycles (e.g. rewrite query and retrieve again if context evaluation fails), and concurrent execution loops natively.

Can I deploy LlamaIndex Workflows as microservices?▶

Yes, because Workflows run on a standard async event loop in Python, they are easily wrapped in FastAPI endpoints, Celery workers, or deployed serverless inside cloud container environments.

How do steps share memory or state in LlamaIndex Workflows?▶

Steps share state via the `Context` object parameter (`ctx`). Inside step methods, you can set values using `await ctx.set("key", value)` and retrieve them in other steps via `await ctx.get("key")`, keeping state access secure and asynchronous.

Can I run steps in parallel in a Workflow?▶

Yes. If multiple steps are decorated to listen to the same incoming Event (e.g., `@step` listening to `QueryEvent`), they will automatically run concurrently in separate async tasks when that event is fired.

How do I stream intermediate progress events from a Workflow?▶

You can write events to the stream buffer inside steps using `ctx.write_event_to_stream(Event)`. The client starting the run can capture these events asynchronously by looping through the handler using `async for event in handler.stream_events()`.

How does human-in-the-loop (HITL) checks work in a Workflow?▶

A step can halt its progress and wait for manual user input by invoking `await ctx.get_input("input_key")`. The workflow server halts execution at that step until an external client API fires the matching input payload event back into the workflow execution handler.