RAG-Powered API with Amazon Bedrock

Build a World Cup fan assistant API using RAG, FastAPI, and Amazon Bedrock.

Introduction

30 Second Summary

The 2026 FIFA World Cup spans 16 cities across three countries, with 104 matches over five weeks. Fans face a flood of questions about schedules, venues, and travel logistics that no single website answers quickly.

In this project, you will build a RAG-powered REST API that answers World Cup questions using verified tournament data, Amazon Bedrock for AI generation, and ChromaDB for vector search.

What You'll Build

A FastAPI application that ingests World Cup data into a vector store, generates embeddings via Amazon Bedrock Titan Embeddings V2, and answers fan questions using Amazon Nova Lite as the LLM.

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

  • A RAG-powered REST API with /query, /health, and /ingest endpoints that returns grounded answers with source citations.
  • A Dockerized application with a GitHub Actions CI/CD pipeline that lints, tests, and builds on every push.
  • A production-grade project structure with IAM least-privilege policies, structured logging, health checks, and a portfolio-ready README.
  • Secret Mission: Add a /matches/today endpoint that delivers AI-generated match-day briefings for fans.

Are there any prerequisites?

This project assumes you have an active AWS account with Bedrock model access enabled, Python 3.10+, Docker, Git, a GitHub account, and the AWS CLI configured.

Step 1 walks you through verifying each of these tools, so you can confirm everything is ready before building.

Set Up Project Structure and IAM

Every production application starts with a solid foundation. Before writing any RAG logic, you need a clean project structure, the right permissions, and verified connectivity to the AI models you will use.

In this step, you will scaffold the folder layout, lock down IAM permissions to only what the app needs, and confirm that Amazon Bedrock responds with real embeddings.

In this step, get ready to:
  • Create the project directory structure and Python virtual environment.
  • Create and attach a least-privilege IAM policy for Bedrock access.
  • Install pinned dependencies and verify Bedrock connectivity.
Create the project directory structure

A consistent folder layout makes your project easier to navigate, test, and containerize later. You will create four top-level directories following Python production conventions.

  • Create the project folders by running this command:
mkdir -p src data tests .github/workflows
  • Create the Python package marker files by running:
touch src/__init__.py tests/__init__.py

What do these folders do?

  • src/ holds your application code (API, RAG logic, ingestion pipeline).
  • data/ stores the World Cup knowledge base files.
  • tests/ contains your automated test suite.
  • .github/workflows/ is where your CI/CD pipeline definition lives.
  • __init__.py files mark directories as Python packages so imports work correctly.
  • Confirm the directories were created by running:
ls

You should see src, data, tests, and .github listed in the output.

Don't see all four folders?

If .github is missing from ls output, it may be hidden by default. Run ls -a to see hidden directories.

If any folder is missing entirely, re-run the mkdir -p command. The -p flag creates nested directories and does not error if they already exist.

Help me troubleshoot my folder structure

Now create a Python virtual environment to isolate your project dependencies.

  • Create the virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

Linux

source venv/bin/activate

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

Seeing a permissions error on Windows?

If PowerShell blocks the activate script, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try activating again.

Help me fix this activation error

Create the IAM policy for Bedrock access

Your application only needs one permission: the ability to invoke two specific Bedrock models. Following the principle of least privilege, you will create a policy that grants exactly that and nothing more.

  • Create a file called iam-policy.json in your project root and paste the following content:
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "BedrockInvokeModels",
            "Effect": "Allow",
            "Action": "bedrock:InvokeModel",
            "Resource": [
                "arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-lite-v1:0",
                "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
            ]
        }
    ]
}

What does this policy do?

  • bedrock:InvokeModel is the only action your app needs. It lets you send prompts and receive responses from Bedrock models.
  • The Resource array scopes access to exactly two models: Nova Lite (your LLM) and Titan Embeddings V2 (your embedding model). No other Bedrock models can be invoked with this policy.
  • The ARN pattern arn:aws:bedrock:us-east-1::foundation-model/ uses a double colon because foundation models are account-agnostic AWS resources.
  • Save iam-policy.json.
  • Create the policy in your AWS account by running:
aws iam create-policy --policy-name BedrockInvokeAccess --policy-document file://iam-policy.json

This command outputs a JSON response containing the policy ARN. Copy the value of the Arn field. You will need it for the next command.

  • Attach the policy to your IAM user by running:
aws iam attach-user-policy --user-name [[IAM_USER="your IAM username"]] --policy-arn [[POLICY_ARN="arn:aws:iam::123456789012:policy/BedrockInvokeAccess"]]

Using an IAM role instead?

If your AWS CLI is configured with a role rather than a user, replace the command above with:

aws iam attach-role-policy --role-name [[IAM_ROLE="your IAM role name"]] --policy-arn [[POLICY_ARN="arn:aws:iam::123456789012:policy/BedrockInvokeAccess"]]

Seeing an EntityAlreadyExists error?

This means a policy named BedrockInvokeAccess already exists in your account. You can either delete the old one first or choose a different name in the create-policy command.

Help me resolve this IAM policy error

Install dependencies and test Bedrock connectivity

With the folder structure and permissions in place, you can now install the Python packages your application depends on. Pinning exact versions ensures reproducible builds across your local machine, Docker, and CI/CD.

  • Create a file called requirements.txt in your project root and paste the following content:
fastapi==0.139.0
uvicorn==0.49.0
boto3==1.43.40
chromadb==1.5.9
pydantic==2.11.7
python-dotenv==1.1.0
httpx==0.28.1
ruff==0.12.0
pytest==8.4.0

What are these packages?

  • fastapi and uvicorn power the REST API server.
  • boto3 is the AWS SDK for calling Bedrock.
  • chromadb is the vector database for storing and searching embeddings.
  • pydantic handles request/response data validation.
  • python-dotenv loads environment variables from a .env file.
  • httpx is used by the test suite for async HTTP testing.
  • ruff and pytest handle linting and testing in the CI pipeline.
  • Save requirements.txt.
  • Install all dependencies by running:
pip install -r requirements.txt

This may take a minute or two as chromadb has several compiled dependencies.

  • Verify all packages installed correctly by running:
python -c "import fastapi, chromadb, boto3; print('All imports OK')"

You should see All imports OK printed in your terminal.

Seeing an ImportError?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, re-run the activate command from earlier in this step.

If a specific package failed to install, check the error output from pip install for missing system dependencies.

Help me troubleshoot my Python imports

Now create the configuration module that centralizes all your project settings.

  • Create a file called config.py inside the src/ folder and paste the following:
import os
from dotenv import load_dotenv

load_dotenv()

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"
LLM_MODEL_ID = "amazon.nova-lite-v1:0"
CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100
TOP_K = 5

What does this config do?

This file defines all the constants your application uses in one place. Model IDs, chunking parameters, and directory paths are set here so you never hardcode them in business logic.

load_dotenv() reads a .env file if one exists, letting you override defaults without changing code.

  • Save src/config.py.

Finally, verify that your IAM policy and Bedrock connectivity are working end-to-end.

  • Run the following Bedrock connectivity test:
python -c "
import json, boto3
client = boto3.client('bedrock-runtime', region_name='us-east-1')
response = client.invoke_model(
    modelId='amazon.titan-embed-text-v2:0',
    contentType='application/json',
    accept='application/json',
    body=json.dumps({'inputText': 'World Cup 2026 test embedding'}),
)
result = json.loads(response['body'].read())
print(f'Embedding dimensions: {len(result[\"embedding\"])}')
"

You should see Embedding dimensions: 1024 printed in your terminal. This confirms that your IAM permissions, AWS region, and model access are all correctly configured.

Seeing an AccessDeniedException?

This means your IAM policy is not attached correctly, or the Bedrock model access is not enabled in the AWS console. Double check that you attached the policy to the correct IAM user or role.

Also verify that you have enabled model access for both amazon.titan-embed-text-v2:0 and amazon.nova-lite-v1:0 in the Bedrock console under Model access.

Help me fix my Bedrock access error

✔️ 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:

iam-policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "BedrockInvokeModels",
            "Effect": "Allow",
            "Action": "bedrock:InvokeModel",
            "Resource": [
                "arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-lite-v1:0",
                "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
            ]
        }
    ]
}

requirements.txt

fastapi==0.139.0
uvicorn==0.49.0
boto3==1.43.40
chromadb==1.5.9
pydantic==2.11.7
python-dotenv==1.1.0
httpx==0.28.1
ruff==0.12.0
pytest==8.4.0

src/config.py

import os
from dotenv import load_dotenv

load_dotenv()

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"
LLM_MODEL_ID = "amazon.nova-lite-v1:0"
CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100
TOP_K = 5

Your project foundation is solid. You have a clean structure, locked-down permissions, and confirmed Bedrock connectivity. Next up, you will fill the data/ folder with World Cup knowledge and build the ingestion pipeline that turns raw text into searchable vectors.

Build the Knowledge Base

In the previous step, you set up your project structure, configured IAM permissions, and confirmed that Bedrock can generate embeddings. Now it's time to give your assistant something to be smart about.

A RAG system is only as good as the data it retrieves from. In this step, you'll create a knowledge base of 2026 World Cup tournament data and build a pipeline that converts those documents into searchable vector embeddings stored in ChromaDB.

In this step, get ready to:
  • Create World Cup data files with match schedules, venue details, and travel information.
  • Build an ingestion pipeline that chunks documents and generates embeddings via Bedrock.
  • Run the pipeline and verify semantic search returns relevant results.
Create the World Cup data files

Your RAG assistant needs source documents to retrieve from. You'll create three markdown files covering match schedules, venues, and travel tips.

Each file uses clear section headers. This matters because the chunking algorithm splits text at predictable intervals, and well-structured headers help each chunk stay self-contained and meaningful.

  • Create a file called data/schedule.md and paste the following content:
# 2026 FIFA World Cup Schedule

## Tournament Overview
- Dates: June 11 - July 19, 2026
- Host Countries: United States, Mexico, Canada
- Teams: 48 (expanded from 32)
- Total Matches: 104
- Venues: 16 stadiums across 16 host cities

## Key Dates
- Opening Match: June 11, 2026 - Mexico vs South Africa at Mexico City Stadium
- Group Stage: June 11 - June 27, 2026
- Round of 32: June 28 - July 3, 2026
- Round of 16: July 4 - July 7, 2026
- Quarter-finals: July 9 - July 11, 2026
- Semi-finals: July 14 - July 15, 2026 (Dallas and Atlanta)
- Third-place Match: July 18, 2026 at Hard Rock Stadium, Miami
- Final: July 19, 2026 at MetLife Stadium, New York/New Jersey

## Group Stage Format
- 12 groups of 4 teams each
- Top 2 from each group plus 8 best third-placed teams advance to Round of 32
- Three full days of rest between matches for 103 of 104 matches

## Notable Matches
- USA vs Paraguay: June 12, Los Angeles (SoFi Stadium)
- Canada vs Bosnia and Herzegovina: June 12, Toronto
- England vs Croatia: June 17, Dallas (AT&T Stadium)
- Brazil vs Morocco: June 13, New York/New Jersey (MetLife Stadium)
- Germany vs Curacao: June 14, Houston (NRG Stadium)
- 1000th World Cup Match: Tunisia vs Japan, June 20, Monterrey

What does this file contain?

This file covers the full tournament timeline: 48 teams playing 104 matches from June 11 to July 19, 2026. It includes key dates, the group stage format, and notable matchups across all 16 host cities.

  • Save data/schedule.md.
  • Create a file called data/venues.md and paste the following content:
# 2026 FIFA World Cup Venues

## United States (11 venues)
- Atlanta: Mercedes-Benz Stadium (8 matches, semi-final host)
- Boston: Gillette Stadium (7 matches)
- Dallas: AT&T Stadium (9 matches - most of any city, semi-final host)
- Houston: NRG Stadium (7 matches)
- Kansas City: Arrowhead Stadium (6 matches)
- Los Angeles: SoFi Stadium (8 matches, USA opener)
- Miami: Hard Rock Stadium (7 matches, third-place match)
- New York/New Jersey: MetLife Stadium (8 matches, THE FINAL)
- Philadelphia: Lincoln Financial Field (6 matches)
- San Francisco Bay Area: Levi's Stadium (6 matches)
- Seattle: Lumen Field (6 matches)

## Mexico (3 venues)
- Mexico City: Mexico City Stadium (5 matches, tournament opener)
- Guadalajara: Estadio Akron (4 matches)
- Monterrey: Estadio BBVA (4 matches)

## Canada (2 venues)
- Toronto: BMO Field (6 matches, Canada opener)
- Vancouver: BC Place (7 matches)

## Venue Facts
- First World Cup hosted across three countries
- Dallas hosts the most matches (9) of any single city
- Mexico City Stadium (formerly Estadio Azteca) hosts an opening match for the third time in World Cup history
- MetLife Stadium capacity: approximately 82,500 for the final
- All venues feature climate considerations for summer matches

What does this file contain?

This file maps all 16 stadiums across three countries: 11 in the USA, 3 in Mexico, and 2 in Canada. Each entry includes the city, stadium name, match count, and any special role (opener, final, semi-final).

  • Save data/venues.md.
  • Create a file called data/travel.md and paste the following content:
# 2026 FIFA World Cup Travel Guide

## Getting Around
The 2026 World Cup spans three countries and 16 cities. The schedule is designed to minimize travel, with matches clustered into Western, Central, and Eastern regions.

## Regions
- Western Region: Vancouver, Seattle, Los Angeles, San Francisco Bay Area, Guadalajara
- Central Region: Mexico City, Monterrey, Houston, Dallas, Kansas City
- Eastern Region: Atlanta, Miami, Philadelphia, New York/New Jersey, Boston, Toronto

## Key Travel Tips
- Matches within groups are clustered by region to reduce fan travel
- Host nations (USA, Mexico, Canada) play all three group games at home venues
- Three days rest between most matches allows travel time for fans following teams
- Major hub airports: LAX, DFW, JFK/EWR, ATL, MIA, YYZ

## Broadcast Information
- All 104 matches available on Peacock (English and Spanish)
- Telemundo broadcasts 92 matches
- Universo carries 12 matches
- Global broadcast partners in all FIFA member nations

## Time Zones
- Matches scheduled to accommodate global audiences
- Kick-off times range from 12:00 to 22:00 local time
- Eastern Time (ET) is the primary reference for US scheduling

What does this file contain?

This file helps fans plan their trip. It breaks the 16 cities into geographic regions, lists major hub airports, and includes broadcast information for following matches remotely.

  • Save data/travel.md.
  • Confirm all three files exist by running:
ls data/

You should see schedule.md, venues.md, and travel.md listed in the output.

Don't see all three files?

Make sure you created each file inside the data/ folder, not in the project root. If the data/ directory doesn't exist, create it with mkdir data first.

help me find my data files

Build the ingestion module

With raw data in place, you need a pipeline to convert those markdown files into searchable vectors. The ingestion module loads each file, splits it into chunks, generates an embedding for each chunk via Amazon Bedrock, and stores everything in ChromaDB.

  • Create a new file called src/ingest.py and start with the imports:
import os
import json
import boto3
import chromadb
from src.config import (
    AWS_REGION,
    EMBED_MODEL_ID,
    CHROMA_PERSIST_DIR,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
)

What do these imports do?

  • boto3 is the AWS SDK for calling Bedrock to generate embeddings.
  • chromadb is the vector database that stores and searches embeddings locally.
  • The config values (CHUNK_SIZE, CHUNK_OVERLAP, etc.) come from the src/config.py you created in Step 1.
  • You won't see a change yet. The next chunk adds the functions that use these imports.
  • Below the imports in src/ingest.py, add the Bedrock client helper and the document loading function:


def get_bedrock_client():
    return boto3.client("bedrock-runtime", region_name=AWS_REGION)


def load_documents(data_dir: str = "data") -> list[dict]:
    documents = []
    for filename in os.listdir(data_dir):
        if filename.endswith(".md"):
            filepath = os.path.join(data_dir, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append({"content": content, "source": filename})
    return documents

What does this code do?

  • get_bedrock_client() creates a boto3 client pointed at the Bedrock Runtime service in your configured region.
  • load_documents() scans the data/ folder for .md files and returns each as a dictionary with its content and source filename.
  • Tracking the source filename lets you attribute answers back to specific documents later.
  • You won't see a change yet. The next function adds the text chunking logic.
  • Below load_documents in src/ingest.py, add the text chunking function:


def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = end - overlap
    return chunks

Why chunk with overlap?

Splitting a document into 800-character chunks makes each piece small enough for the embedding model to capture its meaning accurately. The 100-character overlap ensures that sentences split across chunk boundaries still appear in at least one complete chunk.

Without overlap, a sentence like "The final is at MetLife Stadium on July 19" could get cut in half, making neither chunk useful for answering questions about the final.

  • You won't see a change yet. The next function handles embedding generation.
  • Below chunk_text in src/ingest.py, add the embedding generation function:


def generate_embedding(client, text: str) -> list[float]:
    response = client.invoke_model(
        modelId=EMBED_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": text}),
    )
    result = json.loads(response["body"].read())
    return result["embedding"]

What does this code do?

This function sends a text string to Titan Embeddings V2 via Bedrock and gets back a 1024-dimensional vector. The request body format is {"inputText": "..."} and the response contains an embedding field with the float array.

Each vector captures the semantic meaning of the text chunk. Similar questions and similar chunks end up close together in vector space, which is how semantic search works.

  • You won't see a change yet. The next function ties everything together into the full ingestion pipeline.
  • Below generate_embedding in src/ingest.py, add the main ingestion orchestrator and entry point:


def ingest_documents(data_dir: str = "data") -> int:
    client = get_bedrock_client()
    chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)

    collection = chroma_client.get_or_create_collection(
        name="worldcup2026",
        metadata={"hnsw:space": "cosine"},
    )

    documents = load_documents(data_dir)
    doc_id = 0

    for doc in documents:
        chunks = chunk_text(doc["content"])
        for chunk in chunks:
            embedding = generate_embedding(client, chunk)
            collection.add(
                ids=[f"doc_{doc_id}"],
                embeddings=[embedding],
                documents=[chunk],
                metadatas=[{"source": doc["source"]}],
            )
            doc_id += 1

    return doc_id


if __name__ == "__main__":
    count = ingest_documents()
    print(f"Ingested {count} chunks into ChromaDB")

What does this code do?

  • ingest_documents() orchestrates the full pipeline: load files, chunk each file, embed each chunk, and store it in ChromaDB.
  • PersistentClient saves the vector store to disk at ./chroma_db so embeddings survive between script runs.
  • The cosine similarity metric means vectors are compared by their angle, not magnitude. This is standard for text embeddings.
  • Each chunk gets a unique ID (doc_0, doc_1, etc.) and metadata tracking which source file it came from.
  • Save src/ingest.py.

Seeing an import error?

Make sure your virtual environment is activated. If you see ModuleNotFoundError: No module named 'chromadb', re-run pip install -r requirements.txt inside your activated environment.

help me fix my import error in ingest.py

✔️ Awesome, I've got everything!

Great. Double check you've saved src/ingest.py and move on to the next substep.

ⓧ I'd like to double check the full code

import os
import json
import boto3
import chromadb
from src.config import (
    AWS_REGION,
    EMBED_MODEL_ID,
    CHROMA_PERSIST_DIR,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
)


def get_bedrock_client():
    return boto3.client("bedrock-runtime", region_name=AWS_REGION)


def load_documents(data_dir: str = "data") -> list[dict]:
    documents = []
    for filename in os.listdir(data_dir):
        if filename.endswith(".md"):
            filepath = os.path.join(data_dir, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append({"content": content, "source": filename})
    return documents


def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = end - overlap
    return chunks


def generate_embedding(client, text: str) -> list[float]:
    response = client.invoke_model(
        modelId=EMBED_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": text}),
    )
    result = json.loads(response["body"].read())
    return result["embedding"]


def ingest_documents(data_dir: str = "data") -> int:
    client = get_bedrock_client()
    chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)

    collection = chroma_client.get_or_create_collection(
        name="worldcup2026",
        metadata={"hnsw:space": "cosine"},
    )

    documents = load_documents(data_dir)
    doc_id = 0

    for doc in documents:
        chunks = chunk_text(doc["content"])
        for chunk in chunks:
            embedding = generate_embedding(client, chunk)
            collection.add(
                ids=[f"doc_{doc_id}"],
                embeddings=[embedding],
                documents=[chunk],
                metadatas=[{"source": doc["source"]}],
            )
            doc_id += 1

    return doc_id


if __name__ == "__main__":
    count = ingest_documents()
    print(f"Ingested {count} chunks into ChromaDB")
Run the ingestion pipeline and verify the vector store

Time to execute the pipeline. This will read all three data files, chunk them, call Bedrock to generate embeddings for each chunk, and store everything in a local ChromaDB database.

  • Run the ingestion script by executing this command:
python -m src.ingest

You should see output like Ingested <number> chunks into ChromaDB confirming the pipeline completed. The exact number depends on how the text splits across chunk boundaries.

Why does this take a moment?

Each chunk requires a separate API call to Bedrock for embedding generation. With three data files split into many chunks, the script makes dozens of API calls sequentially. This is normal and should complete within 30-60 seconds.

Seeing an access denied error?

Verify your IAM policy is attached to the correct user or role. The policy must grant bedrock:InvokeModel on both model ARNs (amazon.nova-lite-v1:0 and amazon.titan-embed-text-v2:0). Also confirm your AWS CLI credentials are configured for the us-east-1 region.

help me fix my Bedrock access denied error

  • Now verify the vector store works by running a test similarity search. Create and run this quick test in your terminal:
python -c "
import chromadb
import json
import boto3
from src.config import CHROMA_PERSIST_DIR, EMBED_MODEL_ID, AWS_REGION

client = boto3.client('bedrock-runtime', region_name=AWS_REGION)
chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
collection = chroma_client.get_or_create_collection(name='worldcup2026')

print(f'Total chunks in collection: {collection.count()}')

response = client.invoke_model(
    modelId=EMBED_MODEL_ID,
    contentType='application/json',
    accept='application/json',
    body=json.dumps({'inputText': 'When is the World Cup final?'}),
)
result = json.loads(response['body'].read())
query_embedding = result['embedding']

results = collection.query(query_embeddings=[query_embedding], n_results=3)
print('\nTop results for: When is the World Cup final?')
for i, (doc, meta, dist) in enumerate(zip(results['documents'][0], results['metadatas'][0], results['distances'][0])):
    print(f'\n--- Result {i+1} (similarity: {1-dist:.4f}) ---')
    print(f'Source: {meta["source"]}')
    print(f'Content: {doc[:150]}...')
"

You should see the total chunk count printed, followed by search results that mention MetLife Stadium, New York/New Jersey, and July 19, 2026. The similarity scores should be above 0.7 for the top results.

What is a similarity score?

ChromaDB returns a distance value (lower is more similar for cosine). The test script converts this to a similarity score with 1 - distance. A score above 0.7 means the chunk is highly relevant to your query.

This is the core of semantic search: instead of keyword matching, you're finding text that means something similar to the question, even if it uses different words.

Results don't mention the final?

Check that your data/schedule.md file contains the line about the final at MetLife Stadium on July 19. If you re-run the ingestion after fixing data files, delete the ./chroma_db directory first to avoid duplicate chunks.

help me debug my vector search results

Your knowledge base is loaded and searchable. Next up, you'll build the FastAPI application that takes user questions, searches this vector store, and generates grounded answers using Amazon Nova Lite.

Build the RAG API

Your knowledge base is loaded and ready to answer questions. Now you need a way to actually ask those questions.

In this step, you will build the retrieval and generation engine that turns a user question into a grounded answer. Then you will wrap it in a FastAPI application with production-ready endpoints, middleware, and structured logging.

In this step, get ready to:
  • Build the RAG service that embeds questions, searches ChromaDB, and generates answers via Amazon Nova Lite.
  • Create a FastAPI application with health, query, and ingest endpoints.
  • Test the full flow locally and verify grounded answers with source citations.
Create the RAG service module

The RAG service is the core engine of your application. It takes a plain-text question, converts it to a vector embedding, finds the most relevant document chunks, and sends everything to the LLM for answer generation.

  • Create a new file called src/rag_service.py and add the imports and helper functions by pasting this code:
import json
import time
import chromadb
import boto3
from src.config import (
    AWS_REGION,
    EMBED_MODEL_ID,
    LLM_MODEL_ID,
    CHROMA_PERSIST_DIR,
    TOP_K,
)


# Initialize the Bedrock client for API calls
def get_bedrock_client():
    return boto3.client("bedrock-runtime", region_name=AWS_REGION)


# Connect to the persistent ChromaDB collection
def get_collection():
    chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
    return chroma_client.get_or_create_collection(
        name="worldcup2026",
        metadata={"hnsw:space": "cosine"},
    )


# Generate a vector embedding for a given text string
def generate_embedding(client, text: str) -> list[float]:
    response = client.invoke_model(
        modelId=EMBED_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": text}),
    )
    result = json.loads(response["body"].read())
    return result["embedding"]

What does this code do?

These helper functions set up the two services the RAG pipeline depends on.

  • get_bedrock_client() creates a connection to Amazon Bedrock Runtime in us-east-1.
  • get_collection() opens the same worldcup2026 ChromaDB collection you populated in the previous step.
  • generate_embedding() sends text to Titan Embeddings V2 and returns a 1024-dimension vector.

Now add the main query function. This is where the RAG magic happens.

  • Below the generate_embedding function in src/rag_service.py, paste the query_knowledge_base function:
def query_knowledge_base(question: str) -> dict:
    start_time = time.time()
    client = get_bedrock_client()
    collection = get_collection()

    # Embed the user's question into a vector
    query_embedding = generate_embedding(client, question)

    # Search ChromaDB for the top-5 most similar chunks
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=TOP_K,
    )

    context_chunks = results["documents"][0] if results["documents"] else []
    sources = results["metadatas"][0] if results["metadatas"] else []

    # Join retrieved chunks into a single context block
    context = "\n\n---\n\n".join(context_chunks)

    # Build the prompt instructing Nova Lite to use only the provided context
    prompt_body = {
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "text": (
                            f"You are a helpful 2026 FIFA World Cup assistant. "
                            f"Answer the following question using ONLY the provided context. "
                            f"If the context does not contain enough information, say so. "
                            f"Cite which source documents you used.\n\n"
                            f"Context:\n{context}\n\n"
                            f"Question: {question}"
                        )
                    }
                ],
            }
        ],
        "inferenceConfig": {"maxTokens": 512, "temperature": 0.3, "topP": 0.9},
    }

    # Invoke Nova Lite to generate the answer
    response = client.invoke_model(
        modelId=LLM_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(prompt_body),
    )

    response_body = json.loads(response["body"].read())
    answer = response_body["output"]["message"]["content"][0]["text"]

    # Calculate total processing time in milliseconds
    processing_time = (time.time() - start_time) * 1000

    return {
        "answer": answer,
        "sources": [s["source"] for s in sources],
        "processing_time_ms": round(processing_time, 2),
        "chunks_retrieved": len(context_chunks),
    }

What does this code do?

This function implements the full Retrieval-Augmented Generation pipeline in four stages:

  • Embed the question using the same Titan Embeddings V2 model that encoded the documents.
  • Search ChromaDB for the 5 chunks whose embeddings are closest to the question embedding (cosine similarity).
  • Construct a prompt that passes the retrieved context to Nova Lite with strict instructions to only use that context.
  • Return the answer along with source file names, processing time, and chunk count for transparency.

Why direct boto3 calls instead of LangChain?

Calling Bedrock directly with boto3 means you see every step of the RAG pipeline. You control the embedding call, the search parameters, and the exact prompt format.

This builds deeper understanding and is better for interviews. You can always add an abstraction layer later once you understand what happens underneath.

  • Save src/rag_service.py.

You will not see any output yet. The RAG service is a module that the API will call.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/rag_service.py before moving on.

ⓧ I'd like to double check the full code

import json
import time
import chromadb
import boto3
from src.config import (
    AWS_REGION,
    EMBED_MODEL_ID,
    LLM_MODEL_ID,
    CHROMA_PERSIST_DIR,
    TOP_K,
)


def get_bedrock_client():
    return boto3.client("bedrock-runtime", region_name=AWS_REGION)


def get_collection():
    chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
    return chroma_client.get_or_create_collection(
        name="worldcup2026",
        metadata={"hnsw:space": "cosine"},
    )


def generate_embedding(client, text: str) -> list[float]:
    response = client.invoke_model(
        modelId=EMBED_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": text}),
    )
    result = json.loads(response["body"].read())
    return result["embedding"]


def query_knowledge_base(question: str) -> dict:
    start_time = time.time()
    client = get_bedrock_client()
    collection = get_collection()

    query_embedding = generate_embedding(client, question)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=TOP_K,
    )

    context_chunks = results["documents"][0] if results["documents"] else []
    sources = results["metadatas"][0] if results["metadatas"] else []

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

    prompt_body = {
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "text": (
                            f"You are a helpful 2026 FIFA World Cup assistant. "
                            f"Answer the following question using ONLY the provided context. "
                            f"If the context does not contain enough information, say so. "
                            f"Cite which source documents you used.\n\n"
                            f"Context:\n{context}\n\n"
                            f"Question: {question}"
                        )
                    }
                ],
            }
        ],
        "inferenceConfig": {"maxTokens": 512, "temperature": 0.3, "topP": 0.9},
    }

    response = client.invoke_model(
        modelId=LLM_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(prompt_body),
    )

    response_body = json.loads(response["body"].read())
    answer = response_body["output"]["message"]["content"][0]["text"]

    processing_time = (time.time() - start_time) * 1000

    return {
        "answer": answer,
        "sources": [s["source"] for s in sources],
        "processing_time_ms": round(processing_time, 2),
        "chunks_retrieved": len(context_chunks),
    }
Create the FastAPI application

With the RAG engine ready, you need an HTTP layer so users can send questions and receive answers. FastAPI gives you automatic request validation, interactive documentation, and async support out of the box.

  • Create a new file called src/main.py and add the imports, app setup, CORS middleware, and Pydantic models:
import logging
import json
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import chromadb
from src.config import CHROMA_PERSIST_DIR
from src.rag_service import query_knowledge_base
from src.ingest import ingest_documents

# Configure structured logging for production observability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

# Initialize the FastAPI application with metadata
app = FastAPI(
    title="World Cup 2026 Fan Assistant",
    description="RAG-powered API for 2026 FIFA World Cup information",
    version="1.0.0",
)

# Enable cross-origin requests so frontends can call the API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Define request and response schemas with Pydantic
class QueryRequest(BaseModel):
    question: str


class QueryResponse(BaseModel):
    answer: str
    sources: list[str]
    processing_time_ms: float
    chunks_retrieved: int


class HealthResponse(BaseModel):
    status: str
    documents: int


class IngestResponse(BaseModel):
    status: str
    chunks_ingested: int

What does this code do?

  • Structured logging writes timestamped JSON entries so you can trace requests in production.
  • CORS middleware allows any frontend (React, mobile app, Postman) to call your API without browser security blocks.
  • Pydantic models define the exact shape of incoming requests and outgoing responses. FastAPI validates them automatically and generates documentation from them.

Now add the three API endpoints that expose the RAG functionality over HTTP.

  • Below the Pydantic models in src/main.py, paste the endpoint functions:
@app.get("/health", response_model=HealthResponse)
def health_check():
    try:
        chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
        collection = chroma_client.get_or_create_collection(name="worldcup2026")
        doc_count = collection.count()
        return HealthResponse(status="healthy", documents=doc_count)
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        raise HTTPException(status_code=503, detail="Service unavailable")


@app.post("/query", response_model=QueryResponse)
def query_endpoint(request: QueryRequest):
    logger.info(json.dumps({"event": "query", "question": request.question}))
    try:
        result = query_knowledge_base(request.question)
        return QueryResponse(**result)
    except Exception as e:
        logger.error(f"Query failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/ingest", response_model=IngestResponse)
def ingest_endpoint():
    logger.info(json.dumps({"event": "ingest_start"}))
    try:
        count = ingest_documents()
        logger.info(json.dumps({"event": "ingest_complete", "chunks": count}))
        return IngestResponse(status="complete", chunks_ingested=count)
    except Exception as e:
        logger.error(f"Ingestion failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))

What do these endpoints do?

  • GET /health checks that ChromaDB is reachable and returns the document count. Returns a 503 if the service is down.
  • POST /query accepts a JSON body with a question field, runs the full RAG pipeline, and returns the answer with sources.
  • POST /ingest re-runs the document ingestion pipeline so you can refresh the knowledge base without restarting the server.

Each endpoint wraps its logic in try/except and returns meaningful HTTP status codes (503 for service unavailable, 500 for internal errors). Structured JSON logs capture every query for debugging.

  • Save src/main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/main.py before moving on.

ⓧ I'd like to double check the full code

import logging
import json
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import chromadb
from src.config import CHROMA_PERSIST_DIR
from src.rag_service import query_knowledge_base
from src.ingest import ingest_documents

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="World Cup 2026 Fan Assistant",
    description="RAG-powered API for 2026 FIFA World Cup information",
    version="1.0.0",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class QueryRequest(BaseModel):
    question: str


class QueryResponse(BaseModel):
    answer: str
    sources: list[str]
    processing_time_ms: float
    chunks_retrieved: int


class HealthResponse(BaseModel):
    status: str
    documents: int


class IngestResponse(BaseModel):
    status: str
    chunks_ingested: int


@app.get("/health", response_model=HealthResponse)
def health_check():
    try:
        chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
        collection = chroma_client.get_or_create_collection(name="worldcup2026")
        doc_count = collection.count()
        return HealthResponse(status="healthy", documents=doc_count)
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        raise HTTPException(status_code=503, detail="Service unavailable")


@app.post("/query", response_model=QueryResponse)
def query_endpoint(request: QueryRequest):
    logger.info(json.dumps({"event": "query", "question": request.question}))
    try:
        result = query_knowledge_base(request.question)
        return QueryResponse(**result)
    except Exception as e:
        logger.error(f"Query failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/ingest", response_model=IngestResponse)
def ingest_endpoint():
    logger.info(json.dumps({"event": "ingest_start"}))
    try:
        count = ingest_documents()
        logger.info(json.dumps({"event": "ingest_complete", "chunks": count}))
        return IngestResponse(status="complete", chunks_ingested=count)
    except Exception as e:
        logger.error(f"Ingestion failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))
Test the full flow locally

Both modules are in place. Time to start the server and fire a real question at your RAG pipeline.

  • Start the FastAPI development server by running this command:
uvicorn src.main:app --reload

What does this command do?

  • uvicorn is the ASGI server that runs your FastAPI app.
  • src.main:app tells uvicorn to find the app object inside src/main.py.
  • --reload watches for file changes and auto-restarts the server during development.
  • Wait for the terminal to show Uvicorn running on http://127.0.0.1:8000.
  • Open your browser and navigate to http://localhost:8000/docs to see the interactive Swagger UI.

You should see three endpoints listed: /health, /query, and /ingest.

What is Swagger UI?

FastAPI automatically generates interactive API documentation at the /docs path. You can test every endpoint directly in the browser without writing any client code.

  • Open a second terminal window (keep the server running in the first one).
  • Test the query endpoint by running this command:
curl -X POST http://localhost:8000/query -H "Content-Type: application/json" -d '{"question": "Where is the World Cup final?"}'
  • Verify the response. You should see a JSON object containing an answer that mentions MetLife Stadium, New York/New Jersey, and July 19, 2026. The sources field should list the markdown files the answer was drawn from.

Not seeing a response?

  • Make sure the uvicorn server is still running in your first terminal. If it crashed, check for import errors or typos.
  • Confirm your AWS credentials are configured and your IAM policy grants bedrock:InvokeModel for both model ARNs.
  • If you see a ChromaDB error, verify that the ./chroma_db directory exists from your ingestion step.

My curl command returns an error from the RAG API.

  • Also test the health endpoint to confirm the document count:
curl http://localhost:8000/health

You should see a response like {"status": "healthy", "documents": <count>} where the count matches the number of chunks you ingested earlier.

Your RAG API is live and answering questions with grounded, cited responses. Next up, you will containerize the entire application with Docker so it runs the same everywhere.

Containerize with Docker

Your RAG API works locally, but anyone who wants to run it needs to install Python, every dependency, and configure the same environment you have. Docker solves this by packaging your entire application into a portable container that runs identically on any machine.

In this step, you will write a production-grade Dockerfile using a multi-stage build pattern, wire up Docker Compose for local development, and verify the containerized application returns the same RAG responses as your local server.

In this step, get ready to:
  • Create a multi-stage Dockerfile with a non-root user and health check.
  • Create a Docker Compose file that maps AWS credentials and mounts persistent storage.
  • Build and verify the containerized application responds correctly.
Create the Dockerfile with a multi-stage build

A multi-stage build separates dependency installation from the final runtime image. The first stage downloads and compiles packages, and the second stage copies only the installed artifacts into a slim image.

  • Create a file called Dockerfile in the root of your project folder (the same level as requirements.txt).
  • Paste the following into Dockerfile:
FROM python:3.12-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim

RUN useradd --create-home appuser
WORKDIR /app

COPY --from=builder /install /usr/local
COPY src/ ./src/
COPY data/ ./data/

RUN chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

The build is split into two stages for a smaller final image:

  • builder stage: installs all Python packages from requirements.txt into a dedicated prefix (/install). This layer caches well and avoids bloating the runtime image with build tools.
  • Runtime stage: starts fresh from python:3.12-slim, copies the pre-built packages from the builder, then copies your src/ and data/ directories.
  • useradd --create-home appuser creates a non-root user so the container never runs as root.
  • HEALTHCHECK pings the /health endpoint every 30 seconds using Python's built-in urllib library. Docker marks the container unhealthy if three consecutive checks fail.
  • Save Dockerfile.

Why a non-root user?

Running containers as root is a security risk. If an attacker exploits a vulnerability in your app, they gain root access to the container. The USER appuser instruction limits what a compromised process can do.

Create the Docker Compose file for local development

Docker Compose lets you define how to build and run your container in a single configuration file. It also handles environment variables and persistent volumes.

  • Create a file called docker-compose.yml in the root of your project folder (same level as Dockerfile).
  • Paste the following into docker-compose.yml:
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - AWS_REGION=${AWS_REGION:-us-east-1}
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
      - AWS_SESSION_TOKEN=${AWS_SESSION_TOKEN:-}
      - CHROMA_PERSIST_DIR=/app/chroma_db
    volumes:
      - chroma_data:/app/chroma_db

volumes:
  chroma_data:

What does this Compose file do?

This configuration defines one service called app that builds from the local Dockerfile:

  • ports: maps container port 8000 to your machine's port 8000.
  • environment: passes your local AWS credentials into the container so boto3 can call Bedrock. Credentials are read from your shell environment at runtime and never baked into the image.
  • volumes: mounts a named volume (chroma_data) to /app/chroma_db inside the container. This persists the ChromaDB data between container restarts.

Never bake credentials into images

The ${AWS_ACCESS_KEY_ID} syntax reads the value from your shell environment at runtime. If you hardcoded a key in the Dockerfile or Compose file, anyone who pulls the image could extract it.

  • Save docker-compose.yml.

✔️ Awesome, I've got everything!

Great. Double check you have saved both Dockerfile and docker-compose.yml.

ⓧ I'd like to double check the full code

Here is the complete Dockerfile:

FROM python:3.12-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim

RUN useradd --create-home appuser
WORKDIR /app

COPY --from=builder /install /usr/local
COPY src/ ./src/
COPY data/ ./data/

RUN chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

And the complete docker-compose.yml:

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - AWS_REGION=${AWS_REGION:-us-east-1}
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
      - AWS_SESSION_TOKEN=${AWS_SESSION_TOKEN:-}
      - CHROMA_PERSIST_DIR=/app/chroma_db
    volumes:
      - chroma_data:/app/chroma_db

volumes:
  chroma_data:
Build and verify the containerized application

Time to build the image and confirm the container serves the same RAG responses you tested locally.

  • If you still have uvicorn running from the previous step, stop it by pressing Ctrl+C in that terminal. Port 8000 must be free for Docker to bind to it.
  • Build and start the container by running:
docker compose up --build

What should I see?

Docker will download the python:3.12-slim base image (first time only), install your dependencies, and start the uvicorn server inside the container. You will see log lines ending with something like Uvicorn running on http://0.0.0.0:8000.

Build failed or port in use?

  • If you see a port conflict error, make sure you stopped the local uvicorn process from earlier.
  • If the build fails on pip install, confirm your requirements.txt is saved and has no typos.
  • If Docker itself is not running, open Docker Desktop first, then retry the command.

Ask for help: My docker compose up --build is failing. How do I fix it?

  • Open a new terminal tab (keep the Docker container running in the first tab).
  • Test the health endpoint by running:
curl http://localhost:8000/health
  • You should see a JSON response like {"status": "healthy", "documents": 0}.

Why does it show 0 documents?

The named volume chroma_data starts empty on first run. The container has a fresh ChromaDB directory. You need to ingest your data inside the container.

  • Trigger the ingestion endpoint so the container builds its vector store:
curl -X POST http://localhost:8000/ingest
  • You should see a response like {"status": "complete", "chunks_ingested": <count>} confirming the data was loaded.
  • Now test the RAG query endpoint by running:
curl -X POST http://localhost:8000/query -H "Content-Type: application/json" -d '{"question": "Where is the World Cup final?"}'
  • You should see a JSON response containing an answer that mentions MetLife Stadium, New York/New Jersey, and July 19, 2026, along with source citations.
  • Confirm structured logs are visible by running:
docker compose logs
  • You should see JSON-formatted log entries showing the query event you just triggered.

Query returning an error?

  • If you see a 500 error mentioning credentials, make sure your AWS environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) are exported in the terminal where you ran docker compose up.
  • If you see 0 documents in the health check, run the /ingest endpoint first to populate the vector store inside the container.

Ask for help: My containerized query endpoint is returning an error. How do I debug it?

Your RAG API is now fully containerized. Anyone with Docker installed can run your project with a single command. Next up, you will automate quality checks with a GitHub Actions CI/CD pipeline.

Set Up CI/CD with GitHub Actions

In the last step, you containerized the Fan Assistant with Docker. Your app runs identically inside a container and on your local machine.

A working application is only half the story. Without automated quality checks, bugs can sneak in every time you push new code. CI/CD (Continuous Integration / Continuous Delivery) solves this by running linting, tests, and builds automatically on every push.

In this step, you will create a GitHub Actions workflow that runs three jobs: lint your code with ruff, run tests with pytest, and build your Docker image.

In this step, get ready to:
  • Create a GitHub Actions CI workflow with lint, test, and build jobs.
  • Write automated tests that mock Bedrock API calls.
  • Push to GitHub and verify all three jobs pass with green checkmarks.
Create the GitHub Actions workflow

GitHub Actions workflows live in a special directory at the root of your repository. When you push code, GitHub detects YAML files in this directory and executes the jobs they define.

  • Create the workflow directory by running this command:
mkdir -p .github/workflows
  • Confirm the directory exists by running:
ls .github/workflows

You should see an empty directory listed without errors.

  • Create the CI workflow file by adding the following code to .github/workflows/ci.yml:
name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install ruff
      - run: ruff check src/ tests/

  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: pytest tests/ -v

  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: docker build -t worldcup-fan-assistant .

What does this workflow do?

The on: section triggers the workflow whenever code is pushed to main or a pull request targets main.

  • The lint job installs ruff and checks both src/ and tests/ for style violations.
  • The test job installs all project dependencies and runs pytest with verbose output.
  • The build job verifies the Dockerfile compiles into a valid image.
  • Save .github/workflows/ci.yml.

You won't see a change yet. The workflow runs on GitHub after you push, which you will do after writing the tests.

Write API tests with mocked Bedrock calls

Your CI test job needs actual test files to run. The challenge is that your API calls Amazon Bedrock, which requires real AWS credentials. You solve this by mocking the Bedrock client so tests run anywhere without credentials.

  • Add the health endpoint test by pasting the following code into tests/test_api.py:
from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
from src.main import app

client = TestClient(app)


def test_health_endpoint():
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
    assert "documents" in data

What does this code do?

  • TestClient from FastAPI lets you send HTTP requests to your app without starting a real server.
  • test_health_endpoint() calls GET /health and verifies the response has a 200 status code, a "healthy" status field, and a documents count.
  • This test works without mocking because the health endpoint only reads from ChromaDB, which exists locally.
  • Add the query endpoint test below the test_health_endpoint function in the same file:
@patch("src.rag_service.get_bedrock_client")
@patch("src.rag_service.get_collection")
def test_query_endpoint(mock_collection, mock_bedrock):
    mock_col = MagicMock()
    mock_col.query.return_value = {
        "documents": [["The final is at MetLife Stadium on July 19, 2026."]],
        "metadatas": [[{"source": "schedule.md"}]],
    }
    mock_collection.return_value = mock_col

    mock_client = MagicMock()
    mock_embed_response = MagicMock()
    mock_embed_response.__getitem__ = lambda self, key: (
        MagicMock(read=lambda: b'{"embedding": [0.1] * 1024}') if key == "body" else None
    )
    mock_llm_response = MagicMock()
    mock_llm_response.__getitem__ = lambda self, key: (
        MagicMock(
            read=lambda: b'{"output": {"message": {"content": [{"text": "The World Cup final is at MetLife Stadium."}]}}}'
        )
        if key == "body"
        else None
    )
    mock_client.invoke_model.side_effect = [mock_embed_response, mock_llm_response]
    mock_bedrock.return_value = mock_client

    response = client.post("/query", json={"question": "Where is the final?"})
    assert response.status_code == 200
    data = response.json()
    assert "answer" in data
    assert "sources" in data

What does this test do?

The @patch decorators replace the real Bedrock client and ChromaDB collection with mock objects during the test.

  • mock_col.query.return_value fakes the vector search result so ChromaDB appears to find a relevant chunk about MetLife Stadium.
  • mock_client.invoke_model.side_effect returns a fake embedding on the first call and a fake LLM response on the second call.
  • The test then sends a POST request and verifies the response contains answer and sources fields.
  • Save tests/test_api.py.
  • Verify the tests pass locally by running:
pytest tests/ -v

You should see both tests pass with green output showing 2 passed.

Tests failing locally?

  • If you see ModuleNotFoundError, make sure your virtual environment is activated and all packages from requirements.txt are installed.
  • If the health test fails, confirm that your chroma_db directory exists from the ingestion step.
  • If test_query_endpoint fails, double-check that the @patch paths match exactly: src.rag_service.get_bedrock_client and src.rag_service.get_collection.

Help me debug my failing pytest tests.

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_api.py and that both tests pass locally.

ⓧ I'd like to double check the full code

from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
from src.main import app

client = TestClient(app)


def test_health_endpoint():
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
    assert "documents" in data


@patch("src.rag_service.get_bedrock_client")
@patch("src.rag_service.get_collection")
def test_query_endpoint(mock_collection, mock_bedrock):
    mock_col = MagicMock()
    mock_col.query.return_value = {
        "documents": [["The final is at MetLife Stadium on July 19, 2026."]],
        "metadatas": [[{"source": "schedule.md"}]],
    }
    mock_collection.return_value = mock_col

    mock_client = MagicMock()
    mock_embed_response = MagicMock()
    mock_embed_response.__getitem__ = lambda self, key: (
        MagicMock(read=lambda: b'{"embedding": [0.1] * 1024}') if key == "body" else None
    )
    mock_llm_response = MagicMock()
    mock_llm_response.__getitem__ = lambda self, key: (
        MagicMock(
            read=lambda: b'{"output": {"message": {"content": [{"text": "The World Cup final is at MetLife Stadium."}]}}}'
        )
        if key == "body"
        else None
    )
    mock_client.invoke_model.side_effect = [mock_embed_response, mock_llm_response]
    mock_bedrock.return_value = mock_client

    response = client.post("/query", json={"question": "Where is the final?"})
    assert response.status_code == 200
    data = response.json()
    assert "answer" in data
    assert "sources" in data
Push to GitHub and create the README

With your workflow and tests ready, it is time to push everything to GitHub so the CI pipeline can run.

  • Create a new repository on GitHub. Go to github.com/new in your browser.
  • Name it worldcup-fan-assistant and leave it public.
  • Click Create repository without initializing with a README (you will add your own).
  • Create a .gitignore file to exclude generated files by running:
echo "chroma_db/
__pycache__/
*.pyc
venv/
.env" > .gitignore
  • Initialize your git repository and push by running these commands:
git init
git add .
git commit -m "Initial commit: RAG API with CI/CD"
git branch -M main
git remote add origin https://github.com/[[GITHUB_USERNAME="your-github-username"]]/worldcup-fan-assistant.git
git push -u origin main

What do these commands do?

  • git init creates a new git repository in your project folder.
  • git add . stages all files (except those in .gitignore).
  • git branch -M main renames the default branch to main, which matches your workflow trigger.
  • git push -u origin main pushes your code and sets main as the default upstream branch.
  • Now create README.md in your project root with the following content:
# World Cup 2026 Fan Assistant

RAG-powered REST API for 2026 FIFA World Cup information, built with FastAPI, Amazon Bedrock, and ChromaDB.

## Architecture

```
User Question -> FastAPI -> Titan Embeddings V2 (embed) -> ChromaDB (search) -> Nova Lite (generate) -> Response
```

### Components
- **FastAPI** - REST API framework with automatic OpenAPI documentation
- **Amazon Bedrock** - Managed AI (Titan Embeddings V2 + Nova Lite)
- **ChromaDB** - Local vector database for semantic search
- **Docker** - Containerized deployment with multi-stage builds
- **GitHub Actions** - CI/CD pipeline (lint, test, build)

## Setup Instructions

1. Clone the repository and create a virtual environment:
```bash
git clone https://github.com/<your-username>/worldcup-fan-assistant.git
cd worldcup-fan-assistant
python -m venv venv
source venv/bin/activate
```

2. Install dependencies:
```bash
pip install -r requirements.txt
```

3. Configure AWS credentials with Bedrock access in us-east-1.

4. Ingest the knowledge base:
```bash
python -m src.ingest
```

5. Run the API:
```bash
uvicorn src.main:app --reload
```

## Environment Variables

| Variable | Default | Description |
|----------|---------|-------------|
| `AWS_REGION` | `us-east-1` | AWS region for Bedrock calls |
| `AWS_ACCESS_KEY_ID` | - | AWS access key |
| `AWS_SECRET_ACCESS_KEY` | - | AWS secret key |
| `AWS_SESSION_TOKEN` | - | Optional session token |
| `CHROMA_PERSIST_DIR` | `./chroma_db` | ChromaDB storage path |

## Example API Calls

### Health Check
```bash
curl http://localhost:8000/health
```

### Query the Knowledge Base
```bash
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "Where is the World Cup final?"}'
```

### Trigger Re-ingestion
```bash
curl -X POST http://localhost:8000/ingest
```

## Cost Breakdown

| Service | Usage | Estimated Cost |
|---------|-------|----------------|
| Bedrock Titan Embeddings V2 | ~50 embeddings during ingest + queries | < $0.01 |
| Bedrock Nova Lite | ~20 query responses during development | < $0.50 |
| S3 | Not used (local ChromaDB) | $0.00 |
| **Total** | | **< $1.00** |

What sections does the README cover?

  • Architecture overview shows the data flow from question to answer.
  • Setup instructions give new contributors a step-by-step path to run the project.
  • Environment variable reference documents every configurable setting.
  • Example API calls provide copy-paste curl commands.
  • Cost breakdown shows this project costs under $1 total during development.
  • Save README.md.
  • Push the README to GitHub by running:
git add .
git commit -m "Add README with architecture and setup docs"
git push
  • Open your repository on GitHub in your browser.
  • Click the Actions tab at the top of your repository.
  • Wait for the workflow run to complete. You should see three green checkmarks next to the lint, test, and build jobs.

✔️ I see three green checkmarks

Your CI pipeline is working. Every future push to main will automatically lint, test, and build your project.

ⓧ One or more jobs failed

  • Click on the failed job name to see the full log output.
  • If lint failed, run ruff check src/ tests/ locally to see the exact style issues. Fix them and push again.
  • If test failed, check that tests/test_api.py matches the code above exactly. Missing imports or typos in the mock paths are the most common cause.
  • If build failed, check that your Dockerfile is committed and has no syntax errors.

Help me debug my failed GitHub Actions workflow.

Verify README rendering

Scroll down on your repository's main page. GitHub renders README.md automatically. Confirm the architecture diagram, table formatting, and code blocks all display correctly.

Secret mission

Live Match Day Endpoint

Your API answers general World Cup questions. But what if fans could ask "What's happening today?" and get an AI-generated match preview with venue details and an exciting narrative? Add a live match-day endpoint that turns your RAG system into a real-time fan companion.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with Docker and uses Amazon Bedrock on a pay-per-use basis. There are no ongoing AWS charges when you stop making API calls.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Local project folder on your Desktop (all source code, data files, and virtual environment).
  • Docker image (worldcup-fan-assistant) and named volume (chroma_data).
  • GitHub repository with CI workflow.
  • IAM policy attached to your AWS user or role for Bedrock access.

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the project soon.

  • Your Docker image and ChromaDB volume stay on disk for instant re-launch.
  • Bedrock charges nothing when you are not making API calls.
  • The IAM policy remains attached but grants no access beyond Bedrock model invocation.

Pause - I'll come back to this later

Shut down running containers to free up system memory. Your files and Docker image stay intact so you can resume later.

  • Stop and remove the running containers by running:
docker compose down

Zero cost when containers are stopped. Your image and volume remain on disk.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below.

  • Stop containers and remove the Docker volume by running:
docker compose down -v
  • Remove the Docker image by running:
docker image rm worldcup-fan-assistant
  • Remove any dangling images and unused volumes by running:
docker image prune -f
docker volume prune -f
  • Delete the local project folder. On macOS or Linux, run:
cd ~/Desktop
rm -rf worldcup-fan-assistant
  • On Windows, run:
cd ~\Desktop
rmdir /s /q worldcup-fan-assistant
  • Delete the GitHub repository by opening your repo page on GitHub, clicking Settings, scrolling to the bottom of the General page, and clicking Delete this repository under the Danger Zone section.
  • Remove the IAM policy from your AWS account. Open the IAM console, select Policies in the left pane, find your Bedrock policy, select it, choose Actions, then click Delete. Confirm the deletion when prompted.

Well Done!

Well Done!

Nice work! You just built a production-grade RAG application that answers real fan questions about the 2026 FIFA World Cup using verified tournament data.

You've learned how to:

  • Build a RAG pipeline that ingests documents, generates vector embeddings with Amazon Titan Embeddings V2, and performs semantic search via ChromaDB.
  • Create a FastAPI application with health checks, structured logging, and error handling, then containerize it with Docker multi-stage builds and least-privilege IAM policies.
  • Automate quality gates with GitHub Actions for linting, testing, and Docker build verification on every push.
  • Secret Mission: Added a live match-day briefing endpoint with date-based filtering and dynamic prompt engineering.

Ready to quiz yourself?