Production RAG Agent in Python: Inngest + Qdrant + LlamaIndex

Introduction

Most AI tutorials show you how to build a RAG app that works on your laptop in a happy-path demo. Then real users hit it, the OpenAI API throws a rate limit, a background ingestion job crashes mid-run, a request times out, and the whole thing falls over. The production stuff — observability, retries, throttling, queueing, rate limiting — almost always gets skipped.

This build covers exactly those production concerns. The stack: Python + FastAPI + Streamlit for the app, Qdrant for the vector database, LlamaIndex for PDF ingestion, OpenAI for embeddings and generation, and Inngest for orchestration, retries, throttling, and observability. Free or near-free to run; production-shaped from day one.

📚 Table of contents

What separates a demo RAG app from a production one

Anyone can put together a chatbot that answers questions about a PDF. The things that go wrong once it’s exposed to real users:

• OpenAI rate limits. A single user uploads a 500-page PDF; you make 1,000 embedding calls; you hit the per-minute cap; the ingest dies halfway.
• Network failures. One in 200 API calls fails for transient reasons; without retries, that’s your overall failure rate.
• Background work blocking the UI. Ingestion takes 2 minutes; if it runs inline, the user’s browser times out.
• Lack of visibility. A job failed yesterday; without observability, you find out from a customer complaint.
• Duplicate processing. User clicks “upload” twice; without idempotency, you embed everything twice and double the cost.

Inngest fixes most of those concerns out of the box without you writing a custom queue, worker pool, retry loop, or observability dashboard. That’s why it’s the centerpiece of this build.

The stack and why each piece is here

• FastAPI — the API layer. Routes get wrapped as Inngest functions.
• Inngest — orchestration. Every endpoint that’s an Inngest function gets retries, throttling, scheduling, observability for free.
• LlamaIndex — PDF (and other document type) ingestion with chunking strategies you don’t have to invent.
• Qdrant — vector database. Runs locally via Docker for development; cloud-hosted for production.
• OpenAI — embeddings (text-embedding-3-large, 3,072 dimensions) and chat completion. Swap any provider that’s OpenAI-API-compatible.
• Streamlit — minimal frontend so you can test end-to-end without a separate React build.

Project setup with uv

mkdir prod-rag && cd prod-rag
uv init .
uv add fastapi uvicorn[standard] streamlit python-dotenv pydantic
uv add inngest qdrant-client llama-index openai
uv add llama-index-readers-file

Add OpenAI key to .env. Inngest’s dev server runs as a separate process (npx inngest-cli@latest dev) — we’ll start it later.

Qdrant: the local vector database

Qdrant runs as a single Docker container for development:

docker run -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant

from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams, PointStruct

DIM = 3072  # text-embedding-3-large

class VectorStore:
    def __init__(self, collection: str = "docs"):
        self.client = QdrantClient("localhost", port=6333)
        self.collection = collection
        if not self.client.collection_exists(collection):
            self.client.create_collection(
                collection_name=collection,
                vectors_config=VectorParams(size=DIM, distance=Distance.COSINE),
            )

    def upsert(self, ids, vectors, payloads):
        points = [
            PointStruct(id=i, vector=v, payload=p)
            for i, v, p in zip(ids, vectors, payloads)
        ]
        self.client.upsert(self.collection, points=points)

    def search(self, query_vector, top_k: int = 5):
        return self.client.search(
            collection_name=self.collection,
            query_vector=query_vector,
            limit=top_k,
        )

FastAPI endpoints with Inngest orchestration

Each Inngest function is a Python function decorated with @inngest_client.create_function. Inngest serves them through a single FastAPI endpoint that the dev server discovers:

from dotenv import load_dotenv
load_dotenv()

from fastapi import FastAPI
import inngest
import inngest.fast_api

app = FastAPI()
inngest_client = inngest.Inngest(app_id="prod-rag")

@inngest_client.create_function(
    fn_id="ingest-pdf",
    trigger=inngest.TriggerEvent(event="rag/ingest.pdf"),
    retries=3,
    throttle=inngest.Throttle(limit=10, period=inngest.Duration(minutes=1)),
)
async def ingest_pdf(ctx: inngest.Context, step: inngest.Step):
    file_path = ctx.event.data["file_path"]
    chunks = await step.run("load-and-chunk", lambda: load_and_chunk(file_path))
    vectors = await step.run("embed", lambda: embed_all(chunks))
    return await step.run("upsert", lambda: vector_store.upsert_batch(chunks, vectors))

inngest.fast_api.serve(app, inngest_client, [ingest_pdf])

Three things this gives you for free:

• Retries — failed steps are retried automatically with exponential backoff.
• Throttling — max 10 ingests per minute, queued thereafter.
• Step-level observability — each step.run is independently logged and replayable.

Ingesting PDFs with LlamaIndex

from llama_index.core import SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)

def load_and_chunk(file_path: str):
    docs = SimpleDirectoryReader(input_files=[file_path]).load_data()
    nodes = splitter.get_nodes_from_documents(docs)
    return [
        {"id": n.id_, "text": n.text, "metadata": n.metadata}
        for n in nodes
    ]

SentenceSplitter handles the heavy lifting — chunk size 512 tokens with 50-token overlap is a solid default for technical documents. Tune for your content type.

The embedding + upsert pipeline

from openai import OpenAI

oai = OpenAI()

def embed_batch(texts: list[str]) -> list[list[float]]:
    res = oai.embeddings.create(
        model="text-embedding-3-large",
        input=texts,
        dimensions=3072,
    )
    return [item.embedding for item in res.data]

def embed_all(chunks: list[dict]) -> list[list[float]]:
    BATCH = 50  # OpenAI accepts up to ~2048 inputs per call; 50 is a safe default
    vectors = []
    for i in range(0, len(chunks), BATCH):
        vectors.extend(embed_batch([c["text"] for c in chunks[i:i+BATCH]]))
    return vectors

Batching saves money and time. Wrapping the call inside an Inngest step.run means a transient OpenAI 429 retries automatically without crashing the whole ingest.

Vector search with retries and rate limiting

Same pattern for the query side. Wrap the search function as an Inngest function with retries and a higher throttle for read traffic:

@inngest_client.create_function(
    fn_id="rag-query",
    trigger=inngest.TriggerEvent(event="rag/query"),
    retries=2,
    throttle=inngest.Throttle(limit=120, period=inngest.Duration(minutes=1)),
)
async def rag_query(ctx, step):
    question = ctx.event.data["question"]
    qvec = await step.run("embed-query", lambda: embed_batch([question])[0])
    hits = await step.run("search", lambda: vector_store.search(qvec, top_k=5))
    context = "\n\n".join(h.payload["text"] for h in hits)
    answer = await step.run("answer", lambda: complete(question, context))
    return {"answer": answer, "sources": [h.payload.get("source") for h in hits]}

def complete(question: str, context: str) -> str:
    res = oai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Answer using only the context. If unsure, say so."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return res.choices[0].message.content

The Streamlit frontend

import streamlit as st, requests

INNGEST_EVENT_URL = "http://localhost:8288/e/local"

st.title("Production RAG")
pdf = st.file_uploader("Upload a PDF")
if pdf and st.button("Ingest"):
    path = f"/tmp/{pdf.name}"
    open(path, "wb").write(pdf.read())
    requests.post(INNGEST_EVENT_URL,
                  json={"name": "rag/ingest.pdf", "data": {"file_path": path}})
    st.success("Ingestion queued. Watch Inngest dashboard for progress.")

q = st.text_input("Ask a question")
if q and st.button("Ask"):
    res = requests.post(INNGEST_EVENT_URL,
                        json={"name": "rag/query", "data": {"question": q}})
    # In a real app, poll for completion or use Inngest's send-and-await

For production, you’d add a polling or webhook flow to surface the answer back to the user. Streamlit is the prototype here; React + WebSockets is the production version.

Observability: Inngest’s dev server

Start the Inngest dev server in a separate terminal:

npx inngest-cli@latest dev -u http://localhost:8000/api/inngest --no-discovery

Open localhost:8288 in your browser. You get:

• A live event stream — every triggered function and its data
• Per-function execution history with each step visible
• Failed-run details with stack traces
• Manual retry / replay for any past run
• Throttle status (how many runs are queued behind the rate limit)

Production Inngest (hosted) gives the same UX with real persistence and cross-deploy visibility.

❌ Common mistakes

💡 Pro tips

Conclusion

The gap between “works in a Jupyter notebook” and “survives production traffic” is huge for AI apps, and most of the survivors look like this build: a real queue/orchestrator (Inngest), a real vector DB (Qdrant), proper retries and throttling, and observability you didn’t have to build yourself.

This same architecture scales from a single PDF demo to thousands of concurrent users with minor changes: hosted Qdrant, hosted Inngest, hosted Postgres for metadata. The hard part isn’t making it work; it’s knowing what to build in from day one.

Explore More on DevShelf

• →
  Traditional RAG vs Vectorless RAG

  The architectural decision before you build — when Qdrant vector search wins and when an LLM-tree approach is the better call.

• →
  PG Text Search: BM25 to Replace Elasticsearch

  The hybrid keyword + vector pattern that pairs with Qdrant — how BM25 ranking in Postgres fills the gaps pure vector search misses.