RAG Policy Q&A with Amazon Bedrock

Build a grounded document Q&A system using Bedrock Knowledge Bases and RAG.

Introduction

30 Second Summary

Every company has policy documents that employees constantly search through. Finding the right answer about PTO accrual or expense limits usually means scrolling through dozens of pages. What if you could just ask a question and get the exact answer with a citation pointing to the source paragraph?

In this project, you will build a document Q&A system powered by Amazon Bedrock that answers questions about company policies using RAG (Retrieval-Augmented Generation). You will experience the hallucination problem firsthand, then solve it with a knowledge base, a data validation pipeline, and governed prompt templates.

What You'll Build

You ask your system a question like "What's our PTO policy for someone with 4 years of experience?" and it responds with the correct answer, citing the exact policy section from your uploaded documents.

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

  • A working Q&A system where you ask natural language questions and get accurate, cited answers grounded in your actual policy documents.
  • A data validation pipeline that catches malformed or oversized documents before they corrupt your knowledge base, with CloudWatch metrics tracking ingestion health.
  • Governed prompt templates managed through Bedrock Prompt Management, so changes to your Q&A prompt are versioned, auditable, and rollback-safe.
  • Secret Mission: Extend the pipeline to process image-based policy documents using Bedrock's multimodal capabilities.

Are there any prerequisites?

You need an AWS account with billing enabled and the AWS CLI configured with credentials. This project assumes you are comfortable navigating the AWS console and have worked with Python and boto3 before.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters. You're creating a RAG-powered document Q&A system using Amazon Bedrock that demonstrates the core Domain 1 skills tested on the AWS Generative AI Developer Professional exam: FM integration, data validation, vector storage, retrieval mechanisms, and prompt governance.

Set Up Your Environment

Before you can build a RAG system, you need three things in place: model access in Amazon Bedrock, a Python environment for scripting, and an S3 bucket to hold your documents.

Without model access enabled, every API call to Bedrock will fail. Without a local Python environment, you cannot run the validation and query scripts. This step gets all the foundational pieces ready so you can focus on building in the steps ahead.

In this step, get ready to:
  • Enable Bedrock model access for Titan Embeddings V2 and Claude 3 Haiku.
  • Create a Python project with a virtual environment and S3 bucket.
  • Add sample policy documents that your RAG system will query.
Enable Bedrock model access

Amazon Bedrock automatically enables most serverless foundation models. Anthropic models require a one-time usage form before you can invoke them.

  • Sign in to the AWS Management Console at https://console.aws.amazon.com.
  • Confirm your region is set to US East (N. Virginia) in the top-right dropdown.
  • Search for Bedrock in the search bar and open the Amazon Bedrock console.
  • In the left navigation pane, expand Bedrock configurations and select Model access.
  • Confirm that Amazon Titan Text Embeddings V2 shows Access granted. Amazon models are enabled by default.
  • For Anthropic Claude 3 Haiku, click Modify model access if it does not already show as granted.
  • Select the checkbox next to Anthropic Claude 3 Haiku and click Next.
  • Complete the one-time Anthropic usage form when prompted, then click Submit.

Why two models?

Titan Embeddings V2 converts your documents into vector embeddings for semantic search. Claude 3 Haiku generates natural-language answers from the retrieved context. Together they form the two halves of a RAG pipeline: retrieval and generation.

Access changes can take a few minutes to propagate. While you wait, note the model IDs from the Model access page.

  • Titan Embeddings V2: amazon.titan-embed-text-v2:0.
  • Claude 3 Haiku: anthropic.claude-3-haiku-20240307-v1:0.

Verify your model access is active by running this command in your terminal:

aws bedrock list-foundation-models --region us-east-1 --query "modelSummaries[?contains(modelId, 'titan-embed-text-v2')]"

What does this command do?

This calls the Bedrock API to list all available foundation models, then filters the results using a JMESPath query to show only entries matching titan-embed-text-v2. If access is working, you see a JSON array with the model details.

Seeing an empty array or access denied?

If you see [] (empty), the region may not match. Confirm your CLI is targeting us-east-1. If you see an AccessDeniedException, your IAM user or role needs the AmazonBedrockFullAccess managed policy attached.

help me troubleshoot Bedrock model access errors

Set up Python environment and S3 bucket

You need Python 3.12+ for this project. Check which version you have installed.

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

✔️ I see version 3.12 or higher

Your Python version is ready. Continue to the next instruction below.

ⓧ I see a version lower than 3.12

You need to upgrade. On macOS, the system Python is often outdated and shadows newer installs.

  • Install the latest Python from https://www.python.org/downloads/.
  • After installing, run python3.12 --version to confirm the new version is accessible.
  • Use python3.12 instead of python3 for all commands in this project if your system default remains older.

ⓧ Command not found

Python is not installed on your system.

  • Download and install Python 3.12+ from https://www.python.org/downloads/.
  • During installation on Windows, check the box that says Add Python to PATH.
  • After installing, close and reopen your terminal, then run python3 --version again to confirm.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir bedrock-rag-project
cd bedrock-rag-project
  • Create a Python virtual environment and activate it by running:
python3 -m venv venv
source venv/bin/activate

Windows activation

On Windows, activate the virtual environment with venv\Scripts\activate instead.

  • Create the requirements.txt file with the project dependency by running:
echo 'boto3>=1.35.0' > requirements.txt
  • Install the dependency by running:
pip install -r requirements.txt

You should see output confirming that boto3 and its dependencies were installed successfully.

Pip install failing?

Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, run the activate command again.

help me fix pip install errors in my virtual environment

Next, set your AWS account ID as an environment variable so the config file can reference it.

  • Export your AWS account ID by running:
export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
echo $AWS_ACCOUNT_ID

You should see your 12-digit AWS account ID printed.

  • Create the config.py file. Open a new file called config.py in your editor and paste the following:
import os

AWS_REGION = "us-east-1"
S3_BUCKET_NAME = f"bedrock-rag-policies-{os.environ.get('AWS_ACCOUNT_ID', 'REPLACE-WITH-ACCOUNT-ID')}"

EMBEDDING_MODEL_ID = "amazon.titan-embed-text-v2:0"
GENERATION_MODEL_ID = "anthropic.claude-3-haiku-20240307-v1:0"

VALIDATED_PREFIX = "validated/"
QUARANTINED_PREFIX = "quarantined/"

CLOUDWATCH_NAMESPACE = "BedrockRAG/DocumentValidation"

MAX_TOKENS_PER_CHUNK = 8192
MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024
SUPPORTED_EXTENSIONS = {".txt", ".md", ".pdf", ".html", ".doc", ".docx"}

KNOWLEDGE_BASE_NAME = "nexus-corp-policies-kb"
DATA_SOURCE_NAME = "policy-documents"

CHUNK_SIZE = 300
CHUNK_OVERLAP_PERCENTAGE = 20

What does this config do?

This centralizes all project settings in one place. S3_BUCKET_NAME builds a unique bucket name from your account ID. The model IDs reference the exact Bedrock models you enabled. MAX_TOKENS_PER_CHUNK matches Titan Embeddings V2's 8,192 token input limit.

The chunking settings (CHUNK_SIZE of 300 tokens with 20% overlap) control how documents get split for the knowledge base later.

  • Save config.py.
  • Create the S3 bucket for your policy documents by running:
aws s3 mb s3://bedrock-rag-policies-$AWS_ACCOUNT_ID --region us-east-1

You should see a message confirming the bucket was created.

Bucket creation failed?

If you see a BucketAlreadyExists error, the name is globally taken. Confirm your AWS_ACCOUNT_ID env var is set correctly by running echo $AWS_ACCOUNT_ID. It should be a 12-digit number.

help me troubleshoot S3 bucket creation errors

Create sample policy documents

Your RAG system needs documents to retrieve from. You will create three fictional company policy files with specific details that Claude could never know without access to these documents.

  • Create the sample_policies directory by running:
mkdir sample_policies
  • Create the remote work policy file. Open a new file called sample_policies/remote_work_policy.txt in your editor and paste the following:
Nexus Corp Remote Work Policy
Effective Date: January 15, 2025
Policy Number: HR-2025-003

Section 1: Eligibility
All full-time employees who have completed their 90-day probationary period are eligible for remote work arrangements. Contractors and part-time employees must obtain written approval from their department VP.

Section 2: Schedule Requirements
Remote employees must maintain core hours of 10:00 AM to 3:00 PM Eastern Time, Monday through Friday. Employees may flex their remaining hours between 7:00 AM and 7:00 PM ET.

Section 3: Equipment
Nexus Corp provides a one-time home office stipend of $1,500 for ergonomic equipment. Employees must use company-issued laptops with approved VPN software. Personal devices are not permitted for accessing company systems.

Section 4: Communication
Remote employees must be reachable via Slack during core hours with a maximum response time of 30 minutes. Video cameras must be enabled during all team meetings and client calls.

Section 5: In-Office Requirements
All remote employees must be present in-office a minimum of 4 days per quarter for team collaboration events. Travel expenses for mandatory in-office days are reimbursed per the Travel Policy (FIN-2024-012).

Why specific details?

Notice the precise numbers: $1,500 stipend, 30-minute response time, 4 days per quarter. These are details an FM cannot guess correctly. When you ask Claude about them without RAG in the next step, it will invent different numbers.

  • Save remote_work_policy.txt.
  • Create the expense policy file. Open sample_policies/expense_policy.txt and paste:
Nexus Corp Expense Reimbursement Policy
Effective Date: March 1, 2025
Policy Number: FIN-2025-001

Section 1: Approval Thresholds
- Expenses under $100: No pre-approval required, submit receipt within 5 business days
- Expenses $100-$500: Manager approval required before purchase
- Expenses $500-$5,000: Director approval required before purchase
- Expenses over $5,000: VP approval and Finance review required

Section 2: Travel
- Domestic flights: Economy class required for flights under 4 hours. Business class permitted for flights over 4 hours with manager approval.
- Hotels: Maximum nightly rate of $250 in standard markets, $375 in high-cost markets (NYC, SF, DC, Boston, LA).
- Meals: Per diem of $75/day for domestic travel, $100/day for international travel. Alcohol is not reimbursable.
- Mileage: $0.67 per mile for personal vehicle use.

Section 3: Software and Subscriptions
All software purchases must go through IT procurement. Individual subscriptions over $50/month require IT and manager approval. Annual subscriptions over $500 require director approval.

Section 4: Reimbursement Timeline
Approved expenses are reimbursed within 10 business days of receipt submission. Late submissions (over 30 days) require VP exception approval and may be denied.
  • Save expense_policy.txt.
  • Create the PTO policy file. Open sample_policies/pto_policy.txt and paste:
Nexus Corp Paid Time Off Policy
Effective Date: January 1, 2025
Policy Number: HR-2025-001

Section 1: PTO Accrual
- Years 0-2: 15 days per year (accrued at 1.25 days/month)
- Years 3-5: 20 days per year (accrued at 1.67 days/month)
- Years 6-10: 25 days per year (accrued at 2.08 days/month)
- Years 10+: 30 days per year (accrued at 2.5 days/month)

Section 2: Carryover
Employees may carry over a maximum of 5 unused PTO days into the following calendar year. Carried-over days must be used by March 31 or they are forfeited. No payout is provided for forfeited days.

Section 3: Requesting Time Off
PTO requests of 3 or more consecutive days must be submitted at least 2 weeks in advance. Requests during blackout periods (December 20-31, company all-hands week) require director approval.

Section 4: Sick Leave
Sick leave is separate from PTO. Employees receive 10 sick days per year, non-accruing. Sick leave of 3 or more consecutive days requires a doctor's note upon return. Unused sick days do not carry over.

Section 5: Parental Leave
Primary caregivers receive 16 weeks paid parental leave. Secondary caregivers receive 6 weeks paid parental leave. Parental leave must be taken within 12 months of the qualifying event.
  • Save pto_policy.txt.
  • Create a deliberately malformed file for testing the validation pipeline later. Run this command:
printf '\x00\x01\x02\x03\x80\x81\x82 Intentionally malformed binary file for validation pipeline testing.' > sample_policies/bad_document.bin
  • Confirm all your files exist by running:
ls sample_policies/

You should see four files: bad_document.bin, expense_policy.txt, pto_policy.txt, and remote_work_policy.txt.

Why the bad document?

The .bin file has a non-supported extension and contains binary bytes. Your validation pipeline in Step 3 will catch this file and quarantine it, proving that quality gates work before documents reach the knowledge base.

✔️ Awesome, I've got everything!

Great. Double check you have saved all files.

ⓧ I'd like to double check the full code

Your config.py should look exactly like this:

import os

AWS_REGION = "us-east-1"
S3_BUCKET_NAME = f"bedrock-rag-policies-{os.environ.get('AWS_ACCOUNT_ID', 'REPLACE-WITH-ACCOUNT-ID')}"

EMBEDDING_MODEL_ID = "amazon.titan-embed-text-v2:0"
GENERATION_MODEL_ID = "anthropic.claude-3-haiku-20240307-v1:0"

VALIDATED_PREFIX = "validated/"
QUARANTINED_PREFIX = "quarantined/"

CLOUDWATCH_NAMESPACE = "BedrockRAG/DocumentValidation"

MAX_TOKENS_PER_CHUNK = 8192
MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024
SUPPORTED_EXTENSIONS = {".txt", ".md", ".pdf", ".html", ".doc", ".docx"}

KNOWLEDGE_BASE_NAME = "nexus-corp-policies-kb"
DATA_SOURCE_NAME = "policy-documents"

CHUNK_SIZE = 300
CHUNK_OVERLAP_PERCENTAGE = 20

Your requirements.txt should contain:

boto3>=1.35.0

Scroll up and confirm each policy file in sample_policies/ matches the content shown above.

Your environment is ready. You have model access, a configured Python project, and sample documents waiting to be queried. Next up, you will ask Claude questions about these policies without RAG and see how badly it hallucinates.

Query the FM Without RAG

Your environment is ready with Bedrock model access, a Python project, and policy documents waiting in S3. Now it's time to ask Claude a question about those policies.

The catch? You'll ask Claude directly, without giving it access to your documents. The answers it generates will sound confident and professional, but they'll be completely fabricated. This is the hallucination problem that RAG exists to solve.

In this step, get ready to:
  • Write a script that queries Claude about your company policies using the Bedrock Converse API.
  • Run the script and observe Claude inventing plausible but incorrect policy details.
Write the FM query function

The Bedrock Runtime Converse API lets you send a message to a foundation model and receive a response. You'll use it to ask Claude about Nexus Corp's policies as if it were an HR assistant.

  • Create a new file called query_direct.py in your project folder (the same directory as config.py).
  • Add the following code to define the query function:
import boto3
import json
from config import AWS_REGION, GENERATION_MODEL_ID

def query_fm_directly(question: str) -> str:
    # Create a Bedrock Runtime client for model invocation
    client = boto3.client("bedrock-runtime", region_name=AWS_REGION)

    # Send the question to Claude using the Converse API
    response = client.converse(
        modelId=GENERATION_MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [{"text": question}]
            }
        ],
        system=[
            {"text": "You are a helpful HR assistant for Nexus Corp. Answer questions about company policies."}
        ],
        inferenceConfig={
            "maxTokens": 500,
            "temperature": 0.1
        }
    )

    # Extract the text response from the model output
    output_message = response["output"]["message"]
    return output_message["content"][0]["text"]

What does this code do?

  • It imports GENERATION_MODEL_ID from your config.py, which points to Claude 3 Haiku.
  • The converse() method sends your question with a system prompt that tells Claude to act as an HR assistant.
  • Setting temperature to 0.1 makes the model's answers deterministic and confident, which makes the hallucinations even more convincing.
Add the main function with test questions

You'll ask Claude three specific questions whose real answers you already know from your policy documents. This makes it easy to spot where the model fabricates details.

  • Add the following code below the query_fm_directly function in query_direct.py:
def main():
    # Questions with known answers from our policy documents
    questions = [
        "What is Nexus Corp's PTO policy for employees with 4 years of tenure?",
        "What is the expense approval threshold that requires director sign-off?",
        "How many days per quarter must remote employees be in the office?",
    ]

    print("=" * 60)
    print("QUERYING FM DIRECTLY (NO RAG)")
    print("=" * 60)

    for q in questions:
        print(f"\nQuestion: {q}")
        print("-" * 40)
        answer = query_fm_directly(q)
        print(f"Answer: {answer}")
        print()


if __name__ == "__main__":
    main()

What does this code do?

  • The three questions target facts in your actual policy documents: PTO accrual (20 days at 4 years), expense director threshold ($500-$5,000), and in-office days (4 per quarter).
  • The script loops through each question and prints both the question and the model's response for easy comparison.
  • Save query_direct.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before running it.

ⓧ I'd like to double check the full code

import boto3
import json
from config import AWS_REGION, GENERATION_MODEL_ID

def query_fm_directly(question: str) -> str:
    client = boto3.client("bedrock-runtime", region_name=AWS_REGION)

    response = client.converse(
        modelId=GENERATION_MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [{"text": question}]
            }
        ],
        system=[
            {"text": "You are a helpful HR assistant for Nexus Corp. Answer questions about company policies."}
        ],
        inferenceConfig={
            "maxTokens": 500,
            "temperature": 0.1
        }
    )

    output_message = response["output"]["message"]
    return output_message["content"][0]["text"]
def main():
    questions = [
        "What is Nexus Corp's PTO policy for employees with 4 years of tenure?",
        "What is the expense approval threshold that requires director sign-off?",
        "How many days per quarter must remote employees be in the office?",
    ]

    print("=" * 60)
    print("QUERYING FM DIRECTLY (NO RAG)")
    print("=" * 60)

    for q in questions:
        print(f"\nQuestion: {q}")
        print("-" * 40)
        answer = query_fm_directly(q)
        print(f"Answer: {answer}")
        print()


if __name__ == "__main__":
    main()
Run and observe the hallucination

Before you run this, think about what you expect. Claude has never seen your Nexus Corp documents. Do you think it will admit it doesn't know, or will it try to answer anyway?

  • Run the script in your terminal:
python query_direct.py

Claude responds with detailed, authoritative-sounding answers. The problem is they're wrong.

What should I see?

Claude will confidently state specific numbers for PTO days, expense thresholds, and in-office requirements. Compare them against your actual documents:

  • PTO at 4 years tenure: your policy says 20 days/year (Years 3-5 tier). Claude likely states a different number.
  • Director approval threshold: your policy says $500-$5,000. Claude likely invents a different range.
  • In-office requirement: your policy says 4 days per quarter. Claude likely fabricates a different frequency.

This is hallucination in action. The model generates plausible text, but it has no access to your actual policies.

Seeing an AccessDeniedException?

  • Confirm you enabled model access for Claude 3 Haiku in the Bedrock console (us-east-1). Model access can take a few minutes to propagate after approval.
  • Check that your AWS CLI credentials have the bedrock:InvokeModel permission.
  • Verify the model ID in config.py matches what you see on the Bedrock Model access page.

Why is my script failing with an access error?

Why does the model hallucinate here?

Claude was trained on general internet text. It knows what PTO policies, expense rules, and remote work guidelines look like in general. So when you ask about "Nexus Corp," it generates answers that match the pattern of real policies.

The fundamental problem is that the model has no mechanism to say "I don't have this company's documents." It always produces an answer. RAG solves this by retrieving your actual documents and injecting them into the prompt as context.

You've now seen the hallucination problem firsthand. Claude invents confident, specific answers that could mislead any employee who trusts them. In the next step, you'll build a data validation pipeline that ensures your documents are clean and ready for ingestion into a knowledge base that grounds the model in real data.

Build a Data Validation Pipeline

The hallucinated answers from Step 2 prove you need RAG. But before you can ingest documents into a knowledge base, you need to guarantee data quality.

Malformed files, oversized documents that exceed token limits, or unsupported formats will silently degrade retrieval quality. In this step, you'll build a validation pipeline that catches problems before they reach the vector store and reports health metrics to CloudWatch.

In this step, get ready to:
  • Build validation logic that checks format, size, encoding, and token count.
  • Route valid documents to S3 and quarantine failures with rejection metadata.
  • Add CloudWatch custom metrics for pipeline observability.
Build the document validator

Your validation script needs to run every document through a series of quality gates before it reaches the knowledge base. Each gate catches a different category of bad data.

  • Create a new file called validate_documents.py in your project folder by pasting the following imports and helper function:
import boto3
import os
import sys
from pathlib import Path
from datetime import datetime
from config import (
    AWS_REGION,
    S3_BUCKET_NAME,
    VALIDATED_PREFIX,
    QUARANTINED_PREFIX,
    CLOUDWATCH_NAMESPACE,
    MAX_TOKENS_PER_CHUNK,
    MAX_FILE_SIZE_BYTES,
    SUPPORTED_EXTENSIONS,
)


# Rough token estimator: ~4 characters per token
def estimate_token_count(text: str) -> int:
    return len(text) // 4

What does this code do?

The imports pull in all the threshold constants you defined in config.py earlier. The estimate_token_count function uses the common approximation of 4 characters per token to estimate whether a document fits within Titan Embeddings V2's 8,192 token-per-chunk limit.

  • Below estimate_token_count, add the first half of the validation function that checks file extension and size:
def validate_document(file_path: Path) -> dict:
    # Start optimistic - assume the document passes
    result = {
        "file": file_path.name,
        "status": "passed",
        "reason": None,
        "token_estimate": 0,
    }

    # Gate 1: File extension must be in the supported set
    if file_path.suffix.lower() not in SUPPORTED_EXTENSIONS:
        result["status"] = "failed"
        result["reason"] = f"Unsupported file extension: {file_path.suffix}"
        return result

    # Gate 2: File must not be empty or oversized
    file_size = file_path.stat().st_size
    if file_size == 0:
        result["status"] = "failed"
        result["reason"] = "File is empty"
        return result

    if file_size > MAX_FILE_SIZE_BYTES:
        result["status"] = "failed"
        result["reason"] = f"File exceeds {MAX_FILE_SIZE_BYTES} bytes ({file_size} bytes)"
        return result

What does this code do?

Each gate returns early if the document fails. The function starts with an optimistic result (status: passed) and only flips to failed if a check trips.

Gate 1 rejects unsupported file types like .bin or .exe. Gate 2 rejects empty files and anything over 10 MB.

  • Still inside the validate_document function, add the encoding and token count checks below the size gate:
    # Gate 3: Content must be valid UTF-8
    try:
        content = file_path.read_text(encoding="utf-8")
    except (UnicodeDecodeError, ValueError):
        result["status"] = "failed"
        result["reason"] = "File contains invalid UTF-8 encoding"
        return result

    # Gate 4: Estimated tokens must meet minimum threshold
    token_estimate = estimate_token_count(content)
    result["token_estimate"] = token_estimate

    if token_estimate < 10:
        result["status"] = "failed"
        result["reason"] = f"Content too short ({token_estimate} estimated tokens)"
        return result

    return result

What does this code do?

Gate 3 catches binary or corrupted content that would produce garbage embeddings. Gate 4 rejects documents that are too short to be meaningful. If a document passes all four gates, the function returns with status passed and the token estimate populated.

  • Below validate_document, add the S3 upload function that routes documents to the correct prefix:
def upload_to_s3(file_path: Path, prefix: str, reason: str = None):
    # Upload document to validated/ or quarantined/ prefix
    s3 = boto3.client("s3", region_name=AWS_REGION)
    key = f"{prefix}{file_path.name}"

    # Attach metadata for audit trail
    metadata = {"validation-timestamp": datetime.utcnow().isoformat()}
    if reason:
        metadata["rejection-reason"] = reason[:256]

    s3.upload_file(
        str(file_path),
        S3_BUCKET_NAME,
        key,
        ExtraArgs={"Metadata": metadata},
    )
    print(f"  Uploaded to s3://{S3_BUCKET_NAME}/{key}")

What does this code do?

Valid documents go to the validated/ prefix and quarantined ones to quarantined/. Every upload gets a validation-timestamp in its S3 metadata. Rejected documents also carry a rejection-reason so you can audit failures without re-running the pipeline.

  • Save validate_documents.py.
  • Confirm the file loads without syntax errors by running:
python -c "from validate_documents import validate_document, upload_to_s3; print('Functions loaded OK')"

You should see Functions loaded OK printed. This confirms your validation and upload functions are syntactically correct.

Seeing an ImportError or SyntaxError?

  • Make sure your virtual environment is activated. The config import relies on config.py being in the same directory.
  • Check that the indentation of the validate_document continuation (Gate 3 and 4) is at the same level as Gate 1 and 2. All gates should be indented one level inside the function.
  • help me debug this import error in validate_documents.py
Add CloudWatch observability

Validation results should be visible beyond the terminal. Publishing metrics to CloudWatch lets you track document quality over time and set alarms if failure rates spike.

  • In validate_documents.py, add the publish_metrics function below upload_to_s3:
def publish_metrics(passed: int, failed: int, total: int, avg_tokens: float):
    # Send validation summary to CloudWatch
    cloudwatch = boto3.client("cloudwatch", region_name=AWS_REGION)

    metric_data = [
        {
            "MetricName": "DocumentsValidated",
            "Value": total,
            "Unit": "Count",
        },
        {
            "MetricName": "DocumentsPassed",
            "Value": passed,
            "Unit": "Count",
        },
        {
            "MetricName": "DocumentsFailed",
            "Value": failed,
            "Unit": "Count",
        },
        {
            "MetricName": "AverageTokenCount",
            "Value": avg_tokens,
            "Unit": "Count",
        },
    ]

    cloudwatch.put_metric_data(
        Namespace=CLOUDWATCH_NAMESPACE,
        MetricData=metric_data,
    )
    print(f"\nPublished metrics to CloudWatch namespace: {CLOUDWATCH_NAMESPACE}")

What does this code do?

The put_metric_data call publishes four custom metrics to the BedrockRAG/DocumentValidation namespace: total documents processed, pass count, fail count, and average token length.

CloudWatch custom metrics are free for the first 10 metrics per namespace under the AWS free tier. These four metrics give you a production-quality health dashboard at zero cost for this project.

  • Save validate_documents.py.
  • Confirm the new function loads by running:
python -c "from validate_documents import publish_metrics; print('Metrics function loaded OK')"

You should see Metrics function loaded OK printed.

Test with intentional bad data

Your sample_policies/ folder already contains bad_document.bin. This file has a .bin extension that is not in the supported set. The validation pipeline should catch it and quarantine it while passing your three policy documents.

  • Add the main function that orchestrates the full pipeline. Paste this below publish_metrics:
def main():
    # Locate and verify the documents directory
    docs_dir = Path("sample_policies")
    if not docs_dir.exists():
        print(f"Error: Directory '{docs_dir}' not found.")
        sys.exit(1)

    files = list(docs_dir.iterdir())
    if not files:
        print("No files found in sample_policies/")
        sys.exit(1)

    print("=" * 60)
    print("DOCUMENT VALIDATION PIPELINE")
    print("=" * 60)

    # Validate each file and route to S3
    results = []
    for file_path in files:
        if file_path.is_file():
            result = validate_document(file_path)
            results.append(result)

            status_icon = "PASS" if result["status"] == "passed" else "FAIL"
            print(f"\n[{status_icon}] {result['file']}")

            if result["status"] == "passed":
                print(f"  Estimated tokens: {result['token_estimate']}")
                upload_to_s3(file_path, VALIDATED_PREFIX)
            else:
                print(f"  Reason: {result['reason']}")
                upload_to_s3(file_path, QUARANTINED_PREFIX, result["reason"])

What does this code do?

The main function iterates through every file in sample_policies/. For each file, it runs the validation checks and routes the result: passed documents go to the validated/ prefix, failed ones go to quarantined/ with a rejection reason.

  • Still inside main(), add the summary statistics and entry point below the loop:
    # Calculate and display summary
    passed = sum(1 for r in results if r["status"] == "passed")
    failed = sum(1 for r in results if r["status"] == "failed")
    total = len(results)
    token_counts = [r["token_estimate"] for r in results if r["token_estimate"] > 0]
    avg_tokens = sum(token_counts) / len(token_counts) if token_counts else 0

    print("\n" + "=" * 60)
    print(f"RESULTS: {passed} passed, {failed} failed, {total} total")
    print(f"Average token count: {avg_tokens:.0f}")
    print("=" * 60)

    publish_metrics(passed, failed, total, avg_tokens)


if __name__ == "__main__":
    main()

What does this code do?

The summary block tallies pass/fail counts and computes the average token length across valid documents. It prints a human-readable report, then publishes the same numbers to CloudWatch for long-term tracking.

  • Save validate_documents.py.

Before you run the pipeline, what do you think will happen to bad_document.bin? Will it pass or fail, and at which gate?

  • Run the validation pipeline:
python validate_documents.py

You should see output showing 3 documents passed and 1 failed. The three policy files pass all gates and upload to the validated/ prefix. The bad_document.bin fails at Gate 1 with the reason "Unsupported file extension: .bin" and routes to quarantined/.

What should I see?

The final summary line reads RESULTS: 3 passed, 1 failed, 4 total. You also see a confirmation that metrics were published to the BedrockRAG/DocumentValidation namespace.

  • Confirm the CloudWatch metrics arrived by running:
aws cloudwatch list-metrics --namespace "BedrockRAG/DocumentValidation" --region us-east-1

Within 1-2 minutes, you should see your four metrics listed: DocumentsValidated, DocumentsPassed, DocumentsFailed, and AverageTokenCount.

Metrics not appearing yet?

New CloudWatch metrics can take up to 15 minutes to appear in list-metrics. If you see an empty result, wait a few minutes and try again. The metrics are published as soon as the script completes.

If you see an access denied error, confirm your AWS CLI credentials have the cloudwatch:PutMetricData and cloudwatch:ListMetrics permissions.

help me troubleshoot why my CloudWatch metrics aren't showing up

✔️ Awesome, I've got everything!

Great. Double check you've saved validate_documents.py and that your terminal shows 3 passed and 1 failed.

ⓧ I'd like to double check the full code

Your validate_documents.py is built from these chunks (scroll up to confirm each is present, in order):

  • Imports and estimate_token_count function.
  • validate_document function (Gates 1-2: extension and size checks).
  • validate_document function continued (Gates 3-4: encoding and token count).
  • upload_to_s3 function.
  • publish_metrics function.
  • main function (file loop and routing).
  • main function continued (summary and entry point).

Your data quality gates are in place. Every document that reaches the knowledge base has passed format, encoding, size, and token checks. Next up, you'll create the Managed Knowledge Base and ingest these validated documents to enable semantic retrieval.

Create Your Knowledge Base and Ingest Data

Your validation pipeline proved its value. Clean documents are in S3, bad ones are quarantined, and CloudWatch tracks the pass/fail rate.

Now you need a way to search those documents semantically. A Bedrock Managed Knowledge Base converts your text into vector embeddings, stores them, and handles retrieval automatically. You point it at your S3 bucket and it takes care of the rest.

In this step, get ready to:
  • Create a Managed Knowledge Base and connect it to your validated documents in S3.
  • Trigger an ingestion job that chunks and embeds your policy documents.
  • Test retrieval with the Retrieve API to confirm relevant chunks are returned with source attribution.
Create the Managed Knowledge Base

The Managed Knowledge Base handles vector storage, embedding generation, and retrieval for you. You set the type to MANAGED and Bedrock provisions everything behind the scenes. No OpenSearch cluster, no index mapping, no infrastructure to manage.

  • Create a new file called create_kb.py in your project folder.
  • Add the imports and the knowledge base creation function by pasting this code:
import boto3
import time
import json
from config import (
    AWS_REGION,
    S3_BUCKET_NAME,
    VALIDATED_PREFIX,
    KNOWLEDGE_BASE_NAME,
    DATA_SOURCE_NAME,
)


def create_managed_knowledge_base():
    client = boto3.client("bedrock-agent", region_name=AWS_REGION)

    print("Creating Managed Knowledge Base...")
    response = client.create_knowledge_base(
        name=KNOWLEDGE_BASE_NAME,
        description="Nexus Corp policy documents for RAG Q&A",
        knowledgeBaseConfiguration={
            "type": "MANAGED",
        },
    )

    kb_id = response["knowledgeBase"]["knowledgeBaseId"]
    print(f"Knowledge Base created: {kb_id}")
    print(f"Status: {response['knowledgeBase']['status']}")

    return kb_id

What does this code do?

The imports pull configuration values from your config.py so the knowledge base name, S3 bucket, and region stay consistent across scripts.

The create_managed_knowledge_base() function calls the Bedrock Agent API with type: "MANAGED". This tells Bedrock to handle the vector store, embedding model, and retrieval infrastructure. No additional storage configuration is needed.

  • Below the create_managed_knowledge_base function, add the data source function that connects your S3 bucket to the knowledge base:
def add_s3_data_source(kb_id: str):
    client = boto3.client("bedrock-agent", region_name=AWS_REGION)

    print(f"\nAdding S3 data source to KB {kb_id}...")
    response = client.create_data_source(
        knowledgeBaseId=kb_id,
        name=DATA_SOURCE_NAME,
        description="Validated policy documents from S3",
        dataSourceConfiguration={
            "type": "S3",
            "s3Configuration": {
                "bucketArn": f"arn:aws:s3:::{S3_BUCKET_NAME}",
                "inclusionPrefixes": [VALIDATED_PREFIX],
            },
        },
        vectorIngestionConfiguration={
            "chunkingConfiguration": {
                "chunkingStrategy": "FIXED_SIZE",
                "fixedSizeChunkingConfiguration": {
                    "maxTokens": 300,
                    "overlapPercentage": 20,
                },
            }
        },
    )

    ds_id = response["dataSource"]["dataSourceId"]
    print(f"Data source created: {ds_id}")
    return ds_id

What does the chunking configuration do?

The FIXED_SIZE chunking strategy splits each document into 300-token chunks with 20% overlap. Overlap ensures that sentences at chunk boundaries still appear in adjacent chunks, so retrieval does not miss context that spans a boundary.

The inclusionPrefixes restricts ingestion to only the validated/ prefix. Your quarantined documents are never touched.

  • Save create_kb.py.
  • Confirm the file is syntactically valid by running:
python -c "import create_kb; print('Module loaded successfully')"

You should see Module loaded successfully printed with no errors.

Seeing an ImportError?

Make sure your virtual environment is activated and that config.py is in the same directory as create_kb.py. If you see a SyntaxError, check for missing commas or mismatched brackets in the code you just pasted.

Help me debug this import error in create_kb.py

Sync and verify ingestion

With the knowledge base and data source defined, you need to trigger an ingestion job. This tells Bedrock to read your S3 documents, chunk them, generate embeddings, and store the vectors.

  • Add the ingestion function below add_s3_data_source in create_kb.py:
def start_ingestion(kb_id: str, ds_id: str):
    client = boto3.client("bedrock-agent", region_name=AWS_REGION)

    print(f"\nStarting ingestion job...")
    response = client.start_ingestion_job(
        knowledgeBaseId=kb_id,
        dataSourceId=ds_id,
    )

    job_id = response["ingestionJob"]["ingestionJobId"]
    print(f"Ingestion job started: {job_id}")

    while True:
        job_response = client.get_ingestion_job(
            knowledgeBaseId=kb_id,
            dataSourceId=ds_id,
            ingestionJobId=job_id,
        )
        status = job_response["ingestionJob"]["status"]
        print(f"  Status: {status}")

        if status in ("COMPLETE", "FAILED", "STOPPED"):
            break
        time.sleep(10)

    if status == "COMPLETE":
        stats = job_response["ingestionJob"].get("statistics", {})
        print(f"\nIngestion complete!")
        print(f"  Documents scanned: {stats.get('numberOfDocumentsScanned', 'N/A')}")
        print(f"  Documents indexed: {stats.get('numberOfNewDocumentsIndexed', 'N/A')}")
    else:
        print(f"\nIngestion ended with status: {status}")
        failure_reasons = job_response["ingestionJob"].get("failureReasons", [])
        for reason in failure_reasons:
            print(f"  Failure: {reason}")

    return job_id

What does this code do?

The start_ingestion_job call kicks off the embedding process. The while True loop polls every 10 seconds until the job finishes with COMPLETE, FAILED, or STOPPED.

On success, it prints the number of documents scanned and indexed. These counts confirm your 3 validated policy files made it into the vector store.

Test retrieval

The final piece confirms that the knowledge base can actually return relevant chunks when you ask a question. The Retrieve API performs a vector similarity search and returns the closest matching text with source attribution.

  • Add the test retrieval function below start_ingestion in create_kb.py:
def test_retrieval(kb_id: str):
    runtime_client = boto3.client("bedrock-agent-runtime", region_name=AWS_REGION)

    test_query = "What is the PTO accrual rate for employees with 4 years of tenure?"
    print(f"\nTesting retrieval with query: '{test_query}'")

    response = runtime_client.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": test_query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": 3,
            }
        },
    )

    print(f"\nRetrieved {len(response['retrievalResults'])} results:")
    for i, result in enumerate(response["retrievalResults"], 1):
        content = result["content"]["text"][:200]
        score = result.get("score", "N/A")
        print(f"\n  Result {i} (score: {score}):")
        print(f"  {content}...")

What does the Retrieve API return?

The retrieve call uses the bedrock-agent-runtime client (not the build-time bedrock-agent client). It returns up to 3 results, each containing the chunk text, a relevance score, and the S3 source location. This source attribution is what makes RAG answers traceable.

  • Add the main function that ties everything together:
def main():
    print("=" * 60)
    print("CREATING BEDROCK MANAGED KNOWLEDGE BASE")
    print("=" * 60)

    kb_id = create_managed_knowledge_base()
    ds_id = add_s3_data_source(kb_id)
    start_ingestion(kb_id, ds_id)
    test_retrieval(kb_id)

    print(f"\n\nSave these IDs for the next step:")
    print(f"  KNOWLEDGE_BASE_ID = \"{kb_id}\"")
    print(f"  DATA_SOURCE_ID = \"{ds_id}\"")


if __name__ == "__main__":
    main()

What does main do?

The main() function runs the full pipeline in order: create the knowledge base, attach the S3 data source, ingest documents, and test retrieval. At the end it prints the Knowledge Base ID and Data Source ID that you need for the next step.

  • Save create_kb.py.

Before you run this, what do you think will happen? The script creates a knowledge base, adds your S3 bucket as a data source, ingests 3 documents, and runs a test query. Do you expect the PTO query to return relevant chunks from your policy documents?

  • Run the full knowledge base creation pipeline:
python create_kb.py

The script will take 30-60 seconds as it waits for the ingestion job to complete. You should see output showing the knowledge base creation, data source attachment, ingestion polling, and finally 3 retrieval results with PTO-related text.

What should I see?

  • A Knowledge Base ID printed after creation.
  • A Data Source ID printed after the S3 connection.
  • Ingestion status cycling through IN_PROGRESS then COMPLETE.
  • Documents scanned: 3 and Documents indexed: 3.
  • 3 retrieval results containing PTO accrual text from your policy document.
  • Copy the KNOWLEDGE_BASE_ID value printed at the end. You will need it in the next step.

Ingestion ended with status FAILED?

Check that your S3 bucket name matches what is in config.py. Confirm your AWS_ACCOUNT_ID environment variable is set, or that you replaced the placeholder in S3_BUCKET_NAME.

Also verify your validated documents are in the validated/ prefix by running aws s3 ls s3://your-bucket-name/validated/ and confirming the 3 policy files appear.

Help me debug my Bedrock Knowledge Base ingestion failure

  • Verify in the Bedrock console by navigating to Amazon Bedrock in the AWS Console.
  • Select Knowledge bases in the left sidebar.
  • Confirm that nexus-corp-policies-kb shows status Available.
  • Click into the knowledge base and check the data source sync shows Completed with 3 documents.

✔️ Awesome, I've got everything!

Great. Double check you have saved create_kb.py and noted your Knowledge Base ID for the next step.

ⓧ I'd like to double check the full code

Your create_kb.py contains 4 functions and a main block. Scroll up and confirm these are present in this order:

  • Imports from config at the top.
  • create_managed_knowledge_base() function.
  • add_s3_data_source() function.
  • start_ingestion() function.
  • test_retrieval() function.
  • main() function with the if __name__ == "__main__" block.

Your knowledge base is live and returning relevant policy chunks with source attribution. Next, you will wire this retrieval into a full RAG pipeline with governed prompt templates.

Query with RAG and Prompt Management

You now have a working knowledge base with your policy documents indexed and retrievable. The next challenge is controlling HOW the FM uses that retrieved context when generating answers.

Without governance, the prompt that shapes every answer lives hardcoded in a script. In production, prompts drift, get edited without review, and break silently. Bedrock Prompt Management solves this with versioned, auditable templates that you can roll back if something goes wrong.

In this step, get ready to:
  • Create a governed prompt template in Bedrock Prompt Management with context and question variables.
  • Build the full RAG pipeline that retrieves chunks, injects context, and invokes via the governed prompt.
  • Compare RAG-grounded answers against the hallucinated responses from Step 2.
Create a governed prompt template

Bedrock Prompt Management stores your prompts centrally with named variants, inference config, and template variables. You define the template once, version it, and every invocation uses the governed ARN instead of a raw string in your code.

  • Create a new file called query_rag.py in your project folder.
  • Add the imports, knowledge base ID constant, and the prompt template definition by pasting this code:
import boto3
import json
from config import AWS_REGION, GENERATION_MODEL_ID

# Replace with the KB ID printed by create_kb.py
KNOWLEDGE_BASE_ID = "REPLACE-WITH-YOUR-KB-ID"


def create_qa_prompt():
    # Register a governed prompt in Bedrock Prompt Management
    client = boto3.client("bedrock-agent", region_name=AWS_REGION)

    # Template with variables for retrieved context and the user question
    prompt_template = (
        "You are a helpful HR assistant for Nexus Corp. "
        "Answer the employee's question using ONLY the context provided below. "
        "If the answer is not found in the context, say 'I cannot find this information in the policy documents.' "
        "Always cite the specific policy section that supports your answer.\n\n"
        "Context:\n{{context}}\n\n"
        "Question: {{question}}\n\n"
        "Answer:"
    )

What does this code do?

The KNOWLEDGE_BASE_ID constant is where you paste the ID that create_kb.py printed at the end of the previous step.

The prompt_template string uses {{context}} and {{question}} as Bedrock Prompt Management variables. At invocation time, you supply values for these and the managed template fills them in.

  • Replace REPLACE-WITH-YOUR-KB-ID with the actual Knowledge Base ID from the previous step.
  • Now add the create_prompt API call and versioning logic directly below the prompt_template definition, still inside the create_qa_prompt function:
    # Register the prompt with Bedrock Prompt Management
    response = client.create_prompt(
        name="NexusCorpPolicyQA",
        description="Governed Q&A prompt for policy document retrieval",
        defaultVariant="production",
        variants=[
            {
                "name": "production",
                "modelId": GENERATION_MODEL_ID,
                "templateType": "TEXT",
                "inferenceConfiguration": {
                    "text": {
                        "maxTokens": 500,
                        "temperature": 0.1,
                    }
                },
                "templateConfiguration": {
                    "text": {
                        "text": prompt_template,
                    }
                },
            }
        ],
    )

    prompt_id = response["id"]
    print(f"Created prompt: {response['name']} (ID: {prompt_id})")

    # Freeze this configuration as an immutable version
    version_response = client.create_prompt_version(promptIdentifier=prompt_id)
    prompt_version_arn = version_response["arn"]
    prompt_version = version_response["version"]
    print(f"Created version: {prompt_version} (ARN: {prompt_version_arn})")

    return prompt_id, prompt_version_arn

What does this code do?

The create_prompt call registers your template with Bedrock. It defines a single variant called production that pins Claude 3 Haiku with low temperature for consistent answers.

The create_prompt_version call freezes the current DRAFT into an immutable numbered version. The returned ARN is what you pass to the Converse API at invocation time, so every call uses the exact same governed template.

  • Save query_rag.py.

Seeing a syntax error?

Make sure the second code block is indented inside the create_qa_prompt function body (4 spaces in from the def line). The response = client.create_prompt( line should be at the same indent level as the prompt_template variable above it.

help me fix indentation errors in my query_rag.py file

Build the retrieval and invocation pipeline

With the prompt registered, you need two more functions. One retrieves relevant document chunks from the knowledge base, and the other injects those chunks into the governed template and invokes the FM.

  • Add the retrieve_context function below create_qa_prompt in query_rag.py:
def retrieve_context(question: str) -> str:
    # Fetch the top 5 most relevant chunks from the knowledge base
    runtime_client = boto3.client("bedrock-agent-runtime", region_name=AWS_REGION)

    response = runtime_client.retrieve(
        knowledgeBaseId=KNOWLEDGE_BASE_ID,
        retrievalQuery={"text": question},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": 5,
            }
        },
    )

    # Format each chunk with its S3 source for citation
    context_parts = []
    for result in response["retrievalResults"]:
        text = result["content"]["text"]
        source = result.get("location", {}).get("s3Location", {}).get("uri", "Unknown")
        context_parts.append(f"[Source: {source}]\n{text}")

    return "\n\n---\n\n".join(context_parts)

What does this code do?

The Retrieve API performs a semantic search against your knowledge base's vector index. It returns the 5 chunks most similar to the question, along with their S3 source locations.

Each chunk is tagged with its source URI so the FM can cite which policy document it draws from. The chunks are joined with separators into a single context string.

  • Add the query_with_rag function below retrieve_context:
def query_with_rag(question: str, prompt_version_arn: str) -> str:
    # Invoke the governed prompt with retrieved context
    runtime_client = boto3.client("bedrock-runtime", region_name=AWS_REGION)

    # Retrieve relevant document chunks
    context = retrieve_context(question)

    # Pass the prompt version ARN as modelId and supply template variables
    response = runtime_client.converse(
        modelId=prompt_version_arn,
        promptVariables={
            "context": {"text": context},
            "question": {"text": question},
        },
    )

    return response["output"]["message"]["content"][0]["text"]

Why pass the ARN as modelId?

When you invoke a managed prompt through the Converse API, you pass the prompt version ARN in the modelId field instead of a model ID. Bedrock resolves the ARN to your versioned template, fills in the promptVariables, and routes the request to the model defined in the variant.

This means your application code never contains the raw prompt text. If you need to tweak wording, you create a new version in Prompt Management and point your code at the new ARN.

  • Save query_rag.py.

Not sure about the function placement?

Both retrieve_context and query_with_rag should be module-level functions (no indentation), placed after create_qa_prompt but before the main function you will add next.

help me check the structure of my query_rag.py file

Compare results against Step 2

The final piece is the main function that ties everything together. It creates the governed prompt, then asks the same three questions that Claude hallucinated answers for in Step 2.

  • Add the main function and entry point at the bottom of query_rag.py:
def main():
    print("=" * 60)
    print("RAG QUERY WITH PROMPT MANAGEMENT")
    print("=" * 60)

    # Create and version the governed prompt
    prompt_id, prompt_version_arn = create_qa_prompt()

    # Same questions from Step 2 that produced hallucinated answers
    questions = [
        "What is Nexus Corp's PTO policy for employees with 4 years of tenure?",
        "What is the expense approval threshold that requires director sign-off?",
        "How many days per quarter must remote employees be in the office?",
    ]

    print("\n" + "=" * 60)
    print("QUERYING WITH RAG (GROUNDED ANSWERS)")
    print("=" * 60)

    for q in questions:
        print(f"\nQuestion: {q}")
        print("-" * 40)
        answer = query_with_rag(q, prompt_version_arn)
        print(f"Answer: {answer}")
        print(f"[Using governed prompt version: {prompt_version_arn}]")
        print()

    print("\nCleanup: To delete the prompt, run:")
    print(f"  client.delete_prompt(promptIdentifier='{prompt_id}')")


if __name__ == "__main__":
    main()

What does this code do?

The main function first calls create_qa_prompt() to register and version the prompt template. Then it loops through the exact same questions you tested in Step 2 (where Claude hallucinated wrong policy details).

For each question, query_with_rag retrieves relevant chunks from the knowledge base and invokes the FM through the governed prompt. The ARN is printed after each answer to confirm the governed template was used.

  • Save query_rag.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved query_rag.py with all five sections: imports, constant, create_qa_prompt, retrieve_context, query_with_rag, and main.

ⓧ I'd like to double check the full code

Scroll up and confirm your query_rag.py contains each of these sections in order:

  • Imports (boto3, json, config imports) and the KNOWLEDGE_BASE_ID constant with your actual KB ID.
  • The create_qa_prompt() function defining the template and calling create_prompt then create_prompt_version.
  • The retrieve_context() function that calls the Retrieve API with numberOfResults: 5.
  • The query_with_rag() function that calls converse with promptVariables.
  • The main() function and the if __name__ == "__main__" entry point.

Before you run this, think about what you expect to see. In Step 2, Claude confidently stated wrong PTO days and wrong expense thresholds. Will the answers be different this time?

  • Run the RAG query script:
python query_rag.py

You should see three accurate, cited answers that match your actual policy documents. For example, the PTO answer now correctly states 20 days per year for employees with 3-5 years of tenure, citing Section 1 of the PTO policy. The expense answer correctly reports $500-$5,000 as the director approval threshold. Each answer ends with a line showing the governed prompt version ARN.

What changed compared to Step 2?

In Step 2, Claude had no access to your documents and invented plausible-sounding policies. Now, the Retrieve API fetches actual policy text from the knowledge base, injects it as {{context}} in the governed template, and the model answers strictly from that context.

The prompt template also instructs the model to say "I cannot find this information" if the context does not contain the answer. This prevents hallucination even when a question falls outside the indexed documents.

Getting an AccessDeniedException?

Make sure your KNOWLEDGE_BASE_ID matches the ID printed by create_kb.py in the previous step. Also confirm that Claude 3 Haiku model access is still enabled in the Bedrock console under Model access.

help me debug an AccessDeniedException when running query_rag.py

Your RAG pipeline is complete. You went from hallucinated guesses to grounded, cited answers using a governed prompt template. Next up, the Secret Mission challenges you to extend the pipeline with multimodal document processing.

Secret mission

Multimodal Document Processing

Your pipeline currently rejects image-based documents like scanned PDFs and policy infographics. Real enterprise document sets are full of them. Can you extend your validation pipeline to extract text from images using Bedrock's multimodal capabilities, then ingest that text into your knowledge base?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses on-demand Bedrock API calls and a Managed Knowledge Base that charges $5/GB/month of raw data.

For the few kilobytes of policy documents you uploaded, the monthly Knowledge Base cost rounds to effectively $0.00.

Cost warning

The Bedrock Managed Knowledge Base charges $5.00 per GB of raw data per month. For this project's tiny document set (a few KB), this is negligible.

Bedrock API calls are billed on-demand per invocation. If you re-run scripts frequently, token charges will accumulate. Delete the knowledge base when you are done to avoid any ongoing charges.

Resources you used:

  • S3 bucket (bedrock-rag-policies-{AWS_ACCOUNT_ID}) with policy documents in validated/ and quarantined/ prefixes.
  • Bedrock Managed Knowledge Base (nexus-corp-policies-kb) with data source and ingested documents.
  • Bedrock Prompt Management prompt (NexusCorpPolicyQA) and its versioned deployment.
  • Local project files on your Desktop.

Keep everything running

No action needed. Choose this if you plan to continue building on this project or revisiting the knowledge base for study.

  • The Managed KB charges $5/GB/month. For a few KB of documents, this rounds to $0.00 on your bill.
  • CloudWatch custom metrics in the BedrockRAG/DocumentValidation namespace are covered by the free tier for 10 custom metrics.
  • S3 storage for a handful of small text files is negligible.
  • The Bedrock prompt and version have no storage charges.
  • Your local project files remain on your Desktop.

Pause - I'll come back to this later

Free up local resources while keeping your AWS infrastructure intact. The Managed KB is pay-per-use for retrieval calls, so it costs nothing when idle.

  • Deactivate your Python virtual environment by running:
deactivate
  • Close any terminal sessions running project scripts.
  • Your knowledge base, S3 bucket, prompt, and CloudWatch metrics remain in place at negligible cost.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

  • Delete the Bedrock Knowledge Base by running this command (replace the knowledge base ID with yours):
aws bedrock-agent delete-knowledge-base --knowledge-base-id YOUR_KB_ID --region us-east-1
  • Delete the Bedrock prompt by running this command (replace the prompt identifier with yours):
aws bedrock-agent delete-prompt --prompt-identifier YOUR_PROMPT_ID --region us-east-1
  • Empty and delete the S3 bucket by running:
aws s3 rb s3://bedrock-rag-policies-[[AWS_ACCOUNT_ID="your AWS account ID"]] --force

What about the CloudWatch metrics?

CloudWatch custom metrics cannot be deleted manually. They automatically expire after 15 months if no new data is published to them.

Since you are no longer running the validation script, no new data will be published and the metrics will age out on their own at no cost.

  • Delete your local project files. On macOS or Linux, run:
cd ~/Desktop
rm -rf bedrock-rag-policies
  • On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force bedrock-rag-policies

Well Done!

Well Done!

Nice work. You just built a production-pattern RAG system that takes custom documents from validation through vector storage to grounded retrieval with source citations.

You've learned how to:

  • Built a complete RAG pipeline that solves the FM hallucination problem by grounding answers in your actual policy documents via Amazon Bedrock Knowledge Bases with chunking configuration and managed embeddings.
  • Created a data validation pipeline with CloudWatch observability that catches malformed, oversized, or unsupported documents before they corrupt your knowledge base.
  • Implemented prompt governance with Bedrock Prompt Management, deploying versioned, auditable templates that make prompt changes trackable and rollback-safe.
  • Secret Mission: Extended the pipeline with multimodal document processing to extract text from image-based policy documents using the Bedrock Converse API.

Ready to quiz yourself?