Build a RAG API with AWS Bedrock

Build a FastAPI RAG pipeline that answers questions from your own documents.

Introduction

30 Second Summary

When you ask a chatbot a question and it confidently makes something up, that is what happens when an LLM has no grounding in real data. Production AI assistants solve this by retrieving relevant information first, then generating answers based on what they actually found.

In this project, you will build a REST API that implements this Retrieval-Augmented Generation pattern from scratch. You will upload documents, split them into searchable chunks, and query them with natural language questions answered by Claude on AWS Bedrock.

What You'll Build

A FastAPI application that ingests text documents, generates vector embeddings, stores them in a FAISS index, and answers questions using a retrieval-augmented generation pipeline powered by LangChain and AWS Bedrock.

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

  • A working FastAPI application with /ingest and /query endpoints that form a complete RAG pipeline.
  • A document chunking and vector embedding pipeline using LangChain and Amazon Titan Embeddings.
  • A FAISS vector store for fast similarity search over your document collection.
  • Secret Mission: Add a /evaluate endpoint that scores RAG response quality using context relevance metrics.

Are there any prerequisites?

You need an AWS account with billing enabled, Python 3.10 installed, and VS Code installed. AWS Bedrock does not have a free tier, so this project costs approximately $0.50 to $1.00 in API usage.

Set Up Your Environment

Before you can build a RAG pipeline, you need the right tools talking to each other.

Your API will call AWS Bedrock for both text generation and embeddings. In this step, you will install the AWS CLI, enable Bedrock model access, set up a Python virtual environment, and scaffold your project folder.

In this step, get ready to:
  • Install and configure the AWS CLI, then verify Bedrock model access.
  • Create a Python virtual environment and install project dependencies.
  • Scaffold the project directory and verify a minimal FastAPI server.
Install and configure the AWS CLI

The AWS CLI lets you interact with AWS services from your terminal. Your RAG application uses Bedrock under the hood, and the CLI's credential chain is how your Python code authenticates.

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

✔️ I see a version number

Your AWS CLI is ready. Move on to configuring your credentials below.

ⓧ Command not found

You need to install the AWS CLI. Follow the instructions for your operating system.

macOS

  • Download and install the AWS CLI by running these commands:
curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg"
sudo installer -pkg AWSCLIV2.pkg -target /
  • Your Mac will ask for your password to authorize the install. Type it and press Enter.
  • Verify the install succeeded by running:
aws --version

You should see output starting with aws-cli/2 followed by version details.

Windows

  • Download the MSI installer from this URL:
https://awscli.amazonaws.com/AWSCLIV2.msi
  • Run the downloaded AWSCLIV2.msi file and follow the on-screen prompts.
  • After installation finishes, open a new Command Prompt and verify by running:
aws --version

You should see output starting with aws-cli/2 followed by version details.

  • Configure your AWS credentials by running:
aws configure
  • When prompted, enter your AWS Access Key ID and press Enter.
  • Enter your AWS Secret Access Key and press Enter.
  • For Default region name, type us-east-1 and press Enter.
  • For Default output format, type json and press Enter.

Where do I find my AWS access keys?

In the AWS Management Console, navigate to IAM, select your user, click the Security credentials tab, and create an access key. Copy both the Access Key ID and Secret Access Key before closing the dialog.

  • Verify your credentials are working by running:
aws sts get-caller-identity

You should see a JSON response containing your Account ID, User ARN, and User ID.

Seeing an access denied or invalid credentials error?

  • Double-check that you copied the Access Key ID and Secret Access Key without extra spaces.
  • Make sure your IAM user has the required permissions: bedrock:InvokeModel and bedrock:InvokeModelWithResponseStream.
  • Run aws configure again and re-enter your credentials carefully.

Help me troubleshoot my AWS CLI credentials

Enable Bedrock model access for Anthropic Claude and Titan Embeddings

AWS Bedrock models are accessible by default once you have the right IAM permissions. Anthropic models require a one-time use-case form submission before your first API call.

  • Open the AWS Management Console and search for Bedrock.
  • Make sure you are in the us-east-1 (N. Virginia) region. Check the region selector in the top-right corner.
  • In the left navigation pane, under Bedrock configurations, choose Model access.
  • Click Modify model access.
  • Select the checkboxes for Anthropic (which includes Claude Sonnet 4) and Amazon (which includes Titan Text Embeddings V2).
  • Click Next.
  • If this is your first time using Anthropic models, click Submit use case details when prompted.
  • Fill out the use-case form with a description of your project and a URL (your GitHub profile or portfolio works).
  • Click Submit form.
  • Review the terms and click Submit.

After a few moments, the Access status for both model groups updates to Access granted.

Why is the Anthropic form required?

Anthropic requires a one-time use-case submission per AWS account (or per Organization). This is not an approval process that takes days. Access is granted immediately after you submit the form. You only need to do this once across all regions.

Already submitted the form before?

If you have used Anthropic models on Bedrock previously, you will not see the use-case form prompt. Your access is already active.

Create a virtual environment and install dependencies

A virtual environment isolates your project's Python packages from other projects on your machine. This prevents version conflicts.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rag-api
cd rag-api
  • Create a Python virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Create the requirements.txt file that lists all project dependencies. Open VS Code in the rag-api folder by running:
code .
  • In VS Code, create a new file called requirements.txt in the root of your rag-api folder.
  • Paste the following contents into requirements.txt:
fastapi==0.137.2
uvicorn==0.34.0
langchain-aws==1.6.0
langchain==1.3.10
faiss-cpu==1.14.2
python-multipart==0.0.20

What are these packages?

  • fastapi is the web framework for your API endpoints.
  • uvicorn is the server that runs your FastAPI application.
  • langchain-aws provides ChatBedrock and BedrockEmbeddings wrappers for AWS Bedrock.
  • langchain orchestrates document splitting, retrieval chains, and vector store integration.
  • faiss-cpu is a local vector similarity search library (no GPU required).
  • python-multipart enables file upload handling in FastAPI.
  • Save requirements.txt, then install the dependencies by running this in your terminal:
pip install -r requirements.txt

You should see each package download and install. The process may take a minute or two.

Seeing installation errors?

  • Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
  • Confirm you are running Python 3.10 or higher with python3 --version. All packages in this project require Python 3.10+.

Help me fix my pip install errors

Create the project structure and verify with a FastAPI server

Your project needs a specific folder structure to keep ingestion, querying, and routing logic separate.

  • Create the project directories by running these commands:
mkdir -p app/routes app/services documents tests
touch app/__init__.py app/routes/__init__.py app/services/__init__.py
  • Confirm the folder structure exists by running:
ls app app/routes app/services

You should see __init__.py listed under app/, app/routes/, and app/services/.

  • Create the main application file. In VS Code, create a new file at app/main.py and paste the following code:
from fastapi import FastAPI

app = FastAPI(title="RAG API", version="1.0.0")


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

What does this code do?

  • FastAPI(title="RAG API", version="1.0.0") creates your application instance with a name and version that appear in the auto-generated API docs.
  • The /health endpoint returns a simple JSON response. This is useful for confirming your server is running.
  • Save app/main.py.
  • Start the FastAPI development server by running:
uvicorn app.main:app --reload
  • Open your browser and navigate to http://localhost:8000/health.

You should see {"status":"healthy"} in your browser. Your FastAPI server is running.

Server not starting or port already in use?

  • Make sure you are running the command from the rag-api folder (not from inside app/).
  • If port 8000 is in use, stop any other running servers first.
  • Check that uvicorn was installed correctly by running pip show uvicorn.

Help me get my FastAPI server running

  • Stop the server by pressing Ctrl+C in your terminal.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files.

ⓧ I'd like to double check the full code

Here is what each file should look like at this point.

requirements.txt:

fastapi==0.137.2
uvicorn==0.34.0
langchain-aws==1.6.0
langchain==1.3.10
faiss-cpu==1.14.2
python-multipart==0.0.20

app/main.py:

from fastapi import FastAPI

app = FastAPI(title="RAG API", version="1.0.0")


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

app/routes/__init__.py and app/services/__init__.py should both be empty files.

Your environment is locked in. You have AWS credentials configured, Bedrock model access enabled, and a running FastAPI server. Next up, you will build the document ingestion pipeline that chunks and embeds your text files.

Build the Document Ingestion Pipeline

Your environment is ready and your FastAPI server is responding. Now it is time to build the core engine of your RAG application: the document ingestion pipeline.

Before an LLM can answer questions about your documents, those documents need to be broken into smaller pieces, converted into numerical representations (embeddings), and stored in a searchable index. This step builds that entire pipeline from scratch.

In this step, get ready to:
  • Create a document chunking module that splits text into semantically meaningful segments.
  • Build a vector store service that generates embeddings and persists them to a FAISS index.
  • Wire a FastAPI endpoint that accepts uploaded text files and pushes them through the pipeline.
Create the document chunking module

A full document is too large to send to an LLM or compare meaningfully against a question. The solution is to split it into smaller chunks that each capture one idea or paragraph.

What is document chunking?

Chunking breaks a long document into overlapping segments. The overlap ensures that important sentences sitting at a boundary are not lost when a chunk is cut.

LangChain's RecursiveCharacterTextSplitter tries natural breakpoints first (double newlines, then single newlines, then sentences) before falling back to splitting mid-word.

  • Create a new file called app/services/chunking.py and add the following code:
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document


def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
    # Create a splitter that tries natural breakpoints first
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    # Split the text and attach source metadata to each chunk
    chunks = splitter.create_documents(
        texts=[text],
        metadatas=[{"source": source}],
    )
    return chunks

What does this code do?

  • chunk_document() takes a raw text string and a source filename, then splits the text into overlapping pieces of 512 characters each.
  • The separators list tells the splitter to try paragraph breaks first, then line breaks, then sentence boundaries. This keeps chunks semantically coherent.
  • Each chunk carries metadata recording which file it came from. This lets you cite sources later when answering questions.
  • Save app/services/chunking.py.

You will not see a visible change yet. The next module uses this chunking function to build the full ingestion path.

Seeing an import error?

  • Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
  • Confirm langchain is installed by running pip list | grep langchain. You should see langchain and langchain-aws in the output.

Why is my chunking module not importing correctly?

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/services/chunking.py before moving on.

ⓧ I'd like to double check the full code

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document


def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = splitter.create_documents(
        texts=[text],
        metadatas=[{"source": source}],
    )
    return chunks
Build the vector store service

Chunks of text are not searchable on their own. To find which chunks are relevant to a question, you need to convert them into vector embeddings and store those vectors in a searchable index.

What are vector embeddings?

An embedding is a list of numbers that captures the meaning of a piece of text. Two chunks about similar topics will have embeddings that are close together in vector space.

Amazon Titan Text Embeddings V2 generates these vectors. FAISS (Facebook AI Similarity Search) stores them locally and performs fast nearest-neighbor lookups when you query.

  • Create a new file called app/services/vector_store.py and add the following code:
import os
from langchain_aws import BedrockEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.schema import Document

# Path where the FAISS index is saved to disk
INDEX_PATH = "faiss_index"

# Initialize the embeddings model (runs in AWS Bedrock)
embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1",
)


def get_vector_store() -> FAISS:
    """Load existing FAISS index from disk, or return None if it doesn't exist."""
    if os.path.exists(INDEX_PATH):
        return FAISS.load_local(
            INDEX_PATH, embeddings, allow_dangerous_deserialization=True
        )
    return None


def add_documents_to_store(documents: list[Document]) -> int:
    """Embed documents and add them to the FAISS index."""
    store = get_vector_store()

    if store is None:
        # First time: create a brand new index from the documents
        store = FAISS.from_documents(documents, embeddings)
    else:
        # Index already exists: add new documents to it
        store.add_documents(documents)

    store.save_local(INDEX_PATH)
    return len(documents)


def similarity_search(query: str, k: int = 4) -> list[Document]:
    """Find the k most relevant chunks for a given query."""
    store = get_vector_store()
    if store is None:
        return []
    return store.similarity_search(query, k=k)

What does this code do?

  • BedrockEmbeddings connects to AWS Bedrock and uses the Titan Text Embeddings V2 model to convert text into 1,024-dimensional vectors.
  • get_vector_store() loads a previously saved FAISS index from the faiss_index/ directory, or returns None if no index exists yet.
  • add_documents_to_store() either creates a new index (first upload) or appends to the existing one. It saves the index to disk after every write.
  • similarity_search() embeds a query and returns the closest matching chunks. You will use this in Step 3 when building the query pipeline.
  • Save app/services/vector_store.py.

You will not see a visible change yet. The next substep wires these services to a FastAPI endpoint you can call.

Worried about AWS credentials?

The BedrockEmbeddings client does not connect to AWS until you actually call the endpoint. If your credentials are misconfigured, the error surfaces when you test the /ingest endpoint later in this step.

Help me verify my AWS credentials are configured correctly for Bedrock.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/services/vector_store.py before moving on.

ⓧ I'd like to double check the full code

import os
from langchain_aws import BedrockEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.schema import Document

INDEX_PATH = "faiss_index"

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1",
)


def get_vector_store() -> FAISS:
    if os.path.exists(INDEX_PATH):
        return FAISS.load_local(
            INDEX_PATH, embeddings, allow_dangerous_deserialization=True
        )
    return None


def add_documents_to_store(documents: list[Document]) -> int:
    store = get_vector_store()

    if store is None:
        store = FAISS.from_documents(documents, embeddings)
    else:
        store.add_documents(documents)

    store.save_local(INDEX_PATH)
    return len(documents)


def similarity_search(query: str, k: int = 4) -> list[Document]:
    store = get_vector_store()
    if store is None:
        return []
    return store.similarity_search(query, k=k)
Wire the /ingest endpoint

The chunking and vector store modules are ready. Now you need a FastAPI route that accepts an uploaded file, passes it through the pipeline, and returns a confirmation.

  • Create a new file called app/routes/ingest.py and add the following code:
from fastapi import APIRouter, UploadFile, File
from app.services.vector_store import add_documents_to_store
from app.services.chunking import chunk_document

router = APIRouter()


@router.post("/ingest")
async def ingest_document(file: UploadFile = File(...)):
    # Read the uploaded file content as bytes, then decode to string
    content = await file.read()
    text = content.decode("utf-8")

    # Split the text into chunks and store embeddings in FAISS
    chunks = chunk_document(text, source=file.filename)
    num_chunks = add_documents_to_store(chunks)

    return {
        "filename": file.filename,
        "chunks_created": num_chunks,
        "status": "ingested",
    }

What does this code do?

  • The @router.post("/ingest") decorator registers a POST endpoint at /ingest.
  • File(...) tells FastAPI this endpoint expects a multipart file upload. The file bytes are read and decoded to a UTF-8 string.
  • The function calls your chunk_document() function to split the text, then add_documents_to_store() to embed and index the chunks.
  • The response returns the filename, how many chunks were created, and a status confirmation.
  • Save app/routes/ingest.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/routes/ingest.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, UploadFile, File
from app.services.vector_store import add_documents_to_store
from app.services.chunking import chunk_document

router = APIRouter()


@router.post("/ingest")
async def ingest_document(file: UploadFile = File(...)):
    content = await file.read()
    text = content.decode("utf-8")

    chunks = chunk_document(text, source=file.filename)
    num_chunks = add_documents_to_store(chunks)

    return {
        "filename": file.filename,
        "chunks_created": num_chunks,
        "status": "ingested",
    }

The route exists, but your FastAPI app does not know about it yet. You need to register the router in app/main.py.

  • Open app/main.py and replace the entire contents with the following updated code:
from fastapi import FastAPI
from app.routes import ingest

app = FastAPI(title="RAG API", version="1.0.0")

# Register the /ingest route
app.include_router(ingest.router)


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

What changed?

  • from app.routes import ingest imports the router module you just created.
  • app.include_router(ingest.router) attaches the /ingest POST endpoint to your running app.
  • Save app/main.py.

✔️ Awesome, I've got everything!

Great. Make sure app/main.py is saved with both the import and the include_router line.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from app.routes import ingest

app = FastAPI(title="RAG API", version="1.0.0")

app.include_router(ingest.router)


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

Time to test the full pipeline end-to-end.

  • Create a sample text file in your documents/ folder by running this command:
echo "Retrieval-Augmented Generation (RAG) is a technique that combines information retrieval with text generation. RAG systems first search a knowledge base for relevant documents, then pass those documents as context to a large language model. This approach reduces hallucinations because the model answers based on real data rather than relying solely on its training. RAG is widely used in enterprise applications like customer support bots, internal knowledge assistants, and legal document analysis." > documents/sample.txt
  • Start your FastAPI server by running:
uvicorn app.main:app --reload
  • Open a second terminal window. Press Cmd+Space (macOS) or the Windows key (Windows), type Terminal, and press Enter.
  • Upload your sample document to the /ingest endpoint by running:
curl -X POST "http://localhost:8000/ingest" -F "file=@documents/sample.txt"
  • You should see a JSON response like this:
{
  "filename": "sample.txt",
  "chunks_created": 1,
  "status": "ingested"
}
  • Verify the FAISS index was created on disk by running:
ls faiss_index/

You should see files inside the faiss_index/ directory. This confirms that your document was chunked, embedded via AWS Bedrock, and stored in the FAISS index.

Seeing a credentials or access error?

  • Run aws sts get-caller-identity to confirm your AWS CLI credentials are still valid.
  • Make sure your credentials have the bedrock:InvokeModel permission and the aws-marketplace:Subscribe permission.
  • Confirm you completed the one-time Anthropic use-case form for your AWS account. Titan Embeddings also requires this marketplace acceptance step.
  • Check that your configured region is us-east-1. The model IDs in your code target this specific region.

Help me debug my AWS Bedrock credentials error when calling the /ingest endpoint.

Getting a ModuleNotFoundError?

  • Make sure you are running uvicorn from the project root directory (the folder containing app/).
  • Confirm your virtual environment is activated in the terminal where you run uvicorn.
  • Double-check that both app/routes/__init__.py and app/services/__init__.py exist (they can be empty files).

Help me fix a ModuleNotFoundError in my FastAPI project.

Your ingestion pipeline is live. Documents go in, chunks come out, and embeddings land in a searchable index. Next up, you will build the query side: retrieving relevant chunks and feeding them to Claude to generate grounded answers.

Build the Query and Generation Pipeline

Your ingestion pipeline stores document chunks as vectors. Now you need the other half of RAG: retrieving those chunks and feeding them to an LLM so it can generate a grounded answer.

In this step, you will build the generation side of the pipeline. You will connect to Claude via AWS Bedrock, write a prompt template that keeps responses factual, and expose the full RAG flow through a /query endpoint.

In this step, get ready to:
  • Create the RAG chain service that retrieves context and calls Claude.
  • Write a system prompt that grounds Claude's answers in retrieved documents.
  • Wire the chain to a POST /query endpoint and test the full flow.
Create the RAG chain service

The retrieval logic (similarity_search) already exists in vector_store.py. You need a new module that calls that function, formats the results into a prompt, and sends it to Claude for generation.

  • Create a new file app/services/rag_chain.py and add the following code to initialize the LLM connection:
from langchain_aws import ChatBedrock
from app.services.vector_store import similarity_search

# Initialize Claude via AWS Bedrock for answer generation
llm = ChatBedrock(
    model_id="anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
    model_kwargs={"temperature": 0.1, "max_tokens": 1024},
)

What does this code do?

ChatBedrock is LangChain's wrapper around the AWS Bedrock inference API. It handles authentication, request formatting, and response parsing for you.

The temperature of 0.1 keeps Claude's responses deterministic and factual. A higher temperature would make answers more creative but less reliable for a knowledge assistant.

  • Save app/services/rag_chain.py.

You won't see a change yet. The next chunk adds the prompt template that this LLM will use.

Write the system prompt template

The prompt template is what makes RAG work. It instructs Claude to answer using only the retrieved context, cite its sources, and admit when it does not know the answer.

  • In app/services/rag_chain.py, add the following prompt template below the llm initialization:
SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.

Context:
{context}
"""

Why is prompt engineering important here?

Without explicit grounding instructions, Claude would happily answer from its own training data. That defeats the purpose of RAG. Rule 1 constrains it to retrieved context only.

Rule 2 prevents hallucination on questions the documents do not cover. The {context} placeholder is where retrieved chunks get injected at query time.

  • Save app/services/rag_chain.py.

You won't see a change yet. The next chunk adds the function that ties retrieval and generation together.

  • Still in app/services/rag_chain.py, add the answer_question() function below SYSTEM_PROMPT:
def answer_question(question: str, top_k: int = 4) -> dict:
    # Retrieve the most relevant chunks from the vector store
    relevant_docs = similarity_search(question, k=top_k)

    # Handle case where no documents have been ingested
    if not relevant_docs:
        return {
            "answer": "No documents have been ingested yet. Please upload documents first.",
            "sources": [],
        }

    # Build a context string from retrieved chunks with source labels
    context = "\n\n---\n\n".join(
        [
            f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
            for doc in relevant_docs
        ]
    )

    # Format messages with the system prompt and user question
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
        {"role": "user", "content": question},
    ]

    # Call Claude via Bedrock
    response = llm.invoke(messages)

    # Extract unique source filenames from retrieved documents
    sources = list(set(doc.metadata.get("source", "unknown") for doc in relevant_docs))

    return {"answer": response.content, "sources": sources}

What does this function do?

This is the core RAG orchestration. It follows three steps: retrieve relevant chunks, inject them into the prompt, and call Claude for generation.

The context string labels each chunk with its source filename so Claude can cite where it found the answer. The function returns both the answer text and the list of source documents.

  • Save app/services/rag_chain.py.

Your RAG chain service is complete. It retrieves, builds context, and generates answers.

Seeing an import error?

Make sure your virtual environment is activated and that langchain-aws is installed. Run pip list | grep langchain-aws to confirm.

help me debug a langchain-aws import error in my rag_chain.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved app/services/rag_chain.py.

ⓧ I'd like to double check the full code

from langchain_aws import ChatBedrock
from app.services.vector_store import similarity_search

llm = ChatBedrock(
    model_id="anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
    model_kwargs={"temperature": 0.1, "max_tokens": 1024},
)

SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.

Context:
{context}
"""


def answer_question(question: str, top_k: int = 4) -> dict:
    relevant_docs = similarity_search(question, k=top_k)

    if not relevant_docs:
        return {
            "answer": "No documents have been ingested yet. Please upload documents first.",
            "sources": [],
        }

    context = "\n\n---\n\n".join(
        [
            f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
            for doc in relevant_docs
        ]
    )

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
        {"role": "user", "content": question},
    ]

    response = llm.invoke(messages)

    sources = list(set(doc.metadata.get("source", "unknown") for doc in relevant_docs))

    return {"answer": response.content, "sources": sources}
Wire the RAG chain to the /query endpoint

The RAG logic is ready. Now you need an API endpoint that accepts a question from a user and returns the generated answer with source references.

  • Create a new file app/routes/query.py and add the following code:
from fastapi import APIRouter
from pydantic import BaseModel
from app.services.rag_chain import answer_question

router = APIRouter()


# Define the shape of incoming query requests
class QueryRequest(BaseModel):
    question: str
    top_k: int = 4


# Define the shape of query responses
class QueryResponse(BaseModel):
    answer: str
    sources: list[str]


@router.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    result = answer_question(request.question, request.top_k)
    return result

What does this code do?

QueryRequest defines what the caller sends: a question string and an optional top_k value controlling how many chunks to retrieve (defaults to 4).

QueryResponse defines what comes back: the generated answer and a list of sources (filenames) that contributed to the answer. FastAPI validates both shapes automatically.

  • Save app/routes/query.py.

You won't see a change yet. You still need to register this route in the main app.

  • In app/main.py, find this line at the top of the file:
from app.routes import ingest
  • Replace it with this line to also import the query router:
from app.routes import ingest, query
  • Find this line:
app.include_router(ingest.router)
  • Add the following line directly below it to register the query router:
app.include_router(query.router)
  • Save app/main.py.

Seeing a ModuleNotFoundError?

Check that app/routes/__init__.py exists (even if empty). Python needs it to treat the folder as a package.

help me fix a ModuleNotFoundError when importing my query router

✔️ Awesome, I've got everything!

Great. Double check you have saved both app/routes/query.py and app/main.py.

ⓧ I'd like to double check the full code

from fastapi import APIRouter
from pydantic import BaseModel
from app.services.rag_chain import answer_question

router = APIRouter()


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


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


@router.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    result = answer_question(request.question, request.top_k)
    return result
from fastapi import FastAPI
from app.routes import ingest, query

app = FastAPI(title="RAG API", version="1.0.0")

app.include_router(ingest.router)
app.include_router(query.router)


@app.get("/health")
async def health_check():
    return {"status": "healthy"}
Test the end-to-end RAG flow

Your full pipeline is wired up. Time to prove it works by sending a question about your ingested document and checking that Claude answers from the document content.

  • Start the server by running this command in your terminal:
uvicorn app.main:app --reload
  • Open a second terminal window and send a question about your ingested document by running:
curl -X POST http://127.0.0.1:8000/query -H "Content-Type: application/json" -d '{"question": "What is this document about?"}'

What should I see?

You should receive a JSON response with an answer field containing text drawn from your ingested document. The sources field should list the filename you uploaded earlier.

The response may take a few seconds because Bedrock is calling Claude behind the scenes.

  • Now test the grounding behavior by asking a question your document cannot answer:
curl -X POST http://127.0.0.1:8000/query -H "Content-Type: application/json" -d '{"question": "What is the population of Mars?"}'

Claude should respond with something like "I don't know based on the available documents." This proves your prompt template is working.

Getting a connection error or 500 response?

Make sure the server is still running in your first terminal window. Check the server logs for errors. The most common issue is incorrect AWS credentials or missing Bedrock model access.

Run aws sts get-caller-identity in your second terminal to confirm your AWS credentials are valid.

help me debug a 500 error from my /query endpoint

Your RAG pipeline is fully operational. You can ingest documents and ask questions that get answered with grounded, cited responses. Next up, you will tune your prompts and chunking strategy to squeeze better quality out of the same pipeline.

Tune Your Prompts and Chunking

Your RAG pipeline works end-to-end. You can ingest documents and query them with natural language questions.

But the quality of a RAG system depends heavily on two tunable factors: how you split documents into chunks, and how you instruct the LLM to use retrieved context. In this step, you will experiment with different chunking parameters and prompt variations to understand their impact on response quality.

In this step, get ready to:
  • Test different chunk sizes (256, 512, and 1024) and compare retrieval quality.
  • Refine your system prompt with improved instructions for format, tone, and edge cases.
  • Document at least two prompt iterations that measurably improve response quality.
Experiment with different chunk sizes

Chunk size directly controls the trade-off between precision and context in your retrieval results. Smaller chunks produce more focused matches but risk losing surrounding context. Larger chunks preserve context but may dilute the relevance signal.

Why does chunk size matter for RAG?

Each chunk becomes a single vector in your FAISS index. When you query, the retriever returns the top-k most similar vectors. If a chunk covers three unrelated topics, its vector is vaguely similar to all three but strongly similar to none.

Too small (under 100 characters) and each chunk is a sentence fragment that could mean anything. Too large (over 2000 characters) and you dilute the signal across topics.

Your current chunk_document() function in app/services/chunking.py already accepts chunk_size and chunk_overlap as parameters. You will test three different configurations by modifying the default values, re-ingesting your document, and comparing the query results.

  • Stop your running server by pressing Ctrl+C in the terminal.
  • Delete the existing FAISS index so you start fresh by running this command:
rm -rf faiss_index
  • In app/services/chunking.py, find the function signature with the default chunk size:
def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
  • Change chunk_size from 512 to 256 and chunk_overlap from 50 to 25, so the line now looks like this:
def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 256,
    chunk_overlap: int = 25,
) -> list[Document]:
  • Save app/services/chunking.py.
  • Restart your server by running:
uvicorn app.main:app --reload
  • Re-ingest the same document you used earlier by running:
curl -X POST http://localhost:8000/ingest -F "file=@documents/sample.txt"

Notice how the chunks_created count in the response is higher than before. Smaller chunks produce more pieces from the same document.

  • Query your API with a specific question about your document by running:
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the main topic of this document?"}'

Take note of the answer quality. With 256-character chunks, answers tend to be more focused but may miss broader context that spans multiple paragraphs.

  • Now test the opposite extreme. Stop the server with Ctrl+C, delete the index, and change the defaults to 1024 and 100:
rm -rf faiss_index
  • In app/services/chunking.py, update the function signature so it looks like this:
def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 1024,
    chunk_overlap: int = 100,
) -> list[Document]:
  • Save the file, restart the server, re-ingest, and query again with the same question.
  • Compare the response to what you got with 256. Larger chunks preserve more context per retrieval result, which can help with questions requiring broader understanding.

What should I observe?

  • 256-character chunks: More chunks created, focused answers, may miss cross-paragraph context.
  • 512-character chunks (your original default): A balanced middle ground for most text documents.
  • 1024-character chunks: Fewer chunks, richer context per retrieval, but may include irrelevant content alongside relevant content.
  • After testing, set chunk_size back to 512 and chunk_overlap back to 50 in app/services/chunking.py. This provides the best balance for general-purpose text documents.
def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
  • Save app/services/chunking.py.
  • Delete the index one final time and re-ingest with the optimal settings:
rm -rf faiss_index
uvicorn app.main:app --reload
curl -X POST http://localhost:8000/ingest -F "file=@documents/sample.txt"

Getting different chunk counts than expected?

The chunk_size parameter in RecursiveCharacterTextSplitter is measured in characters, not tokens. A 512-character chunk is roughly 128 tokens for English text. Make sure you saved the file before restarting the server.

Also confirm you deleted the faiss_index folder before each re-ingest. Old vectors from previous runs will still be in the index otherwise.

help me debug my chunking experiment results

Refine the system prompt

Chunking affects what context gets retrieved. The system prompt affects how the LLM uses that context. A well-engineered prompt reduces hallucination and improves citation quality.

Your current SYSTEM_PROMPT in app/services/rag_chain.py already has four core rules. You will now experiment with variations to see how they affect response quality.

  • In app/services/rag_chain.py, find the SYSTEM_PROMPT variable. Here is what it currently looks like:
SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.

Context:
{context}
"""

What makes a good RAG prompt?

A RAG system prompt serves as a contract between you and the LLM. Each rule constrains the model's behavior. Without rule 1, the model freely uses training data and hallucinates. Without rule 2, it invents answers for questions not covered by your documents.

Prompt engineering for RAG is about finding the set of instructions that maximizes grounding (answers come from context) while minimizing refusals on questions the context actually covers.

Try the following experiment. You will test two prompt iterations and document their effect on response quality.

  • For your first iteration, temporarily remove rules 2 and 3 from the prompt. Replace the entire SYSTEM_PROMPT with this stripped-down version:
SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using the information in the context below.
2. Keep answers concise.

Context:
{context}
"""
  • Save app/services/rag_chain.py and let the server reload.
  • Test with a question your document can answer:
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the main topic of this document?"}'
  • Now test with a question your document does NOT answer:
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the population of Mars?"}'

Without the grounding rules, notice how the model may answer confidently even when the context has no relevant information. It may also skip citing sources.

  • For your second iteration, restore the full prompt and add additional instructions for handling ambiguous and multi-hop questions. Replace SYSTEM_PROMPT with this expanded version:
SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.
5. If the question requires combining information from multiple chunks, synthesize the answer and cite all relevant sources.
6. For ambiguous questions, state the ambiguity and provide the most likely interpretation based on available context.
7. Use a professional, neutral tone.

Context:
{context}
"""
  • Save the file and test with the same two questions plus a multi-hop question that requires combining information from different parts of your document.
  • Compare the three prompt versions. Document your observations: which version produces the fewest hallucinations, the best citations, and the most complete answers.

What should I compare across iterations?

  • Grounding: Does the model only use information from the context, or does it add its own knowledge?
  • Refusal quality: When the context lacks an answer, does the model say so clearly?
  • Citation accuracy: Does the model name the source document correctly?
  • Edge cases: How does it handle ambiguous or multi-hop questions?
  • After experimenting, restore the original four-rule SYSTEM_PROMPT from the Complete Code. The four core rules provide the best balance for general-purpose RAG without over-constraining the model:
SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.

Context:
{context}
"""
  • Save app/services/rag_chain.py.

Additional rules are optional in production

Rules 5-7 from the expanded version can improve quality for specific use cases. In production, you would add rules based on the types of questions your users actually ask. For this project, the four core rules cover the most important behaviors: grounding, refusal, citation, and conciseness.

Model not following your prompt rules?

If the model ignores instructions (e.g., still hallucinates despite rule 1), check that the temperature value in your ChatBedrock config is low. Your current setting of 0.1 is ideal for factual RAG. Higher temperatures introduce randomness that can override prompt constraints.

Also verify your prompt actually includes the {context} placeholder. Without it, the model receives no retrieved documents and falls back to its training data.

help me improve my RAG prompt engineering

✔️ Awesome, I've got everything!

Great. Double check you have saved both files with their original values restored.

ⓧ I'd like to double check the full code

Here is app/services/chunking.py with the final settings:

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document


def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = splitter.create_documents(
        texts=[text],
        metadatas=[{"source": source}],
    )
    return chunks

Here is app/services/rag_chain.py with the final prompt:

from langchain_aws import ChatBedrock
from app.services.vector_store import similarity_search

llm = ChatBedrock(
    model_id="anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
    model_kwargs={"temperature": 0.1, "max_tokens": 1024},
)

SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Answer the question using ONLY the information in the context below.
2. If the context does not contain enough information to answer, say "I don't know based on the available documents."
3. Cite which source document(s) your answer comes from.
4. Keep answers concise and factual.

Context:
{context}
"""


def answer_question(question: str, top_k: int = 4) -> dict:
    relevant_docs = similarity_search(question, k=top_k)

    if not relevant_docs:
        return {
            "answer": "No documents have been ingested yet. Please upload documents first.",
            "sources": [],
        }

    context = "\n\n---\n\n".join(
        [
            f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
            for doc in relevant_docs
        ]
    )

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
        {"role": "user", "content": question},
    ]

    response = llm.invoke(messages)

    sources = list(set(doc.metadata.get("source", "unknown") for doc in relevant_docs))

    return {"answer": response.content, "sources": sources}

You now understand how chunking and prompt engineering directly affect RAG quality. Next up, you will take on the Secret Mission and build an evaluation endpoint that scores your RAG responses programmatically.

Secret mission

Add a RAG Evaluation Endpoint

Your RAG pipeline works, but how do you know it's working well? In this secret mission, you'll build a /evaluate endpoint that programmatically scores your RAG responses using a context relevance metric. No more eyeballing quality.

Clean Up Your Resources

Clean Up Your Resources

This project runs locally and uses AWS Bedrock on a pay-per-token basis. There are no ongoing charges when the server is idle.

Decide whether to keep your project files for future experimentation or delete them entirely.

Resources you used:

  • Local project folder containing app/, faiss_index/, requirements.txt, and your Python virtual environment.
  • AWS Bedrock API usage (pay-per-token only, no provisioned infrastructure to tear down).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your RAG API.

  • Your local project files and FAISS index persist on disk.
  • AWS Bedrock is serverless. You only pay when you actively send requests to the API.
  • Your virtual environment stays ready for next time you run uvicorn.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files stay on disk for when you return.

  • Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your Python virtual environment by running:
deactivate
  • To resume later, navigate back to your project folder, activate the virtual environment, and start the server again.

Delete - I don't want to use this again

Remove all local project files. There are no AWS resources to tear down since Bedrock is serverless with no provisioned infrastructure.

  • Stop the uvicorn server by pressing Ctrl+C if it is still running.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to the parent folder that contains your project directory (for example, your Desktop) and delete the project folder by running:
cd ~/Desktop
rm -rf rag-api

What if my folder has a different name?

Replace rag-api in the command above with whatever you named your project folder when you created it in Step 1.

  • On Windows, use this command instead:
cd ~\Desktop
rmdir /s /q rag-api
  • Confirm the folder is gone by running:
ls

You should no longer see your project folder in the output.

Nice Work!

Nice Work!

You just built a production-style RAG API from scratch. Your application ingests documents, chunks them, generates vector embeddings, retrieves relevant context, and generates grounded answers using Claude on AWS Bedrock.

You've learned how to:

  • Build a complete RAG pipeline with FastAPI /ingest and /query endpoints that ingest, chunk, embed, retrieve, and generate.
  • Orchestrate LangChain with AWS Bedrock (Claude for generation, Titan for embeddings) and a local FAISS vector store for similarity search.
  • Engineer prompts that ground LLM responses in retrieved context and experiment with chunking strategies to optimize retrieval quality.
  • Secret Mission: Implement a programmatic /evaluate endpoint with context relevance scoring to measure RAG response quality.

Ready to quiz yourself?