Build a RAG Tool with AWS Bedrock

Build a RAG pipeline that answers questions with source citations.

Introduction

30 Second Summary

You have a folder full of reports, notes, or research papers. Finding the answer to a specific question means opening each file and searching through it manually.

In this project, you will build a command-line RAG tool that ingests any text documents, embeds them using AWS Bedrock, and answers your questions with citations pointing back to the exact source file and passage.

What You'll Build

You drop your own documents into a folder, ask a question in plain English, and get back a grounded answer with numbered citations pointing to the exact file and passage it came from.

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

  • A document ingestion pipeline where you drop any text files into a folder, run one command, and the tool chunks and indexes everything automatically.
  • A question-answering system powered by vector embeddings and FAISS that finds the most relevant passages across all your files and generates answers with numbered citations tracing back to each source file.
  • A cross-document retrieval fix that handles questions spanning multiple files, built after you hit and diagnose a real production failure mode.
  • Secret Mission: Build a retrieval eval framework that scores retrieval quality and catches silent failures.

Are there any prerequisites?

You'll need an AWS account with free credits, Python installed on your machine, and VS Code ready to go. Step 1 walks you through installing everything else.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Environment

Before you write any code, you need the tools and credentials in place to talk to AWS Bedrock. Without these, none of the later steps will work.

In this step, you will install the Python packages that power embedding, vector search, and AWS calls. You will also configure AWS CLI credentials and create the project folder with sample documents to feed your RAG pipeline.

In this step, get ready to:
  • Install Python dependencies and verify all imports work.
  • Configure AWS CLI credentials and confirm Bedrock model access.
  • Create the project directory with sample documents and a configuration file.
Check your Python version and install dependencies

The faiss-cpu library requires Python 3.10 or higher. Before installing anything, confirm you have the right version.

  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

Your Python version is compatible. Move on to installing the dependencies below.

ⓧ I see a version lower than 3.10

You need Python 3.10 or higher for faiss-cpu to work. On macOS, the system Python is often an older version even after you install a newer one.

  • Download and install Python 3.10+ from the official Python downloads page.
  • After installing, try running python3.12 --version (or whichever version you installed) to make sure the new binary is available.
  • Use that specific binary (e.g. python3.12) for all python3 commands in this project.

ⓧ Command not found

Python is not installed or not in your PATH.

  • Download and install Python 3.10+ from the official Python downloads page.
  • Restart your terminal after installation.
  • Run python3 --version again to confirm it prints a version number of 3.10 or higher.
  • Install the three packages your project needs by running this command:
pip install boto3 faiss-cpu numpy

What do these packages do?

  • boto3 is the AWS SDK for Python. It lets you call Bedrock APIs to generate embeddings and text.
  • faiss-cpu is Facebook's vector search library. It stores embeddings and finds the most similar ones to a query.
  • numpy handles the numerical arrays that FAISS expects for its vectors.

Before you move on, what do you think will happen if one of these packages did not install correctly?

  • Verify all three packages imported successfully by running this command:
python3 -c "import boto3, faiss, numpy; print('All imports OK')"

You should see All imports OK printed in your terminal.

Seeing an import error?

  • If faiss fails to import, check that your Python version is 3.10 or higher. faiss-cpu does not support older versions.
  • If pip installed packages to a different Python version than the one you are running, try python3 -m pip install boto3 faiss-cpu numpy instead to make sure they land in the right place.
  • help me debug my import error for boto3, faiss, or numpy
Configure AWS CLI and confirm Bedrock access

Your Python code uses boto3 to call AWS Bedrock APIs. For boto3 to authenticate, it reads credentials from the AWS CLI configuration on your machine.

  • Check whether the AWS CLI is already installed by running this command:
aws --version

✔️ I see a version number

The AWS CLI is installed. Move on to configuring your credentials below.

ⓧ Command not found

The AWS CLI is not installed yet.

  • Follow the official AWS CLI install guide for your operating system.
  • Restart your terminal after installation.
  • Run aws --version again to confirm it prints a version number.
  • Configure your AWS credentials by running this command:
aws configure

What does aws configure ask for?

The command prompts you for four values. Enter them one at a time:

  • AWS Access Key ID and AWS Secret Access Key come from your IAM user in the AWS console. If you do not have these yet, create an IAM user with programmatic access.
  • Default region name: enter us-east-1. Both Titan Text Embeddings V2 and Nova Micro are available in this region.
  • Default output format: enter json.

Do I need to enable Bedrock model access manually?

Amazon Bedrock now provides automatic access to serverless foundation models in your region. Since this project uses Amazon's own models (Titan and Nova), you do not need to manually enable them.

You will verify that your credentials and access work in the next substep when you run a test embedding call.

Create the project directory and verify Bedrock connectivity
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rag-bedrock
cd rag-bedrock
  • Create the docs/ folder and a data/ folder by running this command:
mkdir docs data
  • Confirm both folders exist by running:
ls

You should see data and docs in the output.

Now you need sample text files for your RAG tool to ingest. You can write about anything you like. Create 2-3 short .txt files on different subjects so the cross-document retrieval in Step 5 is observable.

  • Create your first sample document by running this command:
cat > docs/solar_system.txt << 'EOF'
The solar system consists of the Sun and the objects that orbit it. The eight planets in order from the Sun are Mercury, Venus, Earth, Mars, Jupiter, Saturn, Uranus, and Neptune. Earth is the third planet and the only one known to support life. Jupiter is the largest planet, with a mass more than twice that of all other planets combined. Saturn is famous for its prominent ring system made of ice and rock particles.
EOF
  • Create a second sample document on a different topic by running this command:
cat > docs/cooking_basics.txt << 'EOF'
Cooking is the process of preparing food by applying heat. The three main cooking methods are dry heat (roasting, grilling, baking), moist heat (boiling, steaming, poaching), and combination methods (braising, stewing). Seasoning with salt enhances flavor by suppressing bitterness. Mise en place, a French term meaning everything in its place, is the practice of preparing and organizing all ingredients before cooking begins.
EOF
  • Create a third sample document by running this command:
cat > docs/exercise_science.txt << 'EOF'
Regular physical exercise improves cardiovascular health, strengthens muscles, and boosts mental well-being. Aerobic exercises like running, swimming, and cycling improve heart and lung capacity. Resistance training builds muscle strength and increases bone density. The American Heart Association recommends at least 150 minutes of moderate aerobic activity per week. Rest days between intense sessions allow muscles to recover and grow stronger.
EOF
  • Confirm all three files landed in the docs/ folder by running:
ls docs/

You should see cooking_basics.txt, exercise_science.txt, and solar_system.txt listed.

Use your own documents

These sample files are just starting points. Feel free to replace them with text about topics you actually care about. The RAG tool works on any .txt files.

Next, create the configuration file that all your Python scripts will share.

  • Open VS Code in your project folder by running this command:
code .
  • In the Explorer sidebar in VS Code, click the new file icon next to rag-bedrock.
  • Name the file config.py and press Enter.
  • Copy the following code into config.py:
import boto3

# AWS region where both Titan and Nova models are available
REGION = "us-east-1"

# Model IDs for embedding and generation
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"
GEN_MODEL_ID = "amazon.nova-micro-v1:0"

# Embedding vector dimensions (Titan V2 default)
EMBED_DIMENSIONS = 1024

# Chunking parameters (characters)
CHUNK_SIZE = 2000
CHUNK_OVERLAP = 200

# Number of chunks to retrieve
TOP_K = 3
TOP_K_RERANK = 10

# Shared Bedrock client used by all scripts
bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION)

What does this configuration do?

  • EMBED_MODEL_ID points to Amazon Titan Text Embeddings V2, which converts text into 1024-dimensional vectors.
  • GEN_MODEL_ID points to Amazon Nova Micro, the generation model that will synthesize answers from retrieved context.
  • CHUNK_SIZE and CHUNK_OVERLAP control how documents are split into passages. Overlapping chunks prevent context from being cut off at boundaries.
  • bedrock_runtime creates a single boto3 client that every script in the project will import and reuse.
  • Save config.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).

Now verify that your AWS credentials work and Bedrock is reachable. What do you expect to see if everything is configured correctly?

  • Test Bedrock connectivity by running this command from your rag-bedrock folder:
python3 -c "
import json
from config import bedrock_runtime, EMBED_MODEL_ID, EMBED_DIMENSIONS
body = json.dumps({'inputText': 'hello world', 'dimensions': EMBED_DIMENSIONS, 'normalize': True})
response = bedrock_runtime.invoke_model(modelId=EMBED_MODEL_ID, body=body)
result = json.loads(response['body'].read())
print(f'Embedding dimensions: {len(result[\"embedding\"])}')
print('Bedrock connection successful!')
"

You should see output showing Embedding dimensions: 1024 and Bedrock connection successful! printed in your terminal.

Seeing an access denied or credentials error?

  • Make sure you ran aws configure and entered valid access keys with the region set to us-east-1.
  • Confirm your IAM user has the AmazonBedrockFullAccess policy attached. You can check this in the IAM console under your user's permissions.
  • If you see a timeout error, check your internet connection and try again.
  • help me fix my Bedrock access denied or credentials error

✔️ Awesome, I've got everything!

Great. Double check that you have saved config.py.

ⓧ I'd like to double check the full code

Here is the complete config.py file:

import boto3

REGION = "us-east-1"
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"
GEN_MODEL_ID = "amazon.nova-micro-v1:0"
EMBED_DIMENSIONS = 1024
CHUNK_SIZE = 2000
CHUNK_OVERLAP = 200
TOP_K = 3
TOP_K_RERANK = 10

bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION)

Your environment is ready. You have Python dependencies installed, AWS credentials configured, sample documents in place, and a verified connection to Bedrock. Next up, you will write the ingestion script that turns those raw text files into searchable chunks.

Ingest and Chunk Any Document

Your environment is configured and Bedrock is responding to API calls. The next challenge is preparing your documents so the RAG pipeline can actually use them.

A raw text file is too large to embed in one shot. Titan Text Embeddings V2 accepts a maximum of 8,192 tokens per request, so you need to split documents into smaller passages that preserve context. The key design decision is generality: your ingestion must work on any text file dropped into the docs/ folder, not just one hardcoded source.

In this step, get ready to:
  • Write a loader that scans any folder and reads every .txt file with its filename as metadata.
  • Build a chunking function that splits documents into overlapping passages for embedding.
  • Run the ingestion pipeline and inspect the output to verify it makes sense.
Write the document loading function

The first piece of the ingestion pipeline is a function that scans a folder and reads every .txt file it finds. Each document is stored alongside its filename so you can trace any chunk back to its original source later.

  • Create a new file called ingest.py in your project folder, then paste the following code:
import os
import json
from config import CHUNK_SIZE, CHUNK_OVERLAP

DOCS_DIR = "docs"
CHUNKS_PATH = "data/chunks.json"


def load_documents(docs_dir):
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r", encoding="utf-8") as f:
                text = f.read()
            documents.append({"filename": filename, "text": text})
            print(f"  Loaded {filename}: {len(text)} characters")
    return documents

What does this code do?

  • DOCS_DIR and CHUNKS_PATH are constants that set where the tool reads documents from and where it saves the processed chunks.
  • load_documents() scans the given directory, reads every .txt file, and stores each one as a dictionary with the filename and full text. The filename is critical. It becomes the metadata that lets you trace any chunk back to its source.
  • sorted() ensures files are always loaded in alphabetical order, making the output deterministic regardless of operating system.
  • Save ingest.py.
Build the chunking function

Loading whole documents is only half the job. Titan Text Embeddings V2 has a 50,000-character input limit per request, and even within that limit, smaller passages produce better embeddings because each vector represents a focused idea instead of an entire document.

  • In ingest.py, add the following function below load_documents():
def chunk_text(text, source_file, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append({
                "text": chunk.strip(),
                "source_file": source_file,
                "start_char": start,
                "end_char": min(end, len(text)),
            })
        start = end - overlap
    return chunks

What does this code do?

  • chunk_text() walks through the text with a sliding window. Each window is CHUNK_SIZE characters long (2,000 by default from your config.py).
  • The overlap (CHUNK_OVERLAP = 200 characters) means consecutive chunks share their last 200 characters. This prevents important context from being split across a chunk boundary and lost.
  • Each chunk stores source_file, start_char, and end_char metadata. This is what makes citation tracing possible later. You will be able to point any generated answer back to the exact file and character range.
  • Save ingest.py.
  • Verify that the loader works by running this quick test in your terminal:
python -c "from ingest import load_documents; load_documents('docs')"

You should see one line per file in your docs/ folder, each showing the filename and its character count.

Seeing a ModuleNotFoundError?

  • Make sure you are running the command from your project folder (the one containing config.py and ingest.py).
  • Check that config.py exists in the same directory and contains the CHUNK_SIZE and CHUNK_OVERLAP constants from Step 1.
  • If no files are listed, confirm your docs/ folder contains at least one .txt file.

Why isn't my ingest.py finding my documents or config?

Run the full ingestion pipeline

The loader and chunker are individual functions. Now you need an orchestrator that ties them together: load every document, chunk each one, assign a unique ID to every chunk, and save the results.

  • In ingest.py, add the ingest_all() function below chunk_text():
def ingest_all(docs_dir):
    documents = load_documents(docs_dir)
    if not documents:
        print(f"No .txt files found in {docs_dir}/. Add some text files and try again.")
        return []

    all_chunks = []
    for doc in documents:
        doc_chunks = chunk_text(doc["text"], doc["filename"])
        all_chunks.extend(doc_chunks)
        print(f"  {doc['filename']}: {len(doc_chunks)} chunks")

    for i, chunk in enumerate(all_chunks):
        chunk["chunk_id"] = i

    return all_chunks

What does this code do?

  • ingest_all() calls load_documents() and then passes each document through chunk_text(). It collects every chunk into a single list.
  • Each chunk gets a globally unique chunk_id (a simple incrementing integer). This ID is how the vector index and the RAG query will reference specific chunks later.
  • If the folder is empty, the function prints a helpful message instead of crashing. Defensive code like this is what separates a tool you can hand to someone from a script that only works on your machine.
  • Now add the main execution block at the very bottom of ingest.py, below ingest_all():
if __name__ == "__main__":
    os.makedirs("data", exist_ok=True)
    print(f"Loading documents from {DOCS_DIR}/...")
    chunks = ingest_all(DOCS_DIR)

    if chunks:
        with open(CHUNKS_PATH, "w", encoding="utf-8") as f:
            json.dump(chunks, f, indent=2)
        print(f"\nCreated {len(chunks)} chunks total")
        print(f"Sample chunk from {chunks[0]['source_file']}: \"{chunks[0]['text'][:100]}...\"")
        print(f"Saved to {CHUNKS_PATH}")

What does this code do?

  • os.makedirs("data", exist_ok=True) creates the data/ directory if it does not already exist. This is where all pipeline output (chunks, index, metadata) will live.
  • json.dump() serializes every chunk to data/chunks.json so the embedding step can read them back without re-processing the raw documents.
  • The final print statements show a quick summary: total chunk count and a 100-character preview of the first chunk. This lets you sanity-check the output at a glance.
  • Save ingest.py.

Before you run this, think about how many chunks you expect. If your documents are short (a few hundred characters each), you might get one chunk per file. If they are longer, the 2,000-character window should produce several chunks per file. What do you predict?

  • Run the ingestion pipeline:
python ingest.py

You should see output showing each file loaded with its character count, followed by chunk counts per file, a total chunk count, and a sample chunk preview. It looks something like this:

Loading documents from docs/...
  Loaded notes.txt: 3200 characters
  Loaded research.txt: 5100 characters
  notes.txt: 2 chunks
  research.txt: 3 chunks

Created 5 chunks total
Sample chunk from notes.txt: "The quarterly results showed..."
Saved to data/chunks.json

What should I see?

Your exact numbers will differ based on the documents you chose. The important things to confirm are: every .txt file in your docs/ folder appears in the output, each file shows a chunk count, and the sample chunk preview contains recognizable text from your documents.

Seeing zero chunks or a missing file?

  • If you see "No .txt files found", check that your files have the .txt extension (not .text or .md). The loader only reads .txt files.
  • If a file is missing from the output, make sure it is inside the docs/ folder (not a subfolder or the project root).
  • If you see a FileNotFoundError, confirm the docs/ folder exists in the same directory where you are running the script.

Why is my ingestion pipeline not finding or chunking my documents?

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest.py and that data/chunks.json exists in your project folder.

ⓧ I'd like to double check the full code

Your ingest.py should contain these sections in order. Scroll up and confirm each one is present:

  • Imports (os, json, CHUNK_SIZE and CHUNK_OVERLAP from config) and the DOCS_DIR / CHUNKS_PATH constants.
  • load_documents(docs_dir) function that scans for .txt files and returns a list of dictionaries.
  • chunk_text(text, source_file, ...) function that splits text into overlapping chunks with metadata.
  • ingest_all(docs_dir) function that orchestrates loading, chunking, and assigning chunk_id values.
  • if __name__ == "__main__": block that creates the data/ directory, runs ingest_all(), and saves data/chunks.json.

Your documents are loaded, split, and saved. Next up, you will turn these text chunks into vector embeddings using AWS Bedrock and build a searchable FAISS index so you can find the most relevant passages for any question.

Build the Vector Index with Bedrock Embeddings

Your documents are chunked and stored in chunks.json. Each chunk is a passage of text with metadata tracking where it came from.

Right now, there's no way to search those chunks by meaning. If someone asks "What were the quarterly results?", a keyword search won't help if the chunk says "revenue grew 15% year-over-year" without using the word "quarterly." Vector embeddings solve this by converting each chunk into a 1024-dimensional number array. Chunks with similar meaning end up close together in vector space, so a question finds the right passages by mathematical similarity.

In this step, get ready to:
  • Write functions to embed text chunks using Bedrock's Titan Text Embeddings V2.
  • Build a FAISS vector index and save it to disk.
  • Test retrieval by querying the index with a sample question.
Write the embedding functions

You need a function that takes a chunk of text and sends it to Bedrock's Titan Text Embeddings V2 model. The model returns a 1024-dimensional vector representing that text's meaning.

  • Create a new file called embed_index.py in your project folder.
  • Add the imports, file paths, and the core embedding function by pasting this code:
import json
import time
import numpy as np
import faiss
from config import bedrock_runtime, EMBED_MODEL_ID, EMBED_DIMENSIONS

INDEX_PATH = "data/index.faiss"
CHUNKS_PATH = "data/chunks.json"


def get_embedding(text):
    # Build the request body for Titan Text Embeddings V2
    body = json.dumps({
        "inputText": text[:50000],
        "dimensions": EMBED_DIMENSIONS,
        "normalize": True,
    })
    # Call Bedrock to get the embedding vector
    response = bedrock_runtime.invoke_model(
        modelId=EMBED_MODEL_ID,
        body=body,
    )
    result = json.loads(response["body"].read())
    return result["embedding"]

What does this code do?

The imports pull in numpy for array math, faiss for vector search, and the Bedrock client and constants you set up in config.py.

get_embedding() sends a chunk's text to Titan Text Embeddings V2 and gets back a 1024-dimensional vector. The text[:50000] slice caps input at 50,000 characters, which is the model's maximum input length.

  • Save embed_index.py.
  • Verify the embedding function works by running this quick test:
python -c "from embed_index import get_embedding; emb = get_embedding('test query'); print(f'Embedding dimensions: {len(emb)}')"

You should see Embedding dimensions: 1024 printed. This confirms your Bedrock connection is working and the model returns vectors of the expected size.

Seeing an error instead of 1024?

  • If you see a ModuleNotFoundError, make sure you installed all dependencies with pip install boto3 faiss-cpu numpy in Step 1.
  • If you see an AccessDeniedException, double check that you enabled model access for Amazon Titan Text Embeddings V2 in the Bedrock console.
  • If you see a credentials error, confirm your AWS CLI is configured for us-east-1 by running aws configure list.

Help me debug my Bedrock embedding call

Now you need a function that loops through every chunk, embeds it, and handles any rate-limiting errors from the API.

  • In embed_index.py, add the embed_chunks() function below the get_embedding() function:
def embed_chunks(chunks, batch_delay=0.1):
    embeddings = []
    for i, chunk in enumerate(chunks):
        try:
            emb = get_embedding(chunk["text"])
            embeddings.append(emb)
        except Exception as e:
            # If the API throttles us, wait and retry once
            print(f"Error embedding chunk {i}: {e}. Retrying in 2s...")
            time.sleep(2)
            emb = get_embedding(chunk["text"])
            embeddings.append(emb)
        # Print progress every 10 chunks
        if (i + 1) % 10 == 0:
            print(f"Embedded {i + 1}/{len(chunks)} chunks")
        time.sleep(batch_delay)
    return embeddings

What does this code do?

  • embed_chunks() loops through every chunk and calls get_embedding() on each one.
  • The try/except block catches API errors (like throttling from Bedrock) and retries once after a 2-second pause.
  • batch_delay=0.1 adds a small delay between API calls to avoid hitting rate limits in the first place.
  • Save embed_index.py. You'll test this function in the next chunk when you run the full script.
Build and save the FAISS index

With the embedding functions ready, you need one more function to build the actual search index. FAISS stores all the vectors and lets you search for the closest ones to any query vector using Euclidean distance.

  • In embed_index.py, add the build_faiss_index() function and the main execution block below embed_chunks():
def build_faiss_index(embeddings):
    # Convert to numpy float32 array (required by FAISS)
    vectors = np.array(embeddings, dtype="float32")
    # Create an index using L2 (Euclidean) distance
    index = faiss.IndexFlatL2(vectors.shape[1])
    index.add(vectors)
    return index


if __name__ == "__main__":
    # Load the chunks from the ingestion step
    with open(CHUNKS_PATH, "r") as f:
        chunks = json.load(f)

    print(f"Embedding {len(chunks)} chunks...")
    embeddings = embed_chunks(chunks)

    # Build and save the FAISS index
    index = build_faiss_index(embeddings)
    faiss.write_index(index, INDEX_PATH)
    print(f"\nEmbedded {len(chunks)} chunks ({EMBED_DIMENSIONS}-dim vectors)")
    print(f"FAISS index built with {index.ntotal} vectors")
    print(f"Index saved to {INDEX_PATH}")

What does this code do?

  • build_faiss_index() converts all embeddings to a numpy float32 array (the format FAISS requires) and builds an IndexFlatL2 index. L2 means it uses Euclidean distance to measure how similar two vectors are.
  • The __main__ block ties everything together: it loads chunks.json, embeds every chunk, builds the index, and saves it to data/index.faiss.
  • faiss.write_index() persists the index to disk so you don't need to re-embed everything each time you want to search.
  • Save embed_index.py.

Before you run this, think about what you expect to see. How many chunks did your ingestion step create? You should see that same number appear in the embedding output.

  • Run the embedding pipeline by running this command:
python embed_index.py

This takes a few seconds as each chunk is sent to Bedrock one at a time. You should see progress updates every 10 chunks, followed by a summary.

What should I see?

Your output will look something like this (the exact numbers depend on how many chunks your documents produced):

Embedding 24 chunks...
Embedded 10/24 chunks
Embedded 20/24 chunks

Embedded 24 chunks (1024-dim vectors)
FAISS index built with 24 vectors
Index saved to data/index.faiss

Script hanging or erroring partway through?

  • If the script retries repeatedly and then fails, your Bedrock account may be hitting rate limits. Wait 30 seconds and try again.
  • If you see a FileNotFoundError for chunks.json, make sure you ran python ingest.py in Step 2 first.
  • If you see 0 chunks in the output, your docs/ folder may be empty. Add .txt files and re-run python ingest.py before running python embed_index.py again.

Help me debug my embedding pipeline

  • Confirm the index file was created by running:
ls data/

You should see both chunks.json and index.faiss listed in the output.

✔️ Awesome, I've got everything!

Great. Double check you've saved embed_index.py and move on to the retrieval test below.

ⓧ I'd like to double check the full code

Scroll up and confirm your embed_index.py file contains these three sections, in order:

  • Imports and file path constants (INDEX_PATH and CHUNKS_PATH), followed by the get_embedding() function.
  • The embed_chunks() function with try/except retry logic and progress printing.
  • The build_faiss_index() function and the if __name__ == "__main__": block that ties them together.
Test retrieval with a sample query

The index is built, but you haven't proven it actually finds relevant passages. A vector index that returns garbage is worse than no index at all because it gives you false confidence.

  • Test semantic search against your index by running this command:
python -c "
import json, numpy as np, faiss
from embed_index import get_embedding
index = faiss.read_index('data/index.faiss')
chunks = json.load(open('data/chunks.json'))
q = np.array([get_embedding('What are the main topics covered?')], dtype='float32')
D, I = index.search(q, 3)
for i, idx in enumerate(I[0]):
    print(f'Result {i+1} [{chunks[idx][\"source_file\"]}]: \"{chunks[idx][\"text\"][:80]}...\"')
"

What does this test do?

This loads your saved FAISS index from disk, embeds the question "What are the main topics covered?", and uses index.search() to find the 3 closest chunk vectors. It then prints the source filename and a preview of each retrieved chunk.

The results should come from your actual document files and contain text relevant to the question. If the retrieved chunks look like reasonable answers, your semantic search is working.

You should see three results, each showing the source filename in brackets and a preview of the chunk text. The passages should be relevant to the question you asked.

Try different questions

Change the query string in the test command to ask a question specific to your documents. Notice how the results change to match. This is semantic search in action. The index finds passages by meaning, not by matching exact words.

Results don't look relevant?

  • If all three results come from the same file, that's normal for now. Step 5 addresses this exact problem.
  • If the text previews look like random passages, try a question more closely related to what your documents actually discuss.

Help me understand why my FAISS search results don't look right

Your vector index is built and proven to retrieve relevant passages. Next up, you'll wire these retrieved chunks into a full generation pipeline that produces natural language answers with citations tracing back to the exact source files.

Generate Answers with Citations

In the last step you built a vector index that turns natural language into searchable embeddings. You can now find the most relevant chunks for any question.

Finding relevant passages is only half the story. Raw chunks are not answers. You need a generation model to read the retrieved context and synthesize a coherent response. The critical piece: every claim in the answer must trace back to a specific source file and passage. This is what separates a toy chatbot from an enterprise-grade system.

In this step, get ready to:
  • Write retrieval and prompt-building functions that format source citations for the LLM.
  • Call the Bedrock Converse API with Amazon Nova Micro to generate grounded answers.
  • Test the full RAG pipeline with real questions and confirm it refuses to hallucinate.
Write the retrieval and prompt functions

The query script needs to do three things in sequence: load the index, retrieve relevant chunks, and package them into a prompt the LLM can use. You will reuse the get_embedding() function from embed_index.py so the query gets embedded the same way your documents were.

  • Create a new file called rag_query.py in your project folder, then paste the following code:
import sys
import json
import numpy as np
import faiss
from config import bedrock_runtime, GEN_MODEL_ID, TOP_K
from embed_index import get_embedding

# Paths to the saved index and chunk metadata
INDEX_PATH = "data/index.faiss"
CHUNKS_PATH = "data/chunks.json"


def load_index_and_chunks():
    """Load the FAISS index and chunk metadata from disk."""
    index = faiss.read_index(INDEX_PATH)
    with open(CHUNKS_PATH, "r") as f:
        chunks = json.load(f)
    return index, chunks


def retrieve(query, index, chunks, top_k=TOP_K):
    """Embed the query and search FAISS for the closest chunks."""
    query_emb = np.array([get_embedding(query)], dtype="float32")
    distances, indices = index.search(query_emb, top_k)
    results = []
    for i, idx in enumerate(indices[0]):
        if idx < len(chunks):
            results.append({
                "chunk": chunks[idx],
                "distance": float(distances[0][i]),
                "rank": i + 1,
            })
    return results

What does this code do?

This sets up the retrieval layer of your RAG pipeline. load_index_and_chunks() reads the FAISS index and chunk metadata you built in the previous step.

The retrieve() function embeds the user's question with the same Titan V2 model, then searches the index for the closest vectors. It returns the top-k chunks sorted by distance, with each result carrying its chunk metadata for citation tracing.

Now you need the prompt that tells the LLM how to use these retrieved chunks. The prompt is where you enforce citation behavior.

  • In rag_query.py, add the build_prompt() function below the retrieve() function:
def build_prompt(query, retrieved_chunks):
    """Format retrieved chunks into a citation-aware prompt."""
    context_parts = []
    for i, result in enumerate(retrieved_chunks):
        source = result["chunk"]["source_file"]
        chunk_id = result["chunk"]["chunk_id"]
        # Label each chunk with a source number for citation
        context_parts.append(
            f"[Source {i+1}] (File: {source}, Chunk {chunk_id}):\n{result['chunk']['text']}"
        )
    context = "\n\n".join(context_parts)
    prompt = (
        "You are a knowledgeable assistant. Answer the question using ONLY the provided context. "
        "Cite your sources using [Source N] notation. If the context does not contain enough information "
        "to answer the question, say so explicitly.\n\n"
        f"Context:\n{context}\n\n"
        f"Question: {query}\n\n"
        "Answer:"
    )
    return prompt

Why does the prompt matter so much?

The prompt does two critical things. First, it labels each chunk with a [Source N] tag that includes the original filename and chunk ID. This gives the LLM a reference system to cite.

Second, it instructs the model to answer only from the provided context and to say so explicitly when the context is insufficient. This is how you prevent hallucination in a RAG system.

Generate the answer with Bedrock's Converse API

With retrieval and prompt building in place, you need the generation layer. This is where Amazon Nova Micro reads the retrieved context and produces a grounded answer with citations.

  • In rag_query.py, add the following functions below build_prompt():
def generate_answer(prompt):
    """Call Bedrock Converse API with Nova Micro to generate an answer."""
    response = bedrock_runtime.converse(
        modelId=GEN_MODEL_ID,
        messages=[{
            "role": "user",
            "content": [{"text": prompt}],
        }],
        inferenceConfig={"maxTokens": 1024},
    )
    return response["output"]["message"]["content"][0]["text"]


def query_rag(query):
    """Orchestrate the full RAG pipeline: retrieve, prompt, generate."""
    index, chunks = load_index_and_chunks()
    results = retrieve(query, index, chunks)
    prompt = build_prompt(query, results)
    answer = generate_answer(prompt)

    print(f"\nQuestion: {query}")
    print(f"\nAnswer:\n{answer}")
    print(f"\nSources used:")
    for r in results:
        source = r["chunk"]["source_file"]
        preview = r["chunk"]["text"][:80].replace("\n", " ")
        print(f"  [{source}, chunk {r['chunk']['chunk_id']}] \"{preview}...\"")


if __name__ == "__main__":
    query = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else "What are the main topics covered in these documents?"
    query_rag(query)

What does this code do?

The generate_answer() function sends the prompt to Amazon Nova Micro via the Converse API. The maxTokens parameter caps the response length at 1024 tokens.

The query_rag() function ties the whole pipeline together: load the index, retrieve relevant chunks, build the citation-aware prompt, generate the answer, and print everything with source references. The __main__ block lets you pass a question as a command-line argument.

  • Save rag_query.py.

Before you run this, what do you think the output will look like? Will you see just a plain answer, or something more?

  • Test the full pipeline by running:
python rag_query.py "What are the main topics covered in these documents?"

You should see a natural language answer followed by numbered citations. Each citation traces back to a specific file and chunk from your docs/ folder.

Seeing an AccessDeniedException?

This usually means the model is not enabled in your Bedrock console. Return to the AWS Bedrock console, open Model access, and confirm that Amazon Nova Micro shows as enabled in us-east-1.

Also check that your config.py has GEN_MODEL_ID set to amazon.nova-micro-v1:0 and REGION set to us-east-1.

Help me debug an AccessDeniedException when calling Bedrock's Converse API with Amazon Nova Micro.

✔️ Awesome, I've got everything!

Great. Double check you have saved rag_query.py and that your answer included source citations.

ⓧ I'd like to double check the full code

Your rag_query.py file should contain these sections in order. Scroll up and confirm each one is present:

  • Imports and constants (INDEX_PATH, CHUNKS_PATH).
  • The load_index_and_chunks() function.
  • The retrieve() function.
  • The build_prompt() function with the citation-aware system prompt.
  • The generate_answer() function calling the Converse API.
  • The query_rag() function that orchestrates everything and prints citations.
  • The if __name__ == "__main__" block.
Test with real questions and verify hallucination handling

A single test question is not enough. You need to confirm the system handles different types of queries, including ones your documents cannot answer.

  • Try a question about a specific topic covered in one of your files:
python rag_query.py "What does the document say about [[TOPIC_QUERY="a topic from one of your text files"]]?"

The answer should reference content from the file that covers that topic, with citations pointing to the correct source file.

Now for the real test. What happens when you ask something your documents know nothing about?

  • Intentionally ask a question your documents cannot answer:
python rag_query.py "What is the population of Mars?"

The model should respond honestly that the information is not in the provided context. If it fabricates an answer instead, check that your system prompt in build_prompt() includes the instruction to say so explicitly when the context is insufficient.

Why does this matter?

Hallucination handling is the dividing line between a demo and a production system. In enterprise RAG, a wrong answer with confident citations is worse than no answer at all. Your prompt engineering enforces this boundary.

Your RAG tool now retrieves relevant passages, generates grounded answers, and cites its sources. Next up, you will push this system to its limits and see where it breaks.

Hit the Wall and Handle Failure

In the last step, you built a full RAG pipeline that retrieves chunks, generates answers with citations, and even refuses to answer when the documents don't contain the information.

That works well for single-topic questions. But production queries rarely fit neatly into one document. When a question requires information from two different files, pure vector similarity tends to cluster all results from whichever file is closest in embedding space. You'll see this break right now.

In this step, get ready to:
  • Run a cross-document question and observe the retrieval gap.
  • Add a re-ranking function to surface diverse results from multiple files.
  • Verify that the improved retrieval draws from multiple source files.
Ask a cross-document question

Think of a question that needs information from at least two of your documents. For example, if you have one file about cooking and another about nutrition, ask something like "What cooking techniques help preserve the most nutrients?"

Before you run this, what do you think will happen? Will the system pull relevant chunks from both files, or will it cluster on just one?

  • Run a cross-document question against your RAG pipeline by running this command (replace the question with one that spans your own documents):
python rag_query.py "What are the connections between topics in my documents?"

Look at the Sources used: section at the bottom of the output. Every source comes from the same file.

Why does this happen?

FAISS ranks chunks purely by vector distance. Chunks from the same document tend to use similar vocabulary and sentence structure, so their embeddings cluster together in vector space.

When you ask for the top 3, all 3 slots go to the tightest cluster. Relevant chunks from a second file are close but just slightly further away, so they get cut. The answer looks confident but only tells half the story.

This is not a bug. This is a production failure mode that Forward Deployed Engineers encounter constantly. Seeing it break is the first step toward fixing it.

Add a re-ranking function

The fix is a re-ranking step. Instead of taking the top 3 results from FAISS directly, you fetch a larger pool of candidates (top 10) and re-score them using a second signal: keyword overlap with the query. This breaks the single-document cluster by surfacing chunks that share query terms even if they're slightly further in embedding space.

  • In rag_query.py, add the following rerank() function directly below the retrieve() function:
def rerank(query, index, chunks, initial_k=TOP_K_RERANK, final_k=TOP_K):
    candidates = retrieve(query, index, chunks, top_k=initial_k)
    query_words = set(query.lower().split())
    for candidate in candidates:
        chunk_words = set(candidate["chunk"]["text"].lower().split())
        candidate["keyword_overlap"] = len(query_words & chunk_words)
    candidates.sort(key=lambda x: (-x["keyword_overlap"], x["distance"]))
    return candidates[:final_k]

What does this code do?

  • retrieve(query, index, chunks, top_k=initial_k) fetches 10 candidates instead of the usual 3, casting a wider net across the index.
  • query_words & chunk_words calculates how many words in the query also appear in each chunk. This is the re-scoring signal that breaks the single-document cluster.
  • The final sort prioritizes keyword overlap first, then uses FAISS distance as a tiebreaker. The top 3 survivors are the most relevant AND most diverse.

Now you need to wire this function into the main query pipeline.

  • In rag_query.py, find the query_rag() function. Find this line:
def query_rag(query):
    index, chunks = load_index_and_chunks()
    results = retrieve(query, index, chunks)
  • Replace those three lines with this updated version that uses re-ranking by default:
def query_rag(query, use_rerank=True):
    index, chunks = load_index_and_chunks()
    if use_rerank:
        results = rerank(query, index, chunks)
    else:
        results = retrieve(query, index, chunks)

What changed?

The use_rerank=True parameter makes re-ranking the default behavior. The if/else branch lets you toggle between raw retrieval and re-ranked retrieval, which is useful for comparing results.

  • Save rag_query.py.

Seeing a NameError for rerank?

  • Make sure the rerank() function is defined above query_rag() in the file. Python reads top-to-bottom, so the function must exist before it is called.
  • Check that TOP_K_RERANK is imported from config at the top of the file.

Why is my rerank function not being found?

Re-run the cross-document question and compare

Before you run this, predict: will the re-ranking change which files appear in the sources? How many different filenames do you expect to see?

  • Run the same cross-document question you used earlier:
python rag_query.py "What are the connections between topics in my documents?"

This time the Sources used: section should show chunks from multiple files instead of clustering on just one.

What just changed?

Without re-ranking, FAISS returned the 3 closest vectors. All 3 clustered on the same file because its chunks share similar language patterns.

With re-ranking, the system fetches 10 candidates and promotes chunks that share keywords with your query. This breaks the cluster and surfaces relevant chunks from different source files.

Why this matters in FDE interviews

You just diagnosed a real production failure, explained why it happens, and shipped a fix. This is the exact sequence interviewers look for: "tell me about a time something broke and how you fixed it." You now have a concrete, technical answer.

✔️ Awesome, I've got everything!

Great. Make sure you've saved rag_query.py with both the rerank() function and the updated query_rag().

ⓧ I'd like to double check the full code

Your rag_query.py is longer than 30 lines, so scroll through it and confirm each section is present, in this order:

  • Imports at the top: sys, json, numpy, faiss, plus GEN_MODEL_ID, TOP_K, TOP_K_RERANK from config and get_embedding from embed_index.
  • load_index_and_chunks() function that reads the FAISS index and chunks JSON.
  • retrieve() function that embeds a query and searches FAISS.
  • rerank() function directly below retrieve() that re-scores by keyword overlap.
  • build_prompt() and generate_answer() functions from the previous step.
  • query_rag(query, use_rerank=True) with the if/else branch that defaults to re-ranking.
  • The __main__ block that reads the query from command-line arguments.

Your RAG pipeline now handles cross-document questions. Next up, you'll take on the Secret Mission: building an eval framework that proves your system works reliably.

Secret mission

Build a Retrieval Eval Framework

Your RAG tool works, but how do you prove it? FDE interviewers almost always ask "how do you know this works?" Build an eval framework that systematically tests retrieval quality with precision metrics and failure detection.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with no persistent cloud resources. AWS Bedrock charges only when you make API calls, so there are no ongoing costs once you stop running your scripts.

Decide whether to keep your project files, or delete them entirely.

Resources you used:

  • Project folder on your Desktop containing config.py, ingest.py, embed_index.py, rag_query.py, eval_rag.py, the docs/ folder, and the data/ folder (containing chunks.json and index.faiss).
  • Python packages installed via pip: boto3, faiss-cpu, and numpy.
  • AWS Bedrock API usage (pay-per-token, no persistent resources created).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your RAG tool or want to try it on new documents.

  • All local files persist on your machine at no cost.
  • AWS Bedrock only charges when you actively make API calls by running your scripts. No charges accumulate while the project sits idle.
  • To use the tool on new documents, drop new .txt files into your docs/ folder and re-run ingest.py and embed_index.py.

Pause - I'll come back to this later

There are no persistent AWS resources to pause. Your costs are already zero as long as you are not running the Python scripts.

  • Keep your project folder on your Desktop for when you want to return.
  • No action is needed on the AWS side. Bedrock model access stays enabled at no cost.

Delete - I don't want to use this again

Remove all project files from your machine. No AWS resources need to be cleaned up beyond the API calls you already made.

  • Delete your project folder from the Desktop by running this command (macOS/Linux):
cd ~/Desktop
ls

Find your project folder name in the output, then remove it:

rm -rf your-project-folder-name

On Windows, use this instead:

cd ~\Desktop
rmdir /s /q your-project-folder-name
  • Optionally, uninstall the Python packages you installed for this project:
pip uninstall boto3 faiss-cpu numpy -y

What about AWS Bedrock?

There are no persistent AWS resources to delete. Bedrock is purely pay-per-API-call. The model access you enabled in the Bedrock console costs nothing to leave on.

Nice Work!

Nice Work!

You just built a general-purpose RAG tool from scratch using AWS Bedrock and Python. Drop any documents into a folder, ask a question, and get a grounded answer with citations tracing every claim back to its source file.

You've learned how to:

  • Built a complete RAG pipeline from scratch: document ingestion from any folder, overlapping text chunking, vector embedding with Titan Text Embeddings V2, semantic search with FAISS, and LLM-powered answer generation with source-file citations using Amazon Nova Micro.
  • Called enterprise-grade AI models through AWS Bedrock: used boto3 to invoke embedding and generation models via the Bedrock runtime, handling rate limits, retries, and response parsing in production-style Python.
  • Handled a real production failure: discovered that basic vector search clusters on a single document for cross-document questions, then fixed it with a re-ranking step that surfaces diverse, relevant chunks from multiple source files.
  • Secret Mission: built a retrieval eval framework that measures Precision@3 using keyword-based gold labels, flags weak questions, and produces a summary report. This proves you think about reliability, not just capability.

Ready to quiz yourself?