Build a RAG Pipeline with Redis

Introduction

30 Second Summary

When you ask an AI chatbot a question, it can only answer from what it was trained on. Companies solve this by feeding AI relevant information from their own databases at the moment you ask.

In this project, you will build a RAG pipeline powered by Redis that answers questions from a custom knowledge base. Redis serves as both the vector database for document retrieval and the semantic cache for instant repeat answers.

What You'll Build

A Retrieval-Augmented Generation system that uses Redis as both a vector database and a semantic cache, structured with professional Python patterns for production deployment.

By the end of this project, you'll have:

• A Redis-powered vector search system that stores documents as vector embeddings and retrieves them based on semantic similarity.
• A semantic caching layer that recognizes similar questions and serves cached responses instantly, slashing LLM costs.
• A professional Python project with structured logging, custom exceptions, and a GitHub-ready repository layout.
• Secret Mission: Add reranking to improve retrieval quality before passing context to the LLM.

Are there any prerequisites?

This project assumes you have Python experience and basic Git knowledge (init, add, commit, push). You also need Docker Desktop installed and an OpenAI API key ready.

Set Up the Project and Infrastructure

Every production project starts with a solid foundation. Before writing any pipeline logic, you need a clean repository layout, a running database, and the right packages installed.

In this step, you will create a professional Python project structure, spin up a Redis container with Docker, and wire up structured logging and custom exceptions. These patterns separate your project from throwaway scripts.

In this step, get ready to:

• Create a professional project structure and initialize a Git repository.
• Start Redis via Docker and install Python dependencies.
• Set up configuration, structured logging, and custom exceptions.

Create a professional project structure and initialize git

Production Python projects separate concerns into modules. A flat script works for prototyping, but real deployments need config, exceptions, and services in their own files.

• Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal and pressing Enter.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir redis-rag
cd redis-rag

• Initialize a Python project with uv by running:

uv init

What does uv init do?

This creates a pyproject.toml (your project's metadata and dependency manifest), a .python-version file, and a few starter files. It also initializes a Git repository automatically.

• Create the src/ directory and all your project files by running these commands:

mkdir src
touch src/__init__.py src/config.py src/exceptions.py src/embeddings.py src/indexer.py src/search.py src/cache.py src/rag.py
touch main.py .env.example .gitignore README.md

• Confirm all files exist by running:

ls

You should see main.py, README.md, .gitignore, .env.example, pyproject.toml, and the src/ folder in the output.

Don't see the src folder or files?

Make sure you ran mkdir src before the touch commands. The touch command creates files, not directories.

help me create the project file structure for redis-rag

• Open your project in your code editor by running:

code .

If you use Cursor, replace code with cursor.

• Open .gitignore in your editor and paste this content:

__pycache__/
*.py[cod]
*$py.class
*.so
.env
.venv/
env/
venv/
*.egg-info/
dist/
build/
.pytest_cache/
.mypy_cache/

What does this file do?

A .gitignore tells Git which files to exclude from version control. This keeps compiled Python files (__pycache__/), your secret .env file, and virtual environments out of your repository.

• Save .gitignore.

✔️ Awesome, I've got everything!

Great. Make sure you saved the file before moving on.

ⓧ I'd like to double check the full code

__pycache__/
*.py[cod]
*$py.class
*.so
.env
.venv/
env/
venv/
*.egg-info/
dist/
build/
.pytest_cache/
.mypy_cache/

• Open .env.example in your editor and paste this content:

OPENAI_API_KEY=your-api-key-here
REDIS_HOST=localhost
REDIS_PORT=6379

What is .env.example for?

This file shows other developers which environment variables the project needs, without exposing real secrets. The actual .env file (which holds your real API key) is excluded by .gitignore.

• Save .env.example.

✔️ Awesome, I've got everything!

Great. Make sure you saved the file before moving on.

ⓧ I'd like to double check the full code

OPENAI_API_KEY=your-api-key-here
REDIS_HOST=localhost
REDIS_PORT=6379

• Make your first commit using conventional commit format by running these commands in your terminal:

git add .
git commit -m "feat: initialize project structure"

What are conventional commits?

Conventional commits use prefixes like feat:, fix:, docs:, refactor:, and chore: to communicate intent in commit history. This is how professional teams document what changed and why.

Seeing an author identity error?

If Git asks you to set your name and email, run these commands with your own details:

git config --global user.email "you@example.com"
git config --global user.name "Your Name"

Then retry the commit command.

help me fix my git commit author identity error

Start Redis via Docker and install Python dependencies

Redis is an in-memory data store that will serve as both your vector database and semantic cache. Running it in Docker keeps your setup clean and reproducible.

• Start a Redis container by running this command in your terminal:

docker run -d --name redis-rag -p 6379:6379 redis:latest

What does this command do?

• -d runs the container in the background (detached mode).
• --name redis-rag gives the container a memorable name you can reference later.
• -p 6379:6379 maps the container's Redis port to your machine so your Python code can connect.
• redis:latest pulls the official Redis 8 image, which includes vector search capabilities built-in.

• Verify Redis is running by sending a PING command:

docker exec redis-rag redis-cli PING

You should see PONG in the output. This confirms Redis is alive and accepting connections.

Seeing 'No such container' or 'Cannot connect'?

• Make sure Docker Desktop is running. Check for the Docker icon in your menu bar (macOS) or system tray (Windows).
• If you see a port conflict error, another process is using port 6379. Stop it or use a different port like -p 6380:6379.

help me troubleshoot my Redis Docker container not starting

• Install the Python packages your project needs by running:

uv add redis openai numpy python-dotenv

You should see output confirming each package was resolved and installed.

What are these packages?

• redis is the Python client for connecting to your Redis server.
• openai is the official SDK for calling OpenAI's embedding and chat APIs.
• numpy converts embedding vectors to the byte format Redis requires.
• python-dotenv loads your .env file into environment variables at runtime.

Package install failing?

Make sure you are inside the redis-rag folder where pyproject.toml lives. Run ls to confirm you can see it.

help me fix uv add failing to install packages

• Commit the dependency changes by running:

git add .
git commit -m "feat: add redis docker setup and python dependencies"

Set up configuration, logging, and custom exceptions

Now you will write the foundational modules that every other file in the project depends on. These set up how your app loads secrets, reports what it is doing, and communicates what went wrong.

• Open src/__init__.py in your editor and paste the following code:

"""Redis RAG Pipeline - Source package."""

import logging
import sys


def setup_logging(level: str = "INFO") -> logging.Logger:
    """Configure structured logging for the application."""
    logger = logging.getLogger("redis_rag")
    logger.setLevel(getattr(logging, level.upper(), logging.INFO))

    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setLevel(logging.DEBUG)
        formatter = logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s.%(module)s | %(message)s",
            datefmt="%H:%M:%S",
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)

    return logger


logger = setup_logging()

Why logging instead of print?

In production, print() goes to stdout and is lost. Logging has levels (DEBUG, INFO, WARNING, ERROR) that can be filtered without changing code.

The structured format (timestamp | level | module | message) makes it easy to search logs when debugging. You can also route logs to files or monitoring services like CloudWatch or Datadog.

• Save src/__init__.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file.

ⓧ I'd like to double check the full code

"""Redis RAG Pipeline - Source package."""

import logging
import sys


def setup_logging(level: str = "INFO") -> logging.Logger:
    """Configure structured logging for the application."""
    logger = logging.getLogger("redis_rag")
    logger.setLevel(getattr(logging, level.upper(), logging.INFO))

    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setLevel(logging.DEBUG)
        formatter = logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s.%(module)s | %(message)s",
            datefmt="%H:%M:%S",
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)

    return logger


logger = setup_logging()

• Open src/config.py in your editor and paste the following code:

"""Application configuration loaded from environment variables."""

import os

from dotenv import load_dotenv

load_dotenv()

# Redis
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))

# OpenAI
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
VECTOR_DIM = 1536

# Search
INDEX_NAME = "idx:docs"
DOC_PREFIX = "doc:"
CACHE_PREFIX = "cache:"
CACHE_INDEX_NAME = "idx:cache"
CACHE_SIMILARITY_THRESHOLD = 0.95

What does this file do?

• load_dotenv() reads your .env file and injects its values into the environment so os.getenv() can find them.
• EMBEDDING_MODEL and CHAT_MODEL are constants that pin your pipeline to specific OpenAI models.
• VECTOR_DIM = 1536 matches the default output dimension of text-embedding-3-small.
• CACHE_SIMILARITY_THRESHOLD = 0.95 means a cached response is returned only when the new query is at least 95% similar to a previously cached one.

• Save src/config.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file.

ⓧ I'd like to double check the full code

"""Application configuration loaded from environment variables."""

import os

from dotenv import load_dotenv

load_dotenv()

# Redis
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))

# OpenAI
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
VECTOR_DIM = 1536

# Search
INDEX_NAME = "idx:docs"
DOC_PREFIX = "doc:"
CACHE_PREFIX = "cache:"
CACHE_INDEX_NAME = "idx:cache"
CACHE_SIMILARITY_THRESHOLD = 0.95

• Open src/exceptions.py in your editor and paste the following code:

"""Custom exceptions for the Redis RAG pipeline."""


class RedisRAGError(Exception):
    """Base exception for all Redis RAG errors."""

    pass


class RedisConnectionError(RedisRAGError):
    """Raised when Redis connection fails."""

    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        super().__init__(
            f"Failed to connect to Redis at {host}:{port}. "
            f"Is the container running? Try: docker start redis-rag"
        )


class EmbeddingError(RedisRAGError):
    """Raised when embedding generation fails."""

    def __init__(self, model: str, detail: str = ""):
        self.model = model
        super().__init__(
            f"Failed to generate embeddings with model '{model}'. {detail}"
        )


class SearchError(RedisRAGError):
    """Raised when vector search or index operations fail."""

    def __init__(self, index_name: str, detail: str = ""):
        self.index_name = index_name
        super().__init__(
            f"Search error on index '{index_name}'. {detail}"
        )


class CacheError(RedisRAGError):
    """Raised when semantic cache operations fail."""

    def __init__(self, detail: str = ""):
        super().__init__(f"Semantic cache error. {detail}")

Why custom exceptions?

Custom exceptions let callers handle specific failure modes. If Redis is down, your code can retry the connection. If the embedding API rate-limits you, your code can wait and retry. Generic Exception catches give you no way to respond differently to different problems.

Notice how each exception class includes a helpful error message that tells the developer exactly what went wrong and what to try next.

• Save src/exceptions.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file.

ⓧ I'd like to double check the full code

"""Custom exceptions for the Redis RAG pipeline."""


class RedisRAGError(Exception):
    """Base exception for all Redis RAG errors."""

    pass


class RedisConnectionError(RedisRAGError):
    """Raised when Redis connection fails."""

    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        super().__init__(
            f"Failed to connect to Redis at {host}:{port}. "
            f"Is the container running? Try: docker start redis-rag"
        )


class EmbeddingError(RedisRAGError):
    """Raised when embedding generation fails."""

    def __init__(self, model: str, detail: str = ""):
        self.model = model
        super().__init__(
            f"Failed to generate embeddings with model '{model}'. {detail}"
        )


class SearchError(RedisRAGError):
    """Raised when vector search or index operations fail."""

    def __init__(self, index_name: str, detail: str = ""):
        self.index_name = index_name
        super().__init__(
            f"Search error on index '{index_name}'. {detail}"
        )


class CacheError(RedisRAGError):
    """Raised when semantic cache operations fail."""

    def __init__(self, detail: str = ""):
        super().__init__(f"Semantic cache error. {detail}")

Now you need to create the actual .env file with your real OpenAI API key so the config module can load it.

• Create a new file called .env in the root of your redis-rag folder (next to .env.example).
• Paste the following content, replacing the placeholder with your actual OpenAI API key:

OPENAI_API_KEY=your-api-key-here
REDIS_HOST=localhost
REDIS_PORT=6379

• Replace your-api-key-here with your real OpenAI API key (starts with sk-).
• Save .env.

Why is .env not committed to Git?

Your .gitignore already excludes .env. This keeps your secret API key out of version control. The .env.example file shows collaborators what variables they need without exposing real values.

Finally, verify that your OpenAI key works by running a quick test embedding call.

• Run this command to test your API key:

uv run python -c "from openai import OpenAI; from dotenv import load_dotenv; import os; load_dotenv(); c = OpenAI(api_key=os.getenv('OPENAI_API_KEY')); r = c.embeddings.create(input='hello', model='text-embedding-3-small'); print(f'Success! Got {len(r.data[0].embedding)} dimensions')"

You should see Success! Got 1536 dimensions in the output. This confirms your API key is valid and the embedding model is accessible.

Seeing an authentication error?

• Open your .env file and make sure there are no extra spaces around the = sign or trailing whitespace after the key.
• Confirm you copied the full key from OpenAI (it should start with sk-).
• Check that your key has available credits on the OpenAI dashboard.

help me fix my OpenAI API key authentication error

• Commit the configuration, logging, and exceptions by running:

git add .
git commit -m "feat: add configuration, logging, and custom exceptions"

Your foundation is solid. You have a running Redis instance, a clean project layout, and production-grade logging and error handling. Next up, you will load documents into Redis and generate vector embeddings.

Load Documents and Generate Embeddings

Your infrastructure is running and your project skeleton is in place. Now it is time to put data into Redis and turn plain text into something a machine can search.

In this step you will build the knowledge base your RAG pipeline searches against. You will explore how Redis stores structured data, generate vector embeddings with OpenAI, and load everything into Redis using a high-performance pipeline.

In this step, get ready to:

• Explore Redis data commands interactively via the CLI.
• Build an embedding module that converts text to vectors.
• Load 10 knowledge-base documents into Redis with their embeddings.

Explore Redis basics and store documents as JSON

Before writing Python code, it helps to see how Redis stores data by interacting with it directly. Redis is not a traditional disk-based database. It keeps everything in memory, which means reads and writes happen in sub-millisecond time.

• Open the Redis CLI inside your running container by running this command:

docker exec -it redis-rag redis-cli

You should see a 127.0.0.1:6379> prompt. This means you are connected to Redis.

• Try the basic string commands to see how Redis works:

SET greeting "hello world"
GET greeting

You should see OK after SET and "hello world" after GET.

• Now try storing a JSON document, which is how your knowledge base will be stored:

JSON.SET testdoc $ '{"title": "Hello", "category": "demo"}'
JSON.GET testdoc $

You should see OK followed by the JSON object you stored.

• Exit the Redis CLI by typing:

exit

Why is Redis different from a traditional database?

Redis keeps all data in RAM. This is what delivers sub-millisecond latency for reads and writes. A traditional database like PostgreSQL stores data on disk and loads it into memory on demand.

Redis also uses a single-threaded event loop for command processing. This eliminates race conditions and lock contention. Despite being single-threaded, it handles tens of thousands of concurrent connections efficiently using I/O multiplexing.

Now that you understand the data model, you will define the knowledge base and Redis connection in Python.

• Open src/indexer.py in your editor and replace the placeholder content with the imports and the DOCUMENTS list by pasting the following code:

"""Document loading and index creation for Redis."""

import time

import redis
from redis.commands.search.field import TagField, TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType

from src import logger
from src.config import (
    CACHE_INDEX_NAME,
    CACHE_PREFIX,
    DOC_PREFIX,
    INDEX_NAME,
    REDIS_HOST,
    REDIS_PORT,
    VECTOR_DIM,
)
from src.embeddings import get_embeddings_batch
from src.exceptions import RedisConnectionError, SearchError

# Knowledge base
DOCUMENTS = [
    {
        "id": "doc:001",
        "title": "Redis Overview",
        "category": "database",
        "content": "Redis is an open-source, in-memory data structure store used as a database, cache, message broker, and streaming engine. It supports data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes, and streams. Redis has built-in replication, Lua scripting, LRU eviction, transactions, and different levels of on-disk persistence.",
    },
    {
        "id": "doc:002",
        "title": "Redis Vector Search",
        "category": "ai",
        "content": "Redis supports vector similarity search through its integrated search engine. You can store vector embeddings alongside your data in Redis hashes or JSON documents, create secondary indexes with vector fields, and perform KNN or range-based vector queries. Redis supports FLAT, HNSW, and SVS-VAMANA indexing algorithms with cosine, L2, and inner product distance metrics.",
    },
    {
        "id": "doc:003",
        "title": "RAG with Redis",
        "category": "ai",
        "content": "Retrieval Augmented Generation (RAG) enhances LLMs by integrating external data stored in a vector database. Redis supports RAG workflows by providing vector search for semantic retrieval, semantic caching to reduce LLM inference costs, and session management for conversation history. The in-memory architecture ensures sub-millisecond latency for retrieval operations.",
    },
    {
        "id": "doc:004",
        "title": "Redis Caching Patterns",
        "category": "performance",
        "content": "Redis excels as a cache due to its in-memory architecture delivering sub-millisecond response times. Common patterns include cache-aside (lazy loading), write-through, and write-behind caching. Redis supports TTL-based expiration, LRU eviction policies, and can handle millions of operations per second. For AI workloads, semantic caching uses vector similarity to match queries that are semantically equivalent but worded differently.",
    },
    {
        "id": "doc:005",
        "title": "Redis Concurrency",
        "category": "performance",
        "content": "Redis uses a single-threaded event loop for command processing, which eliminates race conditions and lock contention. Despite being single-threaded, Redis handles tens of thousands of concurrent client connections efficiently using I/O multiplexing. This makes it ideal for high-traffic AI applications where many users query the same RAG pipeline simultaneously without performance degradation.",
    },
    {
        "id": "doc:006",
        "title": "Vector Database Comparison",
        "category": "ai",
        "content": "Dedicated vector databases like Pinecone, Weaviate, and Qdrant are purpose-built for vector search. Pinecone is fully managed with a generous free tier. Weaviate offers open-source with built-in hybrid search. Qdrant is written in Rust for high performance. Redis differs by being a general-purpose data platform that adds vector search to its existing capabilities, allowing teams to consolidate their infrastructure instead of adding another service.",
    },
    {
        "id": "doc:007",
        "title": "HNSW Index Algorithm",
        "category": "ai",
        "content": "Hierarchical Navigable Small World (HNSW) is an approximate nearest neighbor algorithm that builds a multi-layered graph structure for fast vector search. It achieves 95-99% recall while being orders of magnitude faster than brute-force flat search on large datasets. HNSW is the default choice for most production vector search applications due to its balance of speed and accuracy.",
    },
    {
        "id": "doc:008",
        "title": "Embeddings Explained",
        "category": "ai",
        "content": "Vector embeddings are numerical representations of data that capture semantic meaning. Text embedding models like OpenAI's text-embedding-3-small convert text into 1536-dimensional vectors where similar texts are positioned close together in vector space. The cosine similarity between two embedding vectors measures how semantically related the original texts are, with values ranging from -1 to 1.",
    },
    {
        "id": "doc:009",
        "title": "Redis JSON",
        "category": "database",
        "content": "Redis JSON allows storing, updating, and retrieving JSON documents in Redis. Documents are stored in a binary tree structure for fast access to nested elements via JSONPath expressions. You can store vector embeddings directly within JSON documents alongside metadata, enabling co-located storage where your vectors and business data live in the same place.",
    },
    {
        "id": "doc:010",
        "title": "Redis Pipelining",
        "category": "performance",
        "content": "Redis pipelining sends multiple commands to the server in a single network round-trip, dramatically reducing latency for batch operations. Instead of waiting for each command's response before sending the next, pipelining buffers commands and sends them together. This is especially useful when ingesting large numbers of documents or embeddings into Redis.",
    },
]

What does this code do?

The imports at the top pull in Redis search field types (TagField, TextField, VectorField) that you will use to define search indexes in the next step.

The DOCUMENTS list is your knowledge base. Each document has an id (the Redis key), a title, a category tag for filtering, and the content that will be embedded as a vector.

Next, add the get_redis_client() function below the DOCUMENTS list. This function creates and verifies a connection to your running Redis container.

• Paste the following code at the bottom of src/indexer.py, below the closing ] of the DOCUMENTS list:

def get_redis_client() -> redis.Redis:
    """Create and verify a Redis connection."""
    try:
        client = redis.Redis(
            host=REDIS_HOST, port=REDIS_PORT, decode_responses=True
        )
        client.ping()
        logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
        return client
    except redis.ConnectionError:
        raise RedisConnectionError(REDIS_HOST, REDIS_PORT)

What does this code do?

The get_redis_client() function connects to Redis and immediately calls ping() to verify the connection is alive. If the container is not running, it raises your custom RedisConnectionError with a helpful message.

The decode_responses=True parameter tells redis-py to return Python strings instead of raw bytes. This makes working with the responses cleaner.

• Save src/indexer.py.

Seeing an import error for src.embeddings?

The import from src.embeddings import get_embeddings_batch will fail until you create src/embeddings.py in the next substep. This is expected. Do not run the file yet.

Help me fix this import error

Generate vector embeddings with OpenAI and store them

Vector embeddings are how your pipeline understands meaning. The text-embedding-3-small model converts each chunk of text into a 1536-dimensional vector. Documents with similar meaning end up close together in vector space, which is what makes semantic search possible.

• Open src/embeddings.py and replace the placeholder content with the imports, client setup, and the get_embedding() function:

"""Embedding generation using OpenAI API."""

from openai import OpenAI

from src import logger
from src.config import EMBEDDING_MODEL, OPENAI_API_KEY
from src.exceptions import EmbeddingError

client = OpenAI(api_key=OPENAI_API_KEY)


def get_embedding(text: str) -> list[float]:
    """Generate an embedding vector for a single text string."""
    try:
        response = client.embeddings.create(
            input=text,
            model=EMBEDDING_MODEL,
        )
        return response.data[0].embedding
    except Exception as e:
        raise EmbeddingError(EMBEDDING_MODEL, str(e))

What does this code do?

The module creates a single OpenAI client at module level so it can be reused across calls without re-authenticating.

The get_embedding() function sends one text string to OpenAI and returns the 1536-float vector. Any failure (rate limit, invalid key, network timeout) is caught and re-raised as your custom EmbeddingError so callers get a clear, debuggable message.

• Now add the batch embedding function below get_embedding() in the same file:

def get_embeddings_batch(texts: list[str]) -> list[list[float]]:
    """Generate embeddings for multiple texts in a single API call."""
    if not texts:
        return []

    try:
        logger.info(f"Generating embeddings for {len(texts)} texts")
        response = client.embeddings.create(
            input=texts,
            model=EMBEDDING_MODEL,
        )
        embeddings = [item.embedding for item in response.data]
        logger.info(f"Generated {len(embeddings)} embeddings ({EMBEDDING_MODEL})")
        return embeddings
    except Exception as e:
        raise EmbeddingError(EMBEDDING_MODEL, str(e))

Why batch instead of one at a time?

OpenAI's embeddings endpoint accepts a list of texts and returns all their vectors in one response. Sending 10 texts in one call is faster and cheaper than making 10 separate requests. The text-embedding-3-small model costs $0.02 per 1M tokens, so your 10 documents cost a fraction of a cent.

• Save src/embeddings.py.

Seeing an authentication error?

Make sure your .env file contains a valid OPENAI_API_KEY value. If you recently created it, wait a few seconds for it to activate.

Help me debug my OpenAI API key issue

With the embeddings module ready, you can now write the function that ties everything together: generating embeddings for all 10 documents and storing them in Redis.

• Switch back to src/indexer.py and add the load_documents() function and the two stub functions below get_redis_client():

def load_documents(client: redis.Redis) -> None:
    """Load all documents into Redis with their embeddings."""
    start = time.time()
    logger.info(f"Loading {len(DOCUMENTS)} documents into Redis...")

    texts = [doc["content"] for doc in DOCUMENTS]
    embeddings = get_embeddings_batch(texts)

    pipeline = client.pipeline()
    for doc, embedding in zip(DOCUMENTS, embeddings):
        doc_data = {
            "title": doc["title"],
            "category": doc["category"],
            "content": doc["content"],
            "embedding": embedding,
        }
        pipeline.json().set(doc["id"], "$", doc_data)

    pipeline.execute()
    elapsed = time.time() - start
    logger.info(f"Loaded {len(DOCUMENTS)} documents in {elapsed:.2f}s")


def create_document_index(client: redis.Redis) -> None:
    pass


def create_cache_index(client: redis.Redis) -> None:
    pass

What does this code do?

The load_documents() function does three things: it extracts all content strings, generates embeddings for all of them in a single batch call, then stores each document in Redis using JSON.SET.

The pipeline object buffers all 10 JSON.SET commands and sends them in a single network round-trip. This is Redis pipelining in action. Without it, each command would require its own round-trip, adding latency.

The two pass stubs (create_document_index and create_cache_index) are placeholders you will fill in during the next step.

• Save src/indexer.py.
• Verify your documents load correctly by running this command from your redis-rag folder:

uv run python -c "from src.indexer import get_redis_client, load_documents; c = get_redis_client(); load_documents(c); print('Done!')"

You should see log output showing the connection, embedding generation for 10 texts, and a confirmation that documents were loaded. The final line prints Done!.

• Confirm the documents are stored in Redis by running:

docker exec redis-rag redis-cli JSON.GET doc:001 $.title

You should see ["Redis Overview"] in the output. This confirms your document was stored with its metadata.

Seeing a connection refused error?

Make sure your Redis container is running. Check with docker ps and look for redis-rag in the output. If it is not listed, restart it with docker start redis-rag.

Help me fix my Redis connection error

✔️ Awesome, I've got everything!

Great. Double check you have saved both src/embeddings.py and src/indexer.py.

ⓧ I'd like to double check the full code

Here is the full src/embeddings.py:

"""Embedding generation using OpenAI API."""

from openai import OpenAI

from src import logger
from src.config import EMBEDDING_MODEL, OPENAI_API_KEY
from src.exceptions import EmbeddingError

client = OpenAI(api_key=OPENAI_API_KEY)


def get_embedding(text: str) -> list[float]:
    """Generate an embedding vector for a single text string."""
    try:
        response = client.embeddings.create(
            input=text,
            model=EMBEDDING_MODEL,
        )
        return response.data[0].embedding
    except Exception as e:
        raise EmbeddingError(EMBEDDING_MODEL, str(e))


def get_embeddings_batch(texts: list[str]) -> list[list[float]]:
    """Generate embeddings for multiple texts in a single API call."""
    if not texts:
        return []

    try:
        logger.info(f"Generating embeddings for {len(texts)} texts")
        response = client.embeddings.create(
            input=texts,
            model=EMBEDDING_MODEL,
        )
        embeddings = [item.embedding for item in response.data]
        logger.info(f"Generated {len(embeddings)} embeddings ({EMBEDDING_MODEL})")
        return embeddings
    except Exception as e:
        raise EmbeddingError(EMBEDDING_MODEL, str(e))

Here is the full src/indexer.py:

"""Document loading and index creation for Redis."""

import time

import redis
from redis.commands.search.field import TagField, TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType

from src import logger
from src.config import (
    CACHE_INDEX_NAME,
    CACHE_PREFIX,
    DOC_PREFIX,
    INDEX_NAME,
    REDIS_HOST,
    REDIS_PORT,
    VECTOR_DIM,
)
from src.embeddings import get_embeddings_batch
from src.exceptions import RedisConnectionError, SearchError

# Knowledge base
DOCUMENTS = [
    {
        "id": "doc:001",
        "title": "Redis Overview",
        "category": "database",
        "content": "Redis is an open-source, in-memory data structure store used as a database, cache, message broker, and streaming engine. It supports data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes, and streams. Redis has built-in replication, Lua scripting, LRU eviction, transactions, and different levels of on-disk persistence.",
    },
    {
        "id": "doc:002",
        "title": "Redis Vector Search",
        "category": "ai",
        "content": "Redis supports vector similarity search through its integrated search engine. You can store vector embeddings alongside your data in Redis hashes or JSON documents, create secondary indexes with vector fields, and perform KNN or range-based vector queries. Redis supports FLAT, HNSW, and SVS-VAMANA indexing algorithms with cosine, L2, and inner product distance metrics.",
    },
    {
        "id": "doc:003",
        "title": "RAG with Redis",
        "category": "ai",
        "content": "Retrieval Augmented Generation (RAG) enhances LLMs by integrating external data stored in a vector database. Redis supports RAG workflows by providing vector search for semantic retrieval, semantic caching to reduce LLM inference costs, and session management for conversation history. The in-memory architecture ensures sub-millisecond latency for retrieval operations.",
    },
    {
        "id": "doc:004",
        "title": "Redis Caching Patterns",
        "category": "performance",
        "content": "Redis excels as a cache due to its in-memory architecture delivering sub-millisecond response times. Common patterns include cache-aside (lazy loading), write-through, and write-behind caching. Redis supports TTL-based expiration, LRU eviction policies, and can handle millions of operations per second. For AI workloads, semantic caching uses vector similarity to match queries that are semantically equivalent but worded differently.",
    },
    {
        "id": "doc:005",
        "title": "Redis Concurrency",
        "category": "performance",
        "content": "Redis uses a single-threaded event loop for command processing, which eliminates race conditions and lock contention. Despite being single-threaded, Redis handles tens of thousands of concurrent client connections efficiently using I/O multiplexing. This makes it ideal for high-traffic AI applications where many users query the same RAG pipeline simultaneously without performance degradation.",
    },
    {
        "id": "doc:006",
        "title": "Vector Database Comparison",
        "category": "ai",
        "content": "Dedicated vector databases like Pinecone, Weaviate, and Qdrant are purpose-built for vector search. Pinecone is fully managed with a generous free tier. Weaviate offers open-source with built-in hybrid search. Qdrant is written in Rust for high performance. Redis differs by being a general-purpose data platform that adds vector search to its existing capabilities, allowing teams to consolidate their infrastructure instead of adding another service.",
    },
    {
        "id": "doc:007",
        "title": "HNSW Index Algorithm",
        "category": "ai",
        "content": "Hierarchical Navigable Small World (HNSW) is an approximate nearest neighbor algorithm that builds a multi-layered graph structure for fast vector search. It achieves 95-99% recall while being orders of magnitude faster than brute-force flat search on large datasets. HNSW is the default choice for most production vector search applications due to its balance of speed and accuracy.",
    },
    {
        "id": "doc:008",
        "title": "Embeddings Explained",
        "category": "ai",
        "content": "Vector embeddings are numerical representations of data that capture semantic meaning. Text embedding models like OpenAI's text-embedding-3-small convert text into 1536-dimensional vectors where similar texts are positioned close together in vector space. The cosine similarity between two embedding vectors measures how semantically related the original texts are, with values ranging from -1 to 1.",
    },
    {
        "id": "doc:009",
        "title": "Redis JSON",
        "category": "database",
        "content": "Redis JSON allows storing, updating, and retrieving JSON documents in Redis. Documents are stored in a binary tree structure for fast access to nested elements via JSONPath expressions. You can store vector embeddings directly within JSON documents alongside metadata, enabling co-located storage where your vectors and business data live in the same place.",
    },
    {
        "id": "doc:010",
        "title": "Redis Pipelining",
        "category": "performance",
        "content": "Redis pipelining sends multiple commands to the server in a single network round-trip, dramatically reducing latency for batch operations. Instead of waiting for each command's response before sending the next, pipelining buffers commands and sends them together. This is especially useful when ingesting large numbers of documents or embeddings into Redis.",
    },
]


def get_redis_client() -> redis.Redis:
    """Create and verify a Redis connection."""
    try:
        client = redis.Redis(
            host=REDIS_HOST, port=REDIS_PORT, decode_responses=True
        )
        client.ping()
        logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
        return client
    except redis.ConnectionError:
        raise RedisConnectionError(REDIS_HOST, REDIS_PORT)


def load_documents(client: redis.Redis) -> None:
    """Load all documents into Redis with their embeddings."""
    start = time.time()
    logger.info(f"Loading {len(DOCUMENTS)} documents into Redis...")

    texts = [doc["content"] for doc in DOCUMENTS]
    embeddings = get_embeddings_batch(texts)

    pipeline = client.pipeline()
    for doc, embedding in zip(DOCUMENTS, embeddings):
        doc_data = {
            "title": doc["title"],
            "category": doc["category"],
            "content": doc["content"],
            "embedding": embedding,
        }
        pipeline.json().set(doc["id"], "$", doc_data)

    pipeline.execute()
    elapsed = time.time() - start
    logger.info(f"Loaded {len(DOCUMENTS)} documents in {elapsed:.2f}s")


def create_document_index(client: redis.Redis) -> None:
    pass


def create_cache_index(client: redis.Redis) -> None:
    pass

Understand embedding models and the vector database landscape

Why text-embedding-3-small?

OpenAI offers several embedding models. The text-embedding-3-small model produces 1536-dimensional vectors by default. It is fast, cheap ($0.02 per 1M tokens), and its output dimensions are the standard accepted by most vector databases.

Cosine similarity is the distance metric you will use for search. It measures how similar two vectors are based on the angle between them, regardless of magnitude. A score of 1.0 means identical direction (same meaning), while 0 means unrelated.

How does Redis compare to dedicated vector databases?

Dedicated vector databases like Pinecone, Weaviate, and Qdrant store vectors as a separate service. Redis keeps vectors co-located with your metadata in the same instance that handles caching and session management.

The trade-off is that Redis stores everything in RAM, which gets expensive for very large corpora (millions of vectors). Dedicated vector databases use disk-based approaches that scale better at that size. For most production RAG systems with thousands to hundreds of thousands of documents, Redis offers the simplest architecture with the fewest moving parts.

• Commit your progress by running:

git add -A && git commit -m "feat: add document ingestion with embeddings"

Your knowledge base is loaded and each document has a 1536-dimensional vector stored alongside its text. In the next step, you will create a vector index so Redis can search these embeddings at scale.

Create a Vector Index and Search

Your documents and embeddings are sitting in Redis, but Redis doesn't yet know how to search them efficiently. Without an index, Redis would need to compare every query vector against all 10 document vectors one by one.

In this step, you'll create an HNSW vector index that organizes your embeddings into a graph structure for fast approximate nearest-neighbor lookups. Then you'll write a search function that embeds a query, runs a KNN search against the index, and returns the most relevant documents.

In this step, get ready to:

• Create an HNSW vector index on your documents.
• Write a KNN vector search function.
• Add category-filtered hybrid search.

Create a vector index with FT.CREATE

A vector index tells Redis how to organize and search your embedding vectors. The HNSW (Hierarchical Navigable Small World) algorithm builds a multi-layered graph that achieves 95-99% recall while being orders of magnitude faster than brute-force comparison.

• In src/indexer.py, find the stub create_document_index() function (the one that contains only pass).
• Replace the entire function with the following code:

def create_document_index(client: redis.Redis) -> None:
    """Create the vector search index for documents."""
    try:
        client.ft(INDEX_NAME).dropindex(delete_documents=False)
        logger.debug(f"Dropped existing index '{INDEX_NAME}'")
    except Exception:
        pass

    schema = (
        TextField("$.title", as_name="title"),
        TagField("$.category", as_name="category"),
        TextField("$.content", as_name="content"),
        VectorField(
            "$.embedding",
            "HNSW",
            {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"},
            as_name="vector",
        ),
    )
    definition = IndexDefinition(prefix=[DOC_PREFIX], index_type=IndexType.JSON)
    try:
        client.ft(INDEX_NAME).create_index(fields=schema, definition=definition)
        info = client.ft(INDEX_NAME).info()
        logger.info(f"Index '{INDEX_NAME}' created: {info['num_docs']} docs indexed")
    except Exception as e:
        raise SearchError(INDEX_NAME, str(e))

What does this code do?

This function defines a search schema that tells Redis what fields to index and how to treat each one:

• TextField indexes title and content for full-text search capabilities.
• TagField indexes category as an exact-match filter (used for hybrid search later).
• VectorField indexes the embedding array using HNSW with 1536 dimensions and COSINE distance.
• The dropindex() call at the top is idempotent cleanup. It lets you re-run the function without errors if the index already exists.
• FT.INFO verifies the index was created successfully and logs how many documents were indexed.

Now you'll also create a cache index for the semantic caching layer you'll build in the next step.

• Still in src/indexer.py, find the stub create_cache_index() function.
• Replace the entire function with the following code:

def create_cache_index(client: redis.Redis) -> None:
    """Create the vector search index for semantic cache."""
    try:
        client.ft(CACHE_INDEX_NAME).dropindex(delete_documents=False)
    except Exception:
        pass

    schema = (
        TextField("$.original_query", as_name="original_query"),
        TextField("$.response", as_name="response"),
        VectorField(
            "$.embedding",
            "FLAT",
            {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"},
            as_name="vector",
        ),
    )
    definition = IndexDefinition(prefix=[CACHE_PREFIX], index_type=IndexType.JSON)
    try:
        client.ft(CACHE_INDEX_NAME).create_index(
            fields=schema, definition=definition
        )
        logger.info(f"Cache index '{CACHE_INDEX_NAME}' created")
    except Exception as e:
        raise SearchError(CACHE_INDEX_NAME, str(e))

Why FLAT instead of HNSW for the cache?

The cache will only hold a small number of entries (dozens to hundreds of cached responses). For small datasets, FLAT (brute-force) search is actually faster than HNSW because there's no graph traversal overhead.

HNSW shines on large datasets (thousands to millions of vectors) where approximate search saves significant time. Choose the algorithm based on your expected dataset size.

• Save src/indexer.py.
• Verify the indexes work by running this command:

uv run python -c "
from src.indexer import get_redis_client, create_document_index, create_cache_index
client = get_redis_client()
create_document_index(client)
create_cache_index(client)
print('Both indexes created successfully.')
"

• You should see log output showing a successful Redis connection, the document index created with 10 docs indexed, and the cache index created.

Seeing a SearchError or connection error?

• Make sure your Redis container is still running. Run docker ps to check. If it's stopped, restart it with docker start redis-rag.
• Confirm that documents were loaded in the previous step. If you restarted the container, you may need to re-run the document loading step first.
• Check that the imports at the top of src/indexer.py include SearchError from src.exceptions.

help me debug my Redis index creation error

✔️ Awesome, I've got everything!

Great. Double check you've saved src/indexer.py before moving on.

ⓧ I'd like to double check the full code

"""Document loading and index creation for Redis."""

import time

import redis
from redis.commands.search.field import TagField, TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType

from src import logger
from src.config import (
    CACHE_INDEX_NAME,
    CACHE_PREFIX,
    DOC_PREFIX,
    INDEX_NAME,
    REDIS_HOST,
    REDIS_PORT,
    VECTOR_DIM,
)
from src.embeddings import get_embeddings_batch
from src.exceptions import RedisConnectionError, SearchError

DOCUMENTS = [
    {"id": "doc:001", "title": "Redis Overview", "category": "database", "content": "Redis is an open-source, in-memory data structure store used as a database, cache, message broker, and streaming engine. It supports data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes, and streams. Redis has built-in replication, Lua scripting, LRU eviction, transactions, and different levels of on-disk persistence."},
    {"id": "doc:002", "title": "Redis Vector Search", "category": "ai", "content": "Redis supports vector similarity search through its integrated search engine. You can store vector embeddings alongside your data in Redis hashes or JSON documents, create secondary indexes with vector fields, and perform KNN or range-based vector queries. Redis supports FLAT, HNSW, and SVS-VAMANA indexing algorithms with cosine, L2, and inner product distance metrics."},
    {"id": "doc:003", "title": "RAG with Redis", "category": "ai", "content": "Retrieval Augmented Generation (RAG) enhances LLMs by integrating external data stored in a vector database. Redis supports RAG workflows by providing vector search for semantic retrieval, semantic caching to reduce LLM inference costs, and session management for conversation history. The in-memory architecture ensures sub-millisecond latency for retrieval operations."},
    {"id": "doc:004", "title": "Redis Caching Patterns", "category": "performance", "content": "Redis excels as a cache due to its in-memory architecture delivering sub-millisecond response times. Common patterns include cache-aside (lazy loading), write-through, and write-behind caching. Redis supports TTL-based expiration, LRU eviction policies, and can handle millions of operations per second. For AI workloads, semantic caching uses vector similarity to match queries that are semantically equivalent but worded differently."},
    {"id": "doc:005", "title": "Redis Concurrency", "category": "performance", "content": "Redis uses a single-threaded event loop for command processing, which eliminates race conditions and lock contention. Despite being single-threaded, Redis handles tens of thousands of concurrent client connections efficiently using I/O multiplexing. This makes it ideal for high-traffic AI applications where many users query the same RAG pipeline simultaneously without performance degradation."},
    {"id": "doc:006", "title": "Vector Database Comparison", "category": "ai", "content": "Dedicated vector databases like Pinecone, Weaviate, and Qdrant are purpose-built for vector search. Pinecone is fully managed with a generous free tier. Weaviate offers open-source with built-in hybrid search. Qdrant is written in Rust for high performance. Redis differs by being a general-purpose data platform that adds vector search to its existing capabilities, allowing teams to consolidate their infrastructure instead of adding another service."},
    {"id": "doc:007", "title": "HNSW Index Algorithm", "category": "ai", "content": "Hierarchical Navigable Small World (HNSW) is an approximate nearest neighbor algorithm that builds a multi-layered graph structure for fast vector search. It achieves 95-99% recall while being orders of magnitude faster than brute-force flat search on large datasets. HNSW is the default choice for most production vector search applications due to its balance of speed and accuracy."},
    {"id": "doc:008", "title": "Embeddings Explained", "category": "ai", "content": "Vector embeddings are numerical representations of data that capture semantic meaning. Text embedding models like OpenAI's text-embedding-3-small convert text into 1536-dimensional vectors where similar texts are positioned close together in vector space. The cosine similarity between two embedding vectors measures how semantically related the original texts are, with values ranging from -1 to 1."},
    {"id": "doc:009", "title": "Redis JSON", "category": "database", "content": "Redis JSON allows storing, updating, and retrieving JSON documents in Redis. Documents are stored in a binary tree structure for fast access to nested elements via JSONPath expressions. You can store vector embeddings directly within JSON documents alongside metadata, enabling co-located storage where your vectors and business data live in the same place."},
    {"id": "doc:010", "title": "Redis Pipelining", "category": "performance", "content": "Redis pipelining sends multiple commands to the server in a single network round-trip, dramatically reducing latency for batch operations. Instead of waiting for each command's response before sending the next, pipelining buffers commands and sends them together. This is especially useful when ingesting large numbers of documents or embeddings into Redis."},
]


def get_redis_client() -> redis.Redis:
    """Create and verify a Redis connection."""
    try:
        client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
        client.ping()
        logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
        return client
    except redis.ConnectionError:
        raise RedisConnectionError(REDIS_HOST, REDIS_PORT)


def load_documents(client: redis.Redis) -> None:
    """Load all documents into Redis with their embeddings."""
    start = time.time()
    logger.info(f"Loading {len(DOCUMENTS)} documents into Redis...")
    texts = [doc["content"] for doc in DOCUMENTS]
    embeddings = get_embeddings_batch(texts)
    pipeline = client.pipeline()
    for doc, embedding in zip(DOCUMENTS, embeddings):
        doc_data = {"title": doc["title"], "category": doc["category"], "content": doc["content"], "embedding": embedding}
        pipeline.json().set(doc["id"], "$", doc_data)
    pipeline.execute()
    elapsed = time.time() - start
    logger.info(f"Loaded {len(DOCUMENTS)} documents in {elapsed:.2f}s")


def create_document_index(client: redis.Redis) -> None:
    """Create the vector search index for documents."""
    try:
        client.ft(INDEX_NAME).dropindex(delete_documents=False)
        logger.debug(f"Dropped existing index '{INDEX_NAME}'")
    except Exception:
        pass

    schema = (
        TextField("$.title", as_name="title"),
        TagField("$.category", as_name="category"),
        TextField("$.content", as_name="content"),
        VectorField(
            "$.embedding",
            "HNSW",
            {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"},
            as_name="vector",
        ),
    )
    definition = IndexDefinition(prefix=[DOC_PREFIX], index_type=IndexType.JSON)
    try:
        client.ft(INDEX_NAME).create_index(fields=schema, definition=definition)
        info = client.ft(INDEX_NAME).info()
        logger.info(f"Index '{INDEX_NAME}' created: {info['num_docs']} docs indexed")
    except Exception as e:
        raise SearchError(INDEX_NAME, str(e))


def create_cache_index(client: redis.Redis) -> None:
    """Create the vector search index for semantic cache."""
    try:
        client.ft(CACHE_INDEX_NAME).dropindex(delete_documents=False)
    except Exception:
        pass

    schema = (
        TextField("$.original_query", as_name="original_query"),
        TextField("$.response", as_name="response"),
        VectorField(
            "$.embedding",
            "FLAT",
            {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"},
            as_name="vector",
        ),
    )
    definition = IndexDefinition(prefix=[CACHE_PREFIX], index_type=IndexType.JSON)
    try:
        client.ft(CACHE_INDEX_NAME).create_index(
            fields=schema, definition=definition
        )
        logger.info(f"Cache index '{CACHE_INDEX_NAME}' created")
    except Exception as e:
        raise SearchError(CACHE_INDEX_NAME, str(e))

Perform KNN vector search

With the index in place, you can now search your documents by meaning rather than keywords. KNN (K-Nearest Neighbors) search finds the K vectors closest to your query vector in the embedding space.

• Open src/search.py and replace everything with the following code:

"""Vector search operations against Redis."""

import time

import numpy as np
import redis
from redis.commands.search.query import Query

from src import logger
from src.config import INDEX_NAME
from src.embeddings import get_embedding
from src.exceptions import SearchError


def vector_search(
    client: redis.Redis,
    query_text: str,
    top_k: int = 3,
    category_filter: str = None,
) -> list:
    """Perform KNN vector search with optional category filtering."""
    start = time.time()

    # Embed the query using the same model as the documents
    query_embedding = get_embedding(query_text)

    # Build filter expression: (*) for all docs, or a tag filter
    if category_filter:
        filter_expr = f"(@category:{{{category_filter}}})"
    else:
        filter_expr = "(*)"

    # Construct the KNN query
    query = (
        Query(f"{filter_expr}=>[KNN {top_k} @vector $query_vector AS vector_score]")
        .sort_by("vector_score")
        .return_fields("vector_score", "title", "content", "category")
        .dialect(2)
    )

    try:
        # Execute search with the query vector as bytes
        results = client.ft(INDEX_NAME).search(
            query,
            {"query_vector": np.array(query_embedding, dtype=np.float32).tobytes()},
        )
    except Exception as e:
        raise SearchError(INDEX_NAME, str(e))

    elapsed_ms = (time.time() - start) * 1000
    logger.info(
        f"Search completed: query='{query_text[:50]}...' "
        f"results={len(results.docs)} time={elapsed_ms:.1f}ms"
    )

    return results.docs

What does this code do?

• get_embedding(query_text) converts the user's question into a 1536-dim vector using the same text-embedding-3-small model that embedded the documents.
• The Query string (*)=>[KNN 3 @vector $query_vector AS vector_score] tells Redis to find the 3 nearest vectors to the query. The (*) means search all documents (no filter).
• np.array(..., dtype=np.float32).tobytes() converts the Python list of floats into raw bytes, which is the format Redis expects for vector queries.
• .dialect(2) is required for vector search queries in Redis.
• The vector_score field is cosine distance (not similarity). Lower scores mean more similar. You can convert to similarity with 1 - score.

• Save src/search.py.
• Test a basic vector search by running this command:

uv run python -c "
from src.indexer import get_redis_client, create_document_index
from src.search import vector_search
client = get_redis_client()
create_document_index(client)
results = vector_search(client, 'How does Redis handle many users at once?')
for doc in results:
    score = round(1 - float(doc.vector_score), 3)
    print(f'  [{score}] {doc.title}')
"

• You should see three documents returned, with "Redis Concurrency" scoring highest since it directly discusses concurrent connections.

Search returns no results or errors?

• Make sure the index was created first. The test command above calls create_document_index() before searching.
• Check that numpy is installed. If you see an import error, run uv add numpy.
• Verify your .env file has a valid OPENAI_API_KEY. The search function needs to call OpenAI to embed the query.

help me debug my Redis vector search

✔️ Awesome, I've got everything!

Great. Double check you've saved src/search.py before moving on.

ⓧ I'd like to double check the full code

"""Vector search operations against Redis."""

import time

import numpy as np
import redis
from redis.commands.search.query import Query

from src import logger
from src.config import INDEX_NAME
from src.embeddings import get_embedding
from src.exceptions import SearchError


def vector_search(
    client: redis.Redis,
    query_text: str,
    top_k: int = 3,
    category_filter: str = None,
) -> list:
    """Perform KNN vector search with optional category filtering."""
    start = time.time()

    query_embedding = get_embedding(query_text)

    if category_filter:
        filter_expr = f"(@category:{{{category_filter}}})"
    else:
        filter_expr = "(*)"

    query = (
        Query(f"{filter_expr}=>[KNN {top_k} @vector $query_vector AS vector_score]")
        .sort_by("vector_score")
        .return_fields("vector_score", "title", "content", "category")
        .dialect(2)
    )

    try:
        results = client.ft(INDEX_NAME).search(
            query,
            {"query_vector": np.array(query_embedding, dtype=np.float32).tobytes()},
        )
    except Exception as e:
        raise SearchError(INDEX_NAME, str(e))

    elapsed_ms = (time.time() - start) * 1000
    logger.info(
        f"Search completed: query='{query_text[:50]}...' "
        f"results={len(results.docs)} time={elapsed_ms:.1f}ms"
    )

    return results.docs

Add filtered vector search (hybrid search)

Your vector_search() function already supports hybrid search through the category_filter parameter. Hybrid search combines vector similarity with metadata filters so you can narrow results to a specific category before comparing embeddings.

What is hybrid search?

Pure vector search finds semantically similar documents regardless of metadata. Hybrid search adds structured filters on top. For example, the filter (@category:{ai})=>[KNN 3 @vector $query_vector AS vector_score] restricts the KNN search to only documents tagged with the ai category.

Redis, Weaviate, and Qdrant all support hybrid search natively as a core feature. Pinecone added metadata filtering later. This is a common production pattern because real queries often have structural constraints (search only my team's docs, only recent docs, only a specific topic).

• Test hybrid search by running this command:

uv run python -c "
from src.indexer import get_redis_client, create_document_index
from src.search import vector_search
client = get_redis_client()
create_document_index(client)
print('Filtered search (category=performance):')
results = vector_search(client, 'fast data access', category_filter='performance')
for doc in results:
    score = round(1 - float(doc.vector_score), 3)
    print(f'  [{score}] {doc.title}')
"

• You should see only documents from the performance category returned (like "Redis Caching Patterns", "Redis Pipelining", and "Redis Concurrency"). Documents tagged ai or database are excluded from the results.

Filtered search returning all documents?

• Check that the filter string uses curly braces around the value: (@category:{performance}). The braces are required for TagField filters in Redis.
• Verify that category is indexed as a TagField (not TextField) in your schema. Tag fields use exact matching. Text fields use full-text search.

help me fix my hybrid search filter

• Commit your progress by running these commands:

git add -A
git commit -m "feat: add vector index creation and search with hybrid filtering"

What does this commit message mean?

The feat: prefix signals that this commit adds new functionality. This conventional commit covers three new capabilities: document index creation, cache index creation, and vector search with hybrid filtering.

Your vector search is working. Next up, you'll wire it into a full RAG pipeline that retrieves relevant context, constructs a prompt, and generates answers with an LLM.

Build the RAG Pipeline with Semantic Caching

You have a vector index ready and a search function that finds relevant documents. The next piece is turning those search results into actual answers.

In this step, you will build the complete RAG pipeline: retrieve context from Redis, augment a prompt with that context, and generate an answer with gpt-4o-mini. You will also add a semantic cache that recognizes paraphrased questions and serves instant cached responses.

In this step, get ready to:

• Implement the Retrieve-Augment-Generate flow that turns user questions into LLM-powered answers.
• Add a semantic caching layer that detects similar queries and returns cached responses instantly.
• Wire up the entry point, run the full pipeline, and finalize your README.

Implement the Retrieve-Augment-Generate flow

A RAG pipeline has three stages. Retrieve finds the most relevant documents from your knowledge base. Augment constructs a prompt that includes the retrieved context alongside the user's question. Generate sends that augmented prompt to an LLM for a grounded answer.

• Open src/rag.py in your editor and replace the placeholder contents with the following code:

"""RAG pipeline orchestration: Retrieve, Augment, Generate."""

import time

import redis
from openai import OpenAI

from src import logger
from src.cache import cache_response, check_cache
from src.config import CHAT_MODEL, OPENAI_API_KEY
from src.search import vector_search

client = OpenAI(api_key=OPENAI_API_KEY)


def rag_query(
    redis_client: redis.Redis,
    question: str,
    use_cache: bool = True,
) -> str:
    """Execute the full RAG pipeline with optional semantic caching."""
    start = time.time()
    logger.info(f"RAG query: '{question}'")

    # Step 1: Check semantic cache
    if use_cache:
        cached = check_cache(redis_client, question)
        if cached:
            elapsed_ms = (time.time() - start) * 1000
            logger.info(f"Response served from cache in {elapsed_ms:.0f}ms")
            return cached

    # Step 2: Retrieve relevant documents
    results = vector_search(redis_client, question, top_k=3)

    context_parts = []
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        logger.debug(f"  Retrieved: {doc.title} (similarity={score})")
        context_parts.append(f"Title: {doc.title}\nContent: {doc.content}")

    context = "\n\n---\n\n".join(context_parts)

    # Step 3: Augment prompt
    system_prompt = (
        "You are a helpful assistant that answers questions based on the provided context. "
        "Only use information from the context to answer. If the context doesn't contain "
        "the answer, say so. Cite which document(s) you used."
    )

    user_prompt = f"""Context:\n{context}\n\nQuestion: {question}\n\nAnswer based on the context above:"""

    # Step 4: Generate
    response = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.3,
        max_tokens=500,
    )

    answer = response.choices[0].message.content

    # Cache the response for future similar queries
    if use_cache:
        cache_response(redis_client, question, answer)

    elapsed_ms = (time.time() - start) * 1000
    logger.info(f"CACHE_MISS: Generated response in {elapsed_ms:.0f}ms")

    return answer

What does this code do?

The rag_query function orchestrates the full pipeline in four stages:

• Cache check: before doing any expensive work, it asks the semantic cache if a similar question was already answered.
• Retrieve: uses your vector_search function from the previous step to find the top 3 most relevant documents.
• Augment: builds a structured prompt with the retrieved context, a system instruction that constrains the LLM to only cite the given documents, and the user's question.
• Generate: calls gpt-4o-mini with low temperature (0.3) for factual answers, caches the result, and logs timing.

• Save src/rag.py.

You won't be able to run this file yet because it imports from src/cache.py, which you will implement in the next substep.

Add semantic caching

Semantic caching is the optimization that makes this pipeline production-ready. Instead of calling the LLM every time, the cache detects when a new question is semantically similar to one already answered and serves the cached response instantly.

• Open src/cache.py and replace the placeholder contents with the check_cache function:

"""Semantic caching layer using Redis vector similarity."""

import time

import numpy as np
import redis
from redis.commands.search.query import Query

from src import logger
from src.config import CACHE_INDEX_NAME, CACHE_PREFIX, CACHE_SIMILARITY_THRESHOLD
from src.embeddings import get_embedding
from src.exceptions import CacheError


def check_cache(client: redis.Redis, query_text: str) -> str | None:
    """Check if a semantically similar query has a cached response."""
    start = time.time()

    try:
        query_embedding = get_embedding(query_text)

        query = (
            Query("(*)=>[KNN 1 @vector $query_vector AS vector_score]")
            .sort_by("vector_score")
            .return_fields("vector_score", "response", "original_query")
            .dialect(2)
        )

        results = client.ft(CACHE_INDEX_NAME).search(
            query,
            {"query_vector": np.array(query_embedding, dtype=np.float32).tobytes()},
        )

        if results.docs:
            score = float(results.docs[0].vector_score)
            similarity = 1 - score

            if similarity >= CACHE_SIMILARITY_THRESHOLD:
                elapsed_ms = (time.time() - start) * 1000
                logger.info(
                    f"CACHE_HIT: similarity={similarity:.3f} "
                    f"time={elapsed_ms:.1f}ms "
                    f"original='{results.docs[0].original_query[:50]}...'"
                )
                return results.docs[0].response

    except Exception:
        logger.debug("Cache lookup failed (cache may be empty)")

    return None

How does the cache lookup work?

The function embeds the incoming query and performs a KNN-1 search against the cache index (idx:cache). Redis returns the closest cached query along with a cosine distance score.

Cosine distance from Redis ranges from 0 (identical) to 2 (opposite). The code converts this to similarity with 1 - score. If similarity is at least 0.95 (the threshold from config.py), the cached response is returned immediately.

• Now add the cache_response function below check_cache in the same file:

def cache_response(
    client: redis.Redis, query_text: str, response_text: str
) -> None:
    """Store a query-response pair in the semantic cache."""
    try:
        query_embedding = get_embedding(query_text)
        cache_key = f"{CACHE_PREFIX}{abs(hash(query_text))}"

        cache_data = {
            "original_query": query_text,
            "response": response_text,
            "embedding": query_embedding,
        }
        client.json().set(cache_key, "$", cache_data)
        logger.debug(f"Cached response for: '{query_text[:50]}...'")

    except Exception as e:
        raise CacheError(str(e))

What does cache_response do?

After the LLM generates an answer, this function stores the query-response pair in Redis as a JSON document at a key like cache:123456789. The document contains the original query text, the LLM response, and the query's embedding vector.

Future queries that are semantically similar (above 0.95 cosine similarity) will match this cached entry and skip the LLM call entirely.

• Save src/cache.py.

Why is Redis ideal for semantic caching?

Redis stores everything in memory, so the cache lookup itself takes sub-millisecond time. You do not need a separate caching service in front of your vector database because Redis is both at once.

This is the pattern that makes Redis stand out for AI workloads. A typical LLM call takes 2-3 seconds. A cached response from Redis takes around 50ms (dominated by the embedding call, not Redis itself). Many users can hit this cache concurrently without degradation thanks to Redis's event loop.

Seeing an import error?

If your editor shows a red underline on from src.cache import cache_response, check_cache in src/rag.py, make sure you saved src/cache.py first. The import resolves once both files are saved.

help me debug an import error in my RAG pipeline

✔️ Awesome, I've got everything!

Great. Make sure both src/rag.py and src/cache.py are saved.

ⓧ I'd like to double check the full code

Here is the complete src/cache.py:

"""Semantic caching layer using Redis vector similarity."""

import time

import numpy as np
import redis
from redis.commands.search.query import Query

from src import logger
from src.config import CACHE_INDEX_NAME, CACHE_PREFIX, CACHE_SIMILARITY_THRESHOLD
from src.embeddings import get_embedding
from src.exceptions import CacheError


def check_cache(client: redis.Redis, query_text: str) -> str | None:
    """Check if a semantically similar query has a cached response."""
    start = time.time()

    try:
        query_embedding = get_embedding(query_text)

        query = (
            Query("(*)=>[KNN 1 @vector $query_vector AS vector_score]")
            .sort_by("vector_score")
            .return_fields("vector_score", "response", "original_query")
            .dialect(2)
        )

        results = client.ft(CACHE_INDEX_NAME).search(
            query,
            {"query_vector": np.array(query_embedding, dtype=np.float32).tobytes()},
        )

        if results.docs:
            score = float(results.docs[0].vector_score)
            similarity = 1 - score

            if similarity >= CACHE_SIMILARITY_THRESHOLD:
                elapsed_ms = (time.time() - start) * 1000
                logger.info(
                    f"CACHE_HIT: similarity={similarity:.3f} "
                    f"time={elapsed_ms:.1f}ms "
                    f"original='{results.docs[0].original_query[:50]}...'"
                )
                return results.docs[0].response

    except Exception:
        logger.debug("Cache lookup failed (cache may be empty)")

    return None


def cache_response(
    client: redis.Redis, query_text: str, response_text: str
) -> None:
    """Store a query-response pair in the semantic cache."""
    try:
        query_embedding = get_embedding(query_text)
        cache_key = f"{CACHE_PREFIX}{abs(hash(query_text))}"

        cache_data = {
            "original_query": query_text,
            "response": response_text,
            "embedding": query_embedding,
        }
        client.json().set(cache_key, "$", cache_data)
        logger.debug(f"Cached response for: '{query_text[:50]}...'")

    except Exception as e:
        raise CacheError(str(e))

And here is the complete src/rag.py:

"""RAG pipeline orchestration: Retrieve, Augment, Generate."""

import time

import redis
from openai import OpenAI

from src import logger
from src.cache import cache_response, check_cache
from src.config import CHAT_MODEL, OPENAI_API_KEY
from src.search import vector_search

client = OpenAI(api_key=OPENAI_API_KEY)


def rag_query(
    redis_client: redis.Redis,
    question: str,
    use_cache: bool = True,
) -> str:
    """Execute the full RAG pipeline with optional semantic caching."""
    start = time.time()
    logger.info(f"RAG query: '{question}'")

    # Step 1: Check semantic cache
    if use_cache:
        cached = check_cache(redis_client, question)
        if cached:
            elapsed_ms = (time.time() - start) * 1000
            logger.info(f"Response served from cache in {elapsed_ms:.0f}ms")
            return cached

    # Step 2: Retrieve relevant documents
    results = vector_search(redis_client, question, top_k=3)

    context_parts = []
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        logger.debug(f"  Retrieved: {doc.title} (similarity={score})")
        context_parts.append(f"Title: {doc.title}\nContent: {doc.content}")

    context = "\n\n---\n\n".join(context_parts)

    # Step 3: Augment prompt
    system_prompt = (
        "You are a helpful assistant that answers questions based on the provided context. "
        "Only use information from the context to answer. If the context doesn't contain "
        "the answer, say so. Cite which document(s) you used."
    )

    user_prompt = f"""Context:\n{context}\n\nQuestion: {question}\n\nAnswer based on the context above:"""

    # Step 4: Generate
    response = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.3,
        max_tokens=500,
    )

    answer = response.choices[0].message.content

    # Cache the response for future similar queries
    if use_cache:
        cache_response(redis_client, question, answer)

    elapsed_ms = (time.time() - start) * 1000
    logger.info(f"CACHE_MISS: Generated response in {elapsed_ms:.0f}ms")

    return answer

Test and demonstrate the complete pipeline

Now you will create the entry point that wires everything together. It will demo vector search, then show the RAG pipeline generating an answer, then prove the cache works by asking a paraphrased question.

• Open main.py in the project root and replace the placeholder contents with the following code:

"""Entry point for the Redis RAG Pipeline."""

from src import logger
from src.cache import check_cache
from src.indexer import (
    create_cache_index,
    create_document_index,
    get_redis_client,
    load_documents,
)
from src.rag import rag_query
from src.search import vector_search


def main():
    """Run the Redis RAG pipeline demonstration."""
    logger.info("=" * 50)
    logger.info("Redis RAG Pipeline - Starting")
    logger.info("=" * 50)

    # Initialize: connect, load docs, create indexes
    client = get_redis_client()
    load_documents(client)
    create_document_index(client)
    create_cache_index(client)

    # Demo: Vector Search
    logger.info("-" * 50)
    logger.info("Demo: Vector Search")
    logger.info("-" * 50)

    print("\nQuery: 'How does Redis handle many users at once?'")
    results = vector_search(client, "How does Redis handle many users at once?")
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        print(f"  [{score}] {doc.title}")

    print("\nQuery: 'fast data access' (filtered: category=performance)")
    results = vector_search(client, "fast data access", category_filter="performance")
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        print(f"  [{score}] {doc.title}")

    # Demo: RAG Pipeline with Caching
    logger.info("-" * 50)
    logger.info("Demo: RAG Pipeline with Semantic Caching")
    logger.info("-" * 50)

    print("\n--- First query (cache miss, will generate) ---")
    answer = rag_query(client, "What makes Redis good for AI applications?")
    print(f"\nAnswer: {answer}")

    print("\n--- Similar query (should hit cache) ---")
    answer = rag_query(client, "Why is Redis suitable for AI workloads?")
    print(f"\nAnswer: {answer}")

    logger.info("=" * 50)
    logger.info("Pipeline complete. Try your own queries!")
    logger.info("=" * 50)


if __name__ == "__main__":
    main()

What does main.py demonstrate?

The script runs two demos back-to-back. First, it performs raw vector searches (one unfiltered, one filtered by category) so you can see retrieval in action.

Second, it asks a question through the full RAG pipeline (cache miss, generates with LLM), then immediately asks a paraphrased version. The second query should hit the semantic cache and return in a fraction of the time.

• Save main.py.
• Run the full pipeline by executing this command in your terminal:

uv run main.py

• Watch the structured log output. You should see a CACHE_MISS log line for the first RAG query (taking 2-3 seconds), followed by a CACHE_HIT log line for the paraphrased query (returning almost instantly).

What about concurrency?

Redis uses a single-threaded event loop for command processing. This eliminates race conditions and lock contention. Despite being single-threaded, Redis handles tens of thousands of concurrent client connections using I/O multiplexing.

This means many users could hit your RAG pipeline simultaneously without performance degradation. The semantic cache stays consistent without any locking code on your side.

Pipeline failing with a connection error?

Make sure your Redis container is still running. If it stopped, restart it:

docker start redis-rag

If you see an OpenAI authentication error, double-check that your .env file contains a valid API key and that OPENAI_API_KEY is set correctly.

help me debug my Redis RAG pipeline

Now finalize your project with a professional README. This documents the architecture, setup instructions, and key concepts for anyone who clones your repository.

• Open README.md in your project root and replace any placeholder contents with the following:

# Redis RAG Pipeline

A production-pattern Retrieval-Augmented Generation system using Redis as both a vector database and semantic cache.

## Why Redis for RAG?

Redis collapses what would normally be 3 separate services into one:
- **Vector Database**: Store and search embeddings with HNSW indexing
- **Semantic Cache**: Recognize similar queries and serve cached LLM responses instantly
- **Session Store**: Handle thousands of concurrent users with sub-millisecond latency

## Architecture

```
User Query -> Embed Query -> Check Semantic Cache
                                |
                         [Cache Hit] -> Return cached response (~50ms)
                                |
                         [Cache Miss] -> Vector Search (KNN)
                                            |
                                      Retrieve Top-K Docs
                                            |
                                      Augment Prompt
                                            |
                                      Generate (LLM)
                                            |
                                      Cache Response
                                            |
                                      Return to User (~2-3s)
```

## Quick Start

### Prerequisites
- Docker Desktop
- Python 3.11+
- OpenAI API key

### Setup

1. Start Redis:
```bash
docker run -d --name redis-rag -p 6379:6379 redis:latest
```

2. Install dependencies:
```bash
uv add redis openai numpy python-dotenv
```

3. Configure environment:
```bash
cp .env.example .env
# Edit .env with your OpenAI API key
```

4. Run:
```bash
uv run main.py
```

## Project Structure

```
redis-rag/
├── src/
│   ├── __init__.py      # Logging configuration
│   ├── config.py        # Environment and constants
│   ├── exceptions.py    # Custom exception classes
│   ├── embeddings.py    # OpenAI embedding generation
│   ├── indexer.py       # Redis index creation and document loading
│   ├── search.py        # Vector search operations
│   ├── cache.py         # Semantic caching layer
│   └── rag.py           # RAG pipeline orchestration
├── .env.example
├── .gitignore
├── main.py              # Entry point
├── pyproject.toml
└── README.md
```

## Key Concepts

- **Vector Embeddings**: Text converted to 1536-dim vectors via OpenAI text-embedding-3-small
- **HNSW Index**: Approximate nearest neighbor search with 95-99% recall
- **Cosine Similarity**: Distance metric for semantic relatedness
- **Semantic Cache**: Vector similarity matching on queries to serve cached LLM responses
- **Hybrid Search**: Combine vector similarity with metadata filters

## Cost

Approximately $0.01-0.05 per session (OpenAI API for embeddings + generation).

## License

MIT

• Save README.md.
• Commit the pipeline and README using conventional commit format by running these commands:

git add src/cache.py src/rag.py main.py
git commit -m "feat: add RAG pipeline with semantic caching"

git add README.md
git commit -m "docs: add README with setup instructions and architecture"

Why two separate commits?

Conventional commits separate concerns. The feat: commit adds functionality (the pipeline code). The docs: commit adds documentation. This makes your git history easy to scan and enables tools that auto-generate changelogs from commit messages.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all files and both commits are in your git log.

ⓧ I'd like to double check the full code

Here is the complete main.py:

"""Entry point for the Redis RAG Pipeline."""

from src import logger
from src.cache import check_cache
from src.indexer import (
    create_cache_index,
    create_document_index,
    get_redis_client,
    load_documents,
)
from src.rag import rag_query
from src.search import vector_search


def main():
    """Run the Redis RAG pipeline demonstration."""
    logger.info("=" * 50)
    logger.info("Redis RAG Pipeline - Starting")
    logger.info("=" * 50)

    client = get_redis_client()
    load_documents(client)
    create_document_index(client)
    create_cache_index(client)

    logger.info("-" * 50)
    logger.info("Demo: Vector Search")
    logger.info("-" * 50)

    print("\nQuery: 'How does Redis handle many users at once?'")
    results = vector_search(client, "How does Redis handle many users at once?")
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        print(f"  [{score}] {doc.title}")

    print("\nQuery: 'fast data access' (filtered: category=performance)")
    results = vector_search(client, "fast data access", category_filter="performance")
    for doc in results:
        score = round(1 - float(doc.vector_score), 3)
        print(f"  [{score}] {doc.title}")

    logger.info("-" * 50)
    logger.info("Demo: RAG Pipeline with Semantic Caching")
    logger.info("-" * 50)

    print("\n--- First query (cache miss, will generate) ---")
    answer = rag_query(client, "What makes Redis good for AI applications?")
    print(f"\nAnswer: {answer}")

    print("\n--- Similar query (should hit cache) ---")
    answer = rag_query(client, "Why is Redis suitable for AI workloads?")
    print(f"\nAnswer: {answer}")

    logger.info("=" * 50)
    logger.info("Pipeline complete. Try your own queries!")
    logger.info("=" * 50)


if __name__ == "__main__":
    main()

Your RAG pipeline is complete. You have a working system that retrieves relevant context from Redis, generates grounded answers with an LLM, and automatically caches responses for similar future queries. Next up, the Secret Mission challenges you to add reranking for even better retrieval quality.

Secret mission

Add Reranking

Your RAG pipeline retrieves documents by vector similarity alone. Sometimes the closest vector isn't the most relevant answer. Add an LLM-based reranking step that scores each retrieved document for relevance and picks the true top results before generating an answer.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs locally with a Docker container, so there are no ongoing cloud costs.

Resources you used:

• A Docker container named redis-rag running Redis (free, local).
• OpenAI API usage: approximately $0.01-0.05 depending on how many queries you ran (already spent, no recurring charge).
• Local project folder redis-rag on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with different queries.

• Your Docker container continues running in the background and uses minimal system resources.
• Your .env file and project code remain on your Desktop, ready for future sessions.
• No ongoing API costs unless you run the pipeline again.

Pause - I'll come back to this later

Stop the Docker container to free up memory. Your data persists inside the container, so you can restart it later without re-loading documents.

• Stop the Redis container by running:

docker stop redis-rag

• When you're ready to resume, restart it by running:

docker start redis-rag

Your documents, embeddings, and cached responses will still be there when you come back.

Delete - I don't want to use this again

Remove all project resources and start fresh. No cloud resources to clean up.

• Remove the Docker container (this also stops it if running) by running:

docker rm -f redis-rag

• Delete the project folder from your Desktop. On macOS or Linux, run:

rm -rf ~/Desktop/redis-rag

• On Windows, run:

rmdir /s /q %USERPROFILE%\Desktop\redis-rag

Your OpenAI API key remains in your OpenAI account. No action needed there unless you want to revoke it from the OpenAI dashboard.

Nice Work!

Nice Work!

You just built a production-grade RAG pipeline powered by Redis vector search and semantic caching. This is the same architecture pattern that teams use to serve AI-powered answers at scale without burning through LLM budgets.

You've learned how to:

• Built a Redis-backed vector search system with HNSW indexing, KNN queries, and hybrid metadata filtering.
• Created a complete RAG pipeline (Retrieve, Augment, Generate) with a semantic caching layer that recognizes similar queries and serves cached responses in milliseconds instead of seconds.
• Structured a professional Python project with production patterns: structured logging, custom exception classes, environment-based configuration, conventional commits, and a GitHub-ready repository layout.
• Secret Mission: Added LLM-based reranking to improve retrieval precision before passing context to the generation model.

Ready to quiz yourself?