LangChain RAG with Redis as the vector store

Redis has become a serious contender for vector search workloads. It runs in-memory, supports hybrid queries, and handles the throughput that RAG pipelines demand at scale. This guide walks through building a complete langchain redis vector store rag system from scratch — document ingestion, embedding, indexing, and retrieval — with runnable code at each step.

Step 1: Install dependencies and configure Redis

Start with a clean environment. You need Redis Stack (which includes the RediSearch module) or a managed equivalent like Redis Cloud. The Redis Stack Docker image is the fastest way to get started locally.

docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

Verify the module is loaded:

docker exec -it redis-stack redis-cli MODULE LIST

You should see search in the output. Now install the Python dependencies:

pip install langchain langchain-openai langchain-redis redis pypdf tiktoken python-dotenv

Create a .env file for your OpenAI key:

OPENAI_API_KEY=sk-...

Step 2: Load and chunk source documents

LangChain’s document loaders handle PDFs, markdown, HTML, and more. For this example we’ll use a directory of markdown files, but the pattern is identical for other formats.

# ingest.py
import os
from pathlib import Path
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

DATA_DIR = Path(__file__).parent / "data"
DATA_DIR.mkdir(exist_ok=True)

# Example: drop your .md/.txt files into ./data
loader = DirectoryLoader(str(DATA_DIR), glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
    separators=["\n\n", "\n", " ", ""],
    length_function=len,
)
chunks = splitter.split_documents(documents)
print(f"Loaded {len(documents)} documents, split into {len(chunks)} chunks")

Run it to verify:

python ingest.py
# Loaded 12 documents, split into 147 chunks

Step 3: Initialize embeddings and the Redis vector store

LangChain’s RedisVectorStore class wraps the RediSearch commands. You need an embedding model and a Redis connection URL. The index name becomes the RediSearch index key prefix.

# vectorstore.py
import os
from langchain_openai import OpenAIEmbeddings
from langchain_redis import RedisVectorStore
from langchain_core.documents import Document

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
INDEX_NAME = "langchain_redis_rag_demo"

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

def get_vectorstore() -> RedisVectorStore:
    return RedisVectorStore(
        embedding=embeddings,
        index_name=INDEX_NAME,
        redis_url=REDIS_URL,
    )

The text-embedding-3-small model produces 1536-dimensional vectors. RediSearch will create an HNSW index automatically on first write.

Step 4: Ingest chunks into Redis

Now write the chunks. RedisVectorStore.from_documents handles index creation, vector computation, and batch insertion.

# ingest.py (continued)
from vectorstore import get_vectorstore
from ingest import chunks  # assumes you kept the variable in scope or re-run loader

vectorstore = get_vectorstore()

# Optional: wipe existing index for a clean run
# vectorstore.index.delete(drop=True)

vectorstore.add_documents(chunks)
print(f"Indexed {len(chunks)} chunks into Redis index '{INDEX_NAME}'")

Run the full ingestion:

python ingest.py
# Loaded 12 documents, split into 147 chunks
# Indexed 147 chunks into Redis index 'langchain_redis_rag_demo'

Verify in Redis CLI:

docker exec -it redis-stack redis-cli FT.INFO langchain_redis_rag_demo

Look for num_docs: 147 and indexing: 0 (indexing complete).

Step 5: Build the retrieval chain

LangChain’s create_retrieval_chain and create_stuff_documents_chain compose the RAG pipeline. We’ll use a strict prompt that forces the model to cite sources.

# rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from vectorstore import get_vectorstore

SYSTEM_PROMPT = """You are a precise technical assistant. Answer the question using ONLY the provided context.
If the context does not contain the answer, say "I don't know based on the provided documents."
Cite sources by referencing the chunk metadata (source file and chunk id)."""

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "Context:\n{context}\n\nQuestion: {input}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
vectorstore = get_vectorstore()
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

combine_docs_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)

Step 6: Query the system

A simple CLI loop lets you test interactively.

# query.py
from rag_chain import rag_chain

def main():
    print("LangChain Redis RAG ready. Type 'exit' to quit.")
    while True:
        question = input("\n> ").strip()
        if question.lower() in {"exit", "quit"}:
            break
        result = rag_chain.invoke({"input": question})
        print(f"\nAnswer: {result['answer']}")
        print("\nSources:")
        for i, doc in enumerate(result["context"]):
            meta = doc.metadata
            print(f"  [{i+1}] {meta.get('source', 'unknown')} chunk={meta.get('chunk_id', '?')}")

if __name__ == "__main__":
    main()

Run it:

python query.py
> How do I configure the chunk overlap in the text splitter?
Answer: The RecursiveCharacterTextSplitter is initialized with chunk_overlap=150 in ingest.py.
Sources:
  [1] /path/to/data/ingest.md chunk=3
  [2] /path/to/data/ingest.md chunk=2

Step 7: Add hybrid search for better recall

Pure vector search misses exact keyword matches. RediSearch supports hybrid queries combining vector similarity with BM25 text scoring. LangChain exposes this through search_type="hybrid".

# rag_chain.py (updated retriever)
retriever = vectorstore.as_retriever(
    search_type="hybrid",
    search_kwargs={"k": 5, "alpha": 0.5},  # alpha=0.5 balances vector + keyword
)

The alpha parameter controls the blend: 1.0 is pure vector, 0.0 is pure keyword. Start at 0.5 and tune per dataset.

Step 8: Implement metadata filtering

Real applications need to filter by tenant, document type, or date. RediSearch supports tag and numeric filters. Add metadata during ingestion:

# ingest.py (enhanced)
from datetime import datetime

for i, chunk in enumerate(chunks):
    chunk.metadata.update({
        "chunk_id": i,
        "indexed_at": datetime.utcnow().isoformat(),
        "doc_type": "technical",
    })

Then filter at query time:

# rag_chain.py (filtered retriever)
from langchain_core.runnables import RunnableLambda

def filtered_retriever(query: str, doc_type: str = "technical"):
    return vectorstore.as_retriever(
        search_type="hybrid",
        search_kwargs={
            "k": 5,
            "alpha": 0.5,
            "filter": f"@doc_type:{{{doc_type}}}",
        },
    ).invoke(query)

# Wrap for chain compatibility
retriever = RunnableLambda(lambda x: filtered_retriever(x["input"], x.get("doc_type", "technical")))

The filter syntax follows RediSearch query language: @field:{value} for tags, @field:[min max] for numerics.

Step 9: Persist and reuse the index

Redis persists to disk via AOF/RDB, but the index definition lives in memory. On restart, RediSearch rebuilds from the document hashes. For large datasets, this rebuild takes time. Two strategies:

Option A: Let RediSearch rebuild (simpler, slower restart)

# No code changes needed. On restart:
docker restart redis-stack
# Wait for FT.INFO to show indexing: 0

Option B: Export and import index definition (faster restart)

# export_index.py
import json
import redis

r = redis.from_url("redis://localhost:6379")
info = r.execute_command("FT.INFO", "langchain_redis_rag_demo")
schema = info[info.index("attributes") + 1]
with open("index_schema.json", "w") as f:
    json.dump(schema, f)

On fresh Redis instance:

# import_index.py
import json
from langchain_redis import RedisVectorStore
from langchain_openai import OpenAIEmbeddings

with open("index_schema.json") as f:
    schema = json.load(f)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = RedisVectorStore(
    embedding=embeddings,
    index_name="langchain_redis_rag_demo",
    redis_url="redis://localhost:6379",
    schema=schema,  # pre-create index with known schema
)

Step 10: Monitor and tune in production

Three metrics matter for a langchain redis vector store rag pipeline:

1. Query latency (p50/p99) — Target <100ms p99 for vector search alone. Add MONITOR in Redis CLI to inspect command latency.
2. Index memory footprint — HNSW indexes consume ~1.5x vector size. For 1M chunks at 1536 dimensions (float32): ~9 GB vectors + ~4.5 GB HNSW graph. Plan Redis memory accordingly.
3. Hit rate on hybrid search — Log the retriever’s returned scores. If keyword matches dominate, increase alpha. If semantic matches dominate, decrease it.

Example latency wrapper:

# monitoring.py
import time
from functools import wraps

def timed_retriever(retriever):
    @wraps(retriever.invoke)
    def wrapper(query):
        start = time.perf_counter()
        docs = retriever.invoke(query)
        elapsed = (time.perf_counter() - start) * 1000
        print(f"[retriever] {elapsed:.1f}ms, {len(docs)} docs")
        return docs
    return type(retriever)(invoke=wrapper)

Step 11: Handle multi-tenancy

If you serve multiple customers from one Redis instance, namespace your indexes:

def get_vectorstore(tenant_id: str) -> RedisVectorStore:
    return RedisVectorStore(
        embedding=embeddings,
        index_name=f"rag_{tenant_id}",
        redis_url=REDIS_URL,
    )

Each tenant gets isolated data and independent scaling. Redis Cluster shards indexes across shards automatically when using hash tags: index_name="{tenant_123}_rag".

Step 12: Verify end-to-end correctness

Run this checklist after deployment:

# 1. Index health
docker exec redis-stack redis-cli FT.INFO langchain_redis_rag_demo | grep -E "num_docs|indexing|memory"

# 2. Sample query returns expected sources
python -c "
from rag_chain import rag_chain
r = rag_chain.invoke({'input': 'What is the chunk size?'})
assert '1000' in r['answer']
print('Smoke test passed')
"

# 3. Filter isolation
python -c "
from vectorstore import get_vectorstore
vs = get_vectorstore()
# Insert a doc with doc_type='restricted'
vs.add_documents([Document(page_content='secret', metadata={'doc_type': 'restricted'})])
# Query with default filter (technical) should not see it
from rag_chain import retriever
docs = retriever.invoke('secret')
assert len(docs) == 0, 'Filter leak detected'
print('Filter isolation verified')
"

All three should pass without errors.

Production notes

• Connection pooling: Use redis.ConnectionPool with max_connections=50 for high-throughput services.
• Embedding batch size: OpenAIEmbeddings batches at 100 by default. Increase to 500-1000 for ingestion speed if your rate limits allow.
• Index updates: RediSearch supports incremental adds. Deletes require vectorstore.delete([doc_ids]) followed by FT.ALTER to reclaim memory, or let garbage collection run.
• Fallback strategy: If Redis is unavailable, fail over to a local FAISS index or a secondary provider. This is where a gateway like n4n.ai helps — it can route embedding requests across providers while your vector store stays Redis-native.

You now have a complete, observable, production-grade RAG pipeline on Redis. The same patterns scale from thousands to tens of millions of documents — just add shards.