Orchestrate RAG with LlamaIndex Workflows

Introduction

When you wire retrieval, reranking, and generation as a single hardcoded async function, every change—adding a quality gate, swapping the reranker, retrying a failed call—forces you to re-read the whole pipeline and risks silently breaking the parts you didn't mean to touch. The cost shows up in production as latency regressions and answer-quality drops that no single span can explain. By the end of this lesson you'll be able to orchestrate a hybrid RAG pipeline as a LlamaIndex Workflow of typed, event-driven steps, run retrieval paths concurrently, and trace every step end-to-end in OpenTelemetry.

Key Terminology

• Workflow: A LlamaIndex orchestration primitive that wires @step coroutines together via typed Event classes; the engine routes events by class, handles concurrency, and validates the step graph at construction time.
• @step: Decorator that marks an async method as a workflow node; its type-annotated input event determines when it fires, and its return type declares the event it emits to downstream consumers.
• StartEvent / StopEvent: The built-in entry and exit events for a Workflow. StartEvent carries the initial query into the first step; StopEvent.result is the dictionary returned to the caller when the pipeline terminates.
• Reciprocal Rank Fusion (RRF): A rank-aggregation technique that merges the semantic and BM25 result lists by summing 1 / (k + rank_i) per document, producing a single fused ordering the cross-encoder reranker scores.
• Cross-encoder reranker: A model that jointly encodes (query, document) pairs to produce a precise relevance score, used after RRF to trim the fused candidate set down to the top_n documents fed to the LLM.

Concepts

Why event-driven orchestration matters for hybrid RAG

A hybrid RAG pipeline involves query embedding, dual-path retrieval, score fusion/reranking, and augmented generation—each with different latency profiles and failure modes. When hardcoded in a single async function, you cannot:

• Run semantic and BM25 retrieval concurrently and merge results at a synchronization point
• Retry a failed reranker call without re-executing retrieval
• Insert a quality gate between reranking and generation that conditionally loops back to retrieval with a reformulated query
• Trace each operation independently in your observability stack

LlamaIndex Workflows address all four constraints. Each @step is an isolated async coroutine that activates when its declared input event arrives. Steps emit output events that route to downstream steps automatically. The workflow engine handles concurrency, fan-out, fan-in, and conditional branching—all through the type system.

Trace analysis for production debugging

Once traces flow into your collector, you can diagnose common hybrid RAG issues by examining span durations and attributes:

• Slow retrieval spans: If retrieve_step consistently exceeds 500ms, check whether your pgvector HNSW index has sufficient ef_search or whether the BM25 tsvector column lacks a GIN index. The semantic_count and bm25_count attributes reveal if one path returns zero results, indicating an indexing gap.
• Reranker bottleneck: Cross-encoder reranking is compute-intensive. If rerank_step dominates latency, reduce the fused_count entering the reranker by tightening the RRF cutoff or pre-filtering by a minimum BM25 score threshold.
• Token budget overruns: The prompt_tokens attribute on generate_step lets you alert when context exceeds your model's window. A sudden spike means the reranker returned unusually long documents—add a truncation policy or increase top_n filtering.
• End-to-end latency distribution: The root span for the entire workflow shows P50/P95/P99 latencies. If P99 exceeds your SLA, drill into child spans to isolate whether the tail comes from LLM cold starts, database connection pool exhaustion, or reranker GPU contention.

Connecting to the broader pipeline

This workflow sits between the document ingestion layer (Crawl4AI + Unstructured chunking into pgvector) and the agentic RAG layer (Pydantic AI agent that decides whether to accept the response or reformulate the query). The StopEvent result dictionary feeds directly into the agent's evaluation logic. If RAGAS faithfulness or answer relevancy scores fall below threshold, the agent emits a new query that re-enters this workflow as a fresh StartEvent—creating the iterative retrieval loop covered in another goal.

The typed event architecture makes this integration clean: the agent does not need to know about internal pipeline steps. It sends a query string and receives a dictionary with response, citations, and query. If you later add a caching step between StartEvent and retrieve, or swap BM25 for a learned sparse retrieval model, the agent's interface remains unchanged. This is the core value of event-driven orchestration—each step is an independent unit of work with a typed contract, and the workflow engine handles routing, concurrency, and observability automatically.

Code Walkthrough

Core architecture: events as typed contracts

The foundation of a LlamaIndex Workflow is the Event class hierarchy. Every piece of data flowing between steps is wrapped in a typed event. This is not merely a naming convention—the workflow engine uses these types to build the execution graph at initialization time, validate that every emitted event has a consumer, and provide type-safe access to payloads.

Loading diagram...

Each arrow in this diagram represents a typed contract. The retrieve_step method declares it consumes StartEvent and emits RetrieveEvent. If you accidentally emit a RerankEvent from retrieve_step, downstream steps still receive the correct type—the engine routes by event class, not by emission order. This means you can later insert a QualityGateEvent between RerankEvent and GenerateEvent without modifying the existing steps.

Defining event classes and step methods

The following code defines the custom event classes—RetrieveEvent, RerankEvent, GenerateEvent—and the RAGWorkflow itself with all four typed @step methods. Each event inherits from Event and uses a dataclass so the workflow engine can validate the graph at construction time and propagate the original query downstream without shared mutable state. The RAGWorkflow init accepts the retriever, reranker, and LLM via dependency injection; each step then wraps its logic in an OpenTelemetry span so the semantic_count, fused_count, prompt_tokens, and other attributes introduced in Concepts become live signals you can query in your tracing backend.

1import asyncio
2from dataclasses import dataclass, field
3from llama_index.core.workflow import Workflow, Event, StartEvent, StopEvent, step
4from opentelemetry import trace
5
6tracer = trace.get_tracer("rag_workflow")
7
8@dataclass
9class RetrieveEvent(Event):
10    """Carries results from dual-path hybrid retrieval."""
11    query: str = ""
12    semantic_docs: list = field(default_factory=list)
13    bm25_docs: list = field(default_factory=list)
14
15@dataclass
16class RerankEvent(Event):
17    """Carries fused and reranked documents with scores."""
18    query: str = ""
19    ranked_docs: list = field(default_factory=list)
20    scores: list[float] = field(default_factory=list)
21
22@dataclass
23class GenerateEvent(Event):
24    """Carries the LLM response and supporting documents."""
25    response: str = ""
26    docs: list = field(default_factory=list)
27    query: str = ""
28
29class RAGWorkflow(Workflow):
30    """Event-driven hybrid RAG pipeline with four typed steps."""
31
32    def __init__(self, retriever, reranker, llm, timeout: int = 120):
33        super().__init__(timeout=timeout)
34        self.retriever = retriever
35        self.reranker = reranker
36        self.llm = llm
37
38    @step
39    async def retrieve(self, event: StartEvent) -> RetrieveEvent:
40        with tracer.start_as_current_span("retrieve_step") as span:
41            query = event.query
42            span.set_attribute("query", query)
43            semantic_task = self.retriever.asemantic_search(query, top_k=20)
44            bm25_task = self.retriever.abm25_search(query, top_k=20)
45            semantic_docs, bm25_docs = await asyncio.gather(
46                semantic_task, bm25_task
47            )
48            span.set_attribute("semantic_count", len(semantic_docs))
49            span.set_attribute("bm25_count", len(bm25_docs))
50            return RetrieveEvent(
51                query=query,
52                semantic_docs=semantic_docs,
53                bm25_docs=bm25_docs,
54            )
55
56    @step
57    async def rerank(self, event: RetrieveEvent) -> RerankEvent:
58        with tracer.start_as_current_span("rerank_step") as span:
59            fused = self._reciprocal_rank_fusion(
60                event.semantic_docs, event.bm25_docs, k=60
61            )
62            ranked = await self.reranker.arerank(event.query, fused, top_n=5)
63            scores = [doc.score for doc in ranked]
64            span.set_attribute("fused_count", len(fused))
65            span.set_attribute("reranked_count", len(ranked))
66            return RerankEvent(
67                query=event.query, ranked_docs=ranked, scores=scores
68            )
69
70    @step
71    async def generate(self, event: RerankEvent) -> GenerateEvent:
72        with tracer.start_as_current_span("generate_step") as span:
73            context = "\n\n".join(
74                f"[{i+1}] {d.text}" for i, d in enumerate(event.ranked_docs)
75            )
76            prompt = (
77                f"Answer using ONLY the provided sources.\n\n"
78                f"{context}\n\nQuestion: {event.query}"
79            )
80            response = await self.llm.acomplete(prompt)
81            span.set_attribute("prompt_tokens", len(prompt.split()))
82            return GenerateEvent(
83                response=str(response),
84                docs=event.ranked_docs,
85                query=event.query,
86            )
87
88    @step
89    async def cite(self, event: GenerateEvent) -> StopEvent:
90        with tracer.start_as_current_span("cite_step"):
91            citations = [
92                {"id": i + 1, "source": d.metadata.get("source", "unknown")}
93                for i, d in enumerate(event.docs)
94            ]
95            return StopEvent(result={
96                "response": event.response,
97                "citations": citations,
98                "query": event.query,
99            })

Instrumenting with OpenTelemetry

Each @step method wraps its logic in a tracer.start_as_current_span() context manager. This creates a child span under the workflow's root span, building a trace tree that mirrors the event flow. To activate tracing, configure the OpenTelemetry SDK at application startup before instantiating the workflow. The following snippet demonstrates configuring a BatchSpanProcessor with an OTLP exporter that sends traces to a local collector endpoint. The Resource object tags all spans with service metadata, enabling you to filter traces by service name in Jaeger or Grafana Tempo. The TracerProvider is set as the global provider so that both your workflow code and any LlamaIndex-internal instrumentation emit spans into the same trace.

1from opentelemetry.sdk.trace import TracerProvider
2from opentelemetry.sdk.trace.export import BatchSpanProcessor
3from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
4from opentelemetry.sdk.resources import Resource
5
6resource = Resource.create({"service.name": "hybrid-rag-pipeline"})
7provider = TracerProvider(resource=resource)
8exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
9provider.add_span_processor(BatchSpanProcessor(exporter))
10trace.set_tracer_provider(provider)
11
12async def run_rag_pipeline(query: str, retriever, reranker, llm):
13    workflow = RAGWorkflow(
14        retriever=retriever, reranker=reranker, llm=llm, timeout=60
15    )
16    result = await workflow.run(query=query)
17    return result

Do's and Don'ts

Do's

1. ✓Do propagate query as a named field on every @dataclass Event subclass — RetrieveEvent, RerankEvent, and GenerateEvent each carry query: str so every step receives full context from its input event alone, with no shared mutable state on RAGWorkflow; this is what allows the engine to run concurrent workflow instances without cross-contaminating in-flight queries.
2. ✓Do fan out asemantic_search and abm25_search with asyncio.gather() inside the retrieve step — creating both coroutines as tasks and gathering them runs the dual retrieval paths in parallel; awaiting them sequentially doubles retrieval wall-clock time and erases the main latency benefit of structuring the pipeline as discrete @step methods.
3. ✓Do wrap every @step body in tracer.start_as_current_span() and attach span attributes like semantic_count, fused_count, and prompt_tokens — without a named span per step, a latency regression or answer-quality drop produces no attributable cause across the four-step chain, which is precisely the production failure mode the Introduction identifies as the cost of a monolithic pipeline.

Don'ts

1. ✗Don't assume the workflow engine routes steps by emission order — the engine matches event class at graph-build time, so emitting a RerankEvent from the retrieve step instead of a RetrieveEvent leaves the rerank step permanently starved with no runtime error; the mismatched type is silently dropped because no registered consumer matches it.
2. ✗Don't store per-run intermediate results as instance attributes on RAGWorkflow — __init__ accepts retriever, reranker, and llm as read-only collaborators via dependency injection; writing self.current_docs or self.last_query during a run means concurrent workflow executions overwrite each other's data, defeating the isolation that event fields exist to enforce.
3. ✗Don't define custom Event subclasses without the @dataclass decorator and field(default_factory=list) — all three event classes in the walkthrough use @dataclass so the workflow engine can validate the execution graph at construction time; omitting it means list-typed fields like semantic_docs and ranked_docs share a single default object across instances, producing cross-run data corruption that only manifests on the second workflow invocation.