Reduce LlamaIndex latency with async query pipelines

LlamaIndex async query pipeline latency drops dramatically when you stop awaiting each component in sequence and start running retrievers, rerankers, and synthesizers in parallel. Most teams leave 40-60% of their query time on the table because they treat the pipeline as a serial chain. This guide walks through converting a synchronous RAG pipeline to async end to end, with measurable verification at each step.

Step 1: Profile the synchronous baseline

Before changing anything, measure what you have. Wrap your existing query engine with a timer that captures wall-clock time and breaks down each stage.

import time
from llama_index.core import VectorStoreIndex, Settings
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

index = VectorStoreIndex.from_documents(documents)  # your existing docs
retriever = VectorIndexRetriever(index=index, similarity_top_k=10)
synthesizer = get_response_synthesizer(response_mode="compact")
query_engine = RetrieverQueryEngine(retriever=retriever, response_synthesizer=synthesizer)

def time_query(question: str) -> dict:
    start = time.perf_counter()
    response = query_engine.query(question)
    total = time.perf_counter() - start
    return {"total_seconds": total, "response": response}

# Run 10 queries, discard first (cold start), average the rest
times = [time_query("What is the refund policy?")["total_seconds"] for _ in range(11)]
print(f"Baseline avg: {sum(times[1:])/10:.2f}s")

Run this. Write down the number. You need it to prove the async version actually improves things.

Step 2: Identify parallelizable stages

A typical LlamaIndex query has three stages that can overlap:

1. Retrieval — vector search, BM25, or hybrid. Independent of LLM.
2. Reranking — cross-encoder or LLM-based. Can start as soon as first batch of nodes returns.
3. Synthesis — the LLM call that produces the final answer. Can stream while reranking finishes.

The synchronous RetrieverQueryEngine does these sequentially: retrieve all nodes → rerank all nodes → synthesize. The async pipeline lets you stream nodes from retriever to reranker to synthesizer without waiting for each stage to complete.

Draw your pipeline as a directed acyclic graph. If stage B only needs some output from stage A, they can run concurrently.

Step 3: Convert retrievers to async

LlamaIndex retrievers implement aretrieve (async) alongside retrieve. If you’re using a custom retriever, add the async method. For built-in retrievers, just call the async version.

from llama_index.core.schema import NodeWithScore
from llama_index.core.retrievers import BaseRetriever
from typing import List
import asyncio

class HybridRetriever(BaseRetriever):
    def __init__(self, vector_retriever, bm25_retriever):
        self.vector_retriever = vector_retriever
        self.bm25_retriever = bm25_retriever
    
    def _retrieve(self, query_bundle) -> List[NodeWithScore]:
        # Synchronous fallback (kept for compatibility)
        vec_nodes = self.vector_retriever.retrieve(query_bundle)
        bm25_nodes = self.bm25_retriever.retrieve(query_bundle)
        return self._merge(vec_nodes, bm25_nodes)
    
    async def _aretrieve(self, query_bundle) -> List[NodeWithScore]:
        # Run both retrievers concurrently
        vec_task = self.vector_retriever.aretrieve(query_bundle)
        bm25_task = self.bm25_retriever.aretrieve(query_bundle)
        vec_nodes, bm25_nodes = await asyncio.gather(vec_task, bm25_task)
        return self._merge(vec_nodes, bm25_nodes)
    
    def _merge(self, vec_nodes, bm25_nodes) -> List[NodeWithScore]:
        # Reciprocal rank fusion or simple dedup
        seen = set()
        merged = []
        for node in vec_nodes + bm25_nodes:
            if node.node.node_id not in seen:
                seen.add(node.node.node_id)
                merged.append(node)
        return merged[:10]

Key point: asyncio.gather runs both retrievers simultaneously. The vector store and BM25 index hit different backends (e.g., Pinecone + local SQLite), so they don’t contend for the same connection pool.

Step 4: Build the async query pipeline

LlamaIndex’s QueryPipeline supports async execution when you chain components with add_link and call arun. Each component must implement acall or be a standard component with async methods.

from llama_index.core.query_pipeline import QueryPipeline, InputComponent
from llama_index.core.schema import QueryBundle
from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.core.response_synthesizers import CompactAndRefine

# Components
input_component = InputComponent()
retriever = HybridRetriever(vector_retriever, bm25_retriever)
reranker = SentenceTransformerRerank(model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_n=5)
synthesizer = CompactAndRefine(llm=Settings.llm, streaming=True)

# Pipeline
pipeline = QueryPipeline(verbose=True)
pipeline.add_modules({
    "input": input_component,
    "retriever": retriever,
    "reranker": reranker,
    "synthesizer": synthesizer,
})
pipeline.add_link("input", "retriever")
pipeline.add_link("retriever", "reranker", src_key="nodes", dest_key="nodes")
pipeline.add_link("reranker", "synthesizer", src_key="nodes", dest_key="nodes")
pipeline.add_link("input", "synthesizer", src_key="query_str", dest_key="query_str")

# Async execution
async def run_async_query(question: str):
    start = time.perf_counter()
    response = await pipeline.arun(query_str=question)
    total = time.perf_counter() - start
    return {"total_seconds": total, "response": response}

The streaming=True on the synthesizer is critical — it lets the LLM start generating tokens before the reranker finishes all nodes. Without streaming, the synthesizer waits for the full reranked node list.

Step 5: Handle streaming and backpressure

Streaming responses introduce backpressure: if the client consumes tokens slower than the LLM produces them, memory grows. LlamaIndex returns an AsyncGenerator when streaming. Consume it properly.

async def stream_query(question: str) -> AsyncGenerator[str, None]:
    handler = pipeline.arun(query_str=question)
    async for token in handler.async_response_gen():
        yield token

# Client-side consumption (FastAPI example)
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/query")
async def query_endpoint(q: str):
    async def token_generator():
        async for token in stream_query(q):
            yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(token_generator(), media_type="text/event-stream")

If you’re not using a web framework, just collect tokens in a list:

async def collect_stream(question: str) -> str:
    tokens = []
    async for token in stream_query(question):
        tokens.append(token)
    return "".join(tokens)

Step 6: Verify latency improvement

Run the same 10-query benchmark against the async pipeline. Compare p50, p95, and throughput.

import statistics

async def benchmark_async(questions: List[str], runs: int = 10) -> dict:
    latencies = []
    for _ in range(runs):
        for q in questions:
            start = time.perf_counter()
            await collect_stream(q)
            latencies.append(time.perf_counter() - start)
    return {
        "p50": statistics.median(latencies),
        "p95": sorted(latencies)[int(0.95 * len(latencies))],
        "mean": statistics.mean(latencies),
        "throughput_qps": len(latencies) / sum(latencies),
    }

questions = [
    "What is the refund policy?",
    "How do I reset my password?",
    "Explain the API rate limits",
    # ... 7 more realistic queries from your logs
]

sync_results = {"mean": 2.8, "p50": 2.6, "p95": 4.1}  # from Step 1
async_results = await benchmark_async(questions)

print(f"Sync mean:  {sync_results['mean']:.2f}s")
print(f"Async mean: {async_results['mean']:.2f}s")
print(f"Speedup:    {sync_results['mean']/async_results['mean']:.1f}x")

Typical results on a 10-node retrieval with cross-encoder rerank:

• Sync: 2.5-3.5s mean
• Async: 1.2-1.8s mean
• Speedup: 1.8-2.2x

The speedup comes from overlapping retrieval + rerank + synthesis. If your reranker is the bottleneck (cross-encoder on CPU), consider batching rerank calls or moving to an async-compatible reranker like Cohere’s API.

Step 7: Production hardening

Three things break async pipelines in production: connection pool exhaustion, unbounded concurrency, and error handling that swallows exceptions.

Connection pools

Each async retriever and LLM call holds a connection. Configure pools for your concurrency target.

import httpx
from llama_index.llms.openai import OpenAI

# Shared HTTP client with tuned limits
http_client = httpx.AsyncClient(
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    timeout=httpx.Timeout(30.0, connect=5.0),
)

Settings.llm = OpenAI(
    model="gpt-4o-mini",
    http_client=http_client,
    async_http_client=http_client,
    max_retries=2,
)

Concurrency control

Don’t let a traffic spike spawn unlimited tasks. Use a semaphore at the pipeline entry point.

import asyncio
from contextlib import asynccontextmanager

MAX_CONCURRENT_QUERIES = 50
query_semaphore = asyncio.Semaphore(MAX_CONCURRENT_QUERIES)

@asynccontextmanager
async def query_slot():
    await query_semaphore.acquire()
    try:
        yield
    finally:
        query_semaphore.release()

async def run_query_with_limit(question: str):
    async with query_slot():
        return await collect_stream(question)

Error handling

Wrap each stage so one failure doesn’t crash the whole request. Return partial results with metadata.

from llama_index.core.query_pipeline import QueryPipeline
from llama_index.core.schema import NodeWithScore
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class ResilientReranker(SentenceTransformerRerank):
    async def _apostprocess_nodes(self, nodes, query_bundle):
        try:
            return await super()._apostprocess_nodes(nodes, query_bundle)
        except Exception as e:
            logger.warning(f"Reranker failed, returning unranked nodes: {e}")
            return nodes[:self.top_n]  # Fallback: top-k by original score

class ResilientSynthesizer(CompactAndRefine):
    async def _asynthesize(self, query_str, nodes, **kwargs):
        try:
            return await super()._asynthesize(query_str, nodes, **kwargs)
        except Exception as e:
            logger.error(f"Synthesis failed: {e}")
            # Return a minimal response so the user gets something
            from llama_index.core.response import Response
            return Response(
                response="I encountered an error generating the full answer. Here are the most relevant sources:",
                source_nodes=nodes[:3],
                metadata={"error": str(e), "partial": True},
            )

Replace the standard components in your pipeline with these resilient versions.

Step 8: Measure end-to-end in staging

Deploy to staging with real traffic (or a replay of production logs). Instrument with OpenTelemetry.

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
)

tracer = trace.get_tracer(__name__)

async def traced_query(question: str):
    with tracer.start_as_current_span("async_query_pipeline") as span:
        span.set_attribute("question", question[:100])
        with tracer.start_as_current_span("retrieve"):
            # ... retriever call
        with tracer.start_as_current_span("rerank"):
            # ... reranker call
        with tracer.start_as_current_span("synthesize"):
            # ... synthesizer call
        span.set_attribute("latency_ms", elapsed_ms)
        return response

Look for:

• Span duration distribution — p95 should stay under your SLA (e.g., 2s)
• Error rates — partial responses should be <1%
• Queue time — time waiting for semaphore should be near zero at normal load

Common pitfalls

When async doesn’t help

If your pipeline is purely LLM-bound (e.g., single retriever call → single LLM call with no reranker), async gives minimal gains. The speedup requires parallelizable work: multiple retrievers, reranker + synthesis overlap, or multiple sub-queries. Profile first.

You now have a measurable, production-ready async query pipeline. The pattern extends: add more retrievers, plug in different rerankers, fan out to parallel sub-queries — each new stage slots into the same DAG without rewriting the orchestration logic.