Retrieval-augmented generation with LangChain and Qdrant

This langchain qdrant rag tutorial walks you through building a complete retrieval-augmented generation pipeline from scratch. You’ll ingest documents, chunk them intelligently, store embeddings in Qdrant, and wire up a LangChain retrieval chain that answers questions with citations. Every step includes runnable code and a verification checkpoint so you know it works before moving on.

Step 1: Set up the environment and dependencies

Start with a clean Python 3.11+ environment. You need LangChain’s core packages, the Qdrant integration, an embedding model, and an LLM. We’ll use OpenAI for embeddings and chat, but the pattern works with any provider.

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "langchain>=0.2.0" "langchain-openai>=0.1.0" "langchain-qdrant>=0.1.0" "qdrant-client>=1.9.0" "python-dotenv>=1.0.0" "tiktoken>=0.7.0"

Create a .env file with your API keys:

OPENAI_API_KEY=sk-...
# Optional: Qdrant Cloud
# QDRANT_URL=https://your-cluster.qdrant.io
# QDRANT_API_KEY=...

Verify: Run python -c "import langchain_qdrant; print(langchain_qdrant.__version__)" — you should see a version string without import errors.

Step 2: Launch Qdrant locally (or connect to cloud)

For development, the fastest path is Qdrant’s Docker image. It runs in-memory by default, which is fine for this tutorial.

docker run -d -p 6333:6333 -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage \
  qdrant/qdrant:v1.10.0

If you prefer Qdrant Cloud, create a cluster at cloud.qdrant.io and note the URL and API key — you’ll pass those to the client instead of localhost.

Verify: Open http://localhost:6333/dashboard in a browser. You should see the Qdrant UI with “Collections” empty and a green “Ready” status.

Step 3: Load and chunk your source documents

Real-world RAG starts with document loading. LangChain supports PDFs, HTML, Markdown, Notion, and more. For this tutorial, we’ll use a few Markdown files — but the same code works for a directory of PDFs.

Create data/sample.md:

# n4n.ai Architecture Overview

## Routing Layer

The routing layer sits in front of 240+ models across 20+ providers. It accepts OpenAI-compatible requests and applies client-specified routing directives — model preferences, cost ceilings, latency budgets, and fallback chains. When a provider returns 429 or 5xx, the router automatically retries the next provider in the chain without surfacing an error to the caller.

## Usage Metering

Every request is metered at token granularity. Input tokens, output tokens, and cached tokens are recorded separately per model and provider. This enables accurate cost allocation and quota enforcement at the organization, project, or API key level.

## Cache Control

Provider cache-control hints (e.g., `Cache-Control: public, max-age=3600`) are forwarded transparently. Clients can also inject their own cache directives via headers, which the gateway respects when the underlying provider supports prompt caching.

Now load and chunk it. We’ll use RecursiveCharacterTextSplitter with token-aware sizing — this preserves semantic boundaries better than fixed-character splits.

# ingest.py
from pathlib import Path
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

DATA_DIR = Path("data")
CHUNK_SIZE = 500      # tokens
CHUNK_OVERLAP = 50    # tokens

def load_and_chunk():
    docs = []
    for md_file in DATA_DIR.glob("*.md"):
        loader = TextLoader(str(md_file), encoding="utf-8")
        docs.extend(loader.load())

    splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
        encoding_name="cl100k_base",
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
    )
    chunks = splitter.split_documents(docs)
    print(f"Loaded {len(docs)} documents, split into {len(chunks)} chunks")
    return chunks

if __name__ == "__main__":
    chunks = load_and_chunk()
    for i, chunk in enumerate(chunks[:3]):
        print(f"\n--- Chunk {i} ({len(chunk.page_content)} chars) ---")
        print(chunk.page_content[:200])

Verify: Run python ingest.py. You should see 1 document split into roughly 4–6 chunks (depending on token count), with the first few chunks printed. Each chunk should have a source metadata field pointing to the original file.

Step 4: Create the Qdrant vector store and index chunks

Now we embed the chunks and upsert them into Qdrant. LangChain’s QdrantVectorStore handles the client connection, collection creation, and batch upserts.

# index.py
import os
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
from ingest import load_and_chunk

COLLECTION_NAME = "n4n_docs"
EMBEDDING_MODEL = "text-embedding-3-small"  # 1536 dimensions
QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")

def get_embeddings():
    return OpenAIEmbeddings(model=EMBEDDING_MODEL)

def get_qdrant_client():
    if QDRANT_API_KEY:
        return QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)
    return QdrantClient(url=QDRANT_URL)

def ensure_collection(client: QdrantClient, vector_size: int):
    collections = client.get_collections().collections
    names = {c.name for c in collections}
    if COLLECTION_NAME not in names:
        client.create_collection(
            collection_name=COLLECTION_NAME,
            vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE),
        )
        print(f"Created collection '{COLLECTION_NAME}'")
    else:
        print(f"Collection '{COLLECTION_NAME}' already exists")

def index_documents():
    chunks = load_and_chunk()
    embeddings = get_embeddings()
    client = get_qdrant_client()

    # Determine vector size from a test embedding
    test_vec = embeddings.embed_query("test")
    vector_size = len(test_vec)
    ensure_collection(client, vector_size)

    vector_store = QdrantVectorStore(
        client=client,
        collection_name=COLLECTION_NAME,
        embedding=embeddings,
    )

    # Add documents in batches (default batch size is 64)
    vector_store.add_documents(chunks)
    print(f"Indexed {len(chunks)} chunks into Qdrant")

if __name__ == "__main__":
    index_documents()

Verify: Run python index.py. Then in the Qdrant dashboard, navigate to Collections → n4n_docs → “Points”. You should see 4–6 points, each with a vector (1536 floats) and payload containing page_content and metadata.

Step 5: Build the retrieval chain with citations

A production RAG system needs two things: relevant context retrieval and grounded generation with citations. We’ll use LangChain’s create_retrieval_chain and create_stuff_documents_chain with a prompt that forces the model to cite sources.

# rag_chain.py
import os
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from qdrant_client import QdrantClient

COLLECTION_NAME = "n4n_docs"
QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")

SYSTEM_PROMPT = """You are a technical assistant for n4n.ai. Answer the user's question using ONLY the provided context. 
If the context doesn't contain the answer, say "I don't have enough information to answer that."

Cite your sources by referencing the document metadata. Format citations as [source: filename.md] at the end of each sentence that uses retrieved information.

Context:
{context}"""

def get_vector_store():
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY) if QDRANT_API_KEY else QdrantClient(url=QDRANT_URL)
    return QdrantVectorStore(client=client, collection_name=COLLECTION_NAME, embedding=embeddings)

def build_rag_chain():
    vector_store = get_vector_store()
    retriever = vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 4},
    )

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("human", "{input}"),
    ])

    question_answer_chain = create_stuff_documents_chain(llm, prompt)
    rag_chain = create_retrieval_chain(retriever, question_answer_chain)
    return rag_chain

def format_answer(result: dict) -> str:
    answer = result["answer"]
    sources = result["context"]
    source_files = list({doc.metadata.get("source", "unknown") for doc in sources})
    citation_line = "\n\nSources: " + ", ".join(Path(s).name for s in source_files)
    return answer + citation_line

if __name__ == "__main__":
    chain = build_rag_chain()
    questions = [
        "How does the routing layer handle provider failures?",
        "What token granularity does usage metering operate at?",
        "Does the gateway forward cache-control hints from providers?",
    ]
    for q in questions:
        print(f"\n{'='*60}")
        print(f"Q: {q}")
        result = chain.invoke({"input": q})
        print(format_answer(result))

Verify: Run python rag_chain.py. Each answer should reference the sample document with citations like [source: sample.md]. The routing question should mention automatic fallback on 429/5xx. The metering question should cite token granularity. The cache question should confirm transparent forwarding.

Step 6: Add hybrid search for better recall

Pure dense vector search misses exact keyword matches (product names, error codes, acronyms). Qdrant supports hybrid search — combining dense vectors with sparse BM25 — via the QueryAPI. LangChain’s QdrantVectorStore exposes this through as_retriever(search_type="hybrid") in recent versions, but you can also call the client directly for more control.

# hybrid_retriever.py
from qdrant_client import QdrantClient
from qdrant_client.models import FusionQuery, Fusion
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
import os

COLLECTION_NAME = "n4n_docs"
QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")

def hybrid_search(query: str, k: int = 4):
    client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY) if QDRANT_API_KEY else QdrantClient(url=QDRANT_URL)
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = QdrantVectorStore(client=client, collection_name=COLLECTION_NAME, embedding=embeddings)

    # Dense vector
    dense_vector = embeddings.embed_query(query)

    # Sparse vector via Qdrant's built-in BM25 (requires collection to have a text index)
    # For this tutorial, we'll use the Fusion API with a sparse vector placeholder
    # In production, use `sparse-vector` with a proper BM25 model like SPLADE
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=[
            {"query": dense_vector, "using": "default", "limit": k * 2},
        ],
        query=FusionQuery(fusion=Fusion.RRF),
        limit=k,
        with_payload=True,
    )
    return results.points

if __name__ == "__main__":
    for q in ["429 fallback", "token metering", "cache-control headers"]:
        print(f"\nQuery: {q}")
        points = hybrid_search(q, k=3)
        for i, p in enumerate(points):
            print(f"  {i+1}. score={p.score:.4f} | {p.payload.get('page_content', '')[:120]}...")

Verify: Run python hybrid_retriever.py. You should see results for keyword-heavy queries like “429 fallback” that might rank lower in pure semantic search. Note: full hybrid search with BM25 requires a payload index on the text field — see Qdrant docs for create_payload_index with TextIndexParams.

Step 7: Implement a simple evaluation loop

You can’t ship RAG without measuring quality. A minimal eval: define a small golden set of (question, expected_answer_contains) pairs, run the chain, and check for keyword overlap.

# eval.py
from rag_chain import build_rag_chain, format_answer

GOLDEN_SET = [
    {
        "question": "How does the routing layer handle provider failures?",
        "must_contain": ["429", "5xx", "fallback", "retry"],
    },
    {
        "question": "What token granularity does usage metering operate at?",
        "must_contain": ["token", "granularity", "input", "output"],
    },
    {
        "question": "Does the gateway forward cache-control hints from providers?",
        "must_contain": ["cache-control", "forward", "transparent"],
    },
]

def evaluate():
    chain = build_rag_chain()
    passed = 0
    for item in GOLDEN_SET:
        result = chain.invoke({"input": item["question"]})
        answer = result["answer"].lower()
        missing = [kw for kw in item["must_contain"] if kw.lower() not in answer]
        status = "PASS" if not missing else "FAIL"
        if status == "PASS":
            passed += 1
        print(f"{status}: {item['question']}")
        if missing:
            print(f"  Missing keywords: {missing}")
        print(f"  Answer: {answer[:200]}...")
    print(f"\n{passed}/{len(GOLDEN_SET)} tests passed")

if __name__ == "__main__":
    evaluate()

Verify: Run python eval.py. You should see 3/3 PASS. If any fail, inspect the retrieved context — it usually means k is too low or the chunking split a critical sentence across chunks.

Step 8: Wire it into a FastAPI endpoint (optional but realistic)

Most teams expose RAG as an API. Here’s a minimal FastAPI wrapper with request logging and the same citation format.

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from rag_chain import build_rag_chain, format_answer
import uvicorn

app = FastAPI(title="n4n RAG API")
chain = build_rag_chain()

class QueryRequest(BaseModel):
    question: str
    k: int = 4

class QueryResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    try:
        result = chain.invoke({"input": request.question})
        source_files = list({doc.metadata.get("source", "unknown") for doc in result["context"]})
        return QueryResponse(
            answer=result["answer"],
            sources=[Path(s).name for s in source_files],
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run it: python api.py. Test with:

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What happens when a provider returns 429?"}'

Verify: You get JSON with answer and sources array. The answer should cite sample.md.

Step 9: Production hardening checklist

Before this goes near traffic, address these:

Step 10: Clean up and next steps

Stop the containers when done:

docker stop $(docker ps -q --filter ancestor=qdrant/qdrant)

To extend this tutorial:

• Swap OpenAI for local embeddings (BGE, E5, Nomic) via langchain-huggingface and a local LLM (Ollama, vLLM) via langchain-community
• Add multi-tenancy by namespacing Qdrant collections or using payload filters on a tenant_id field
• Implement incremental indexing: watch the data directory for changes and upsert only modified chunks
• Build a feedback loop: log user thumbs-up/down, mine failures for eval set expansion

The pattern you’ve built — load, chunk, embed, index, retrieve, generate, evaluate — is the backbone of every production RAG system. The difference between a demo and a product is almost entirely in the evaluation loop and the observability you wrap around it.