Implementing RAG with Terraform using AWS S3, Bedrock KnowledgeBase, OpenSearch Serverless, IAM

Published @ AWS Builder Center

In this post, we will implement a Retrieval-Augmented Generation (RAG) system on AWS using Terraform, Amazon S3, Amazon Bedrock Knowledge Bases (KB), and Amazon OpenSearch Serverless.

We will walk through how documents move from S3 through the RAG ingestion pipeline, how embeddings are generated and stored, and finally how user queries retrieve relevant information to generate grounded answers.

What is RAG?

Retrieval-Augmented Generation (RAG) is a pattern that allows an LLM to answer questions using information from an external knowledge source instead of relying only on the knowledge learned during model training.

In a typical RAG system, the process can be divided into two main phases: ingestion and querying/retrieval.

During ingestion, documents are loaded into a document store, split into smaller pieces called chunks, and converted into numerical representations called embeddings using an embedding model (Titan Text Embeddings v2). These embeddings, together with the original text and metadata, are then stored in a vector database, allowing the system to efficiently search for semantically relevant information later.

Document Ingestion

- Upload documents → Amazon S3
- Split documents → Chunks
- Generate embeddings → Embedding Model
- Store embeddings + text + metadata → Vector Database

Query & Retrieval

- User submits a question
- Convert question → Query Embedding
- Search vector database → Similarity Search
- Retrieve the most relevant → Document Chunks
- Send question + retrieved context (retrieval-augmented) → LLM
- Generate the final answer + citations

In this small project, Amazon S3 acts as the document source, while Amazon Bedrock Knowledge Bases (KB) manages the ingestion pipeline. When an ingestion job is triggered;

• User uploads a document through the Streamlit UI.
• The app writes it straight to the S3 docs bucket.
• User clicks Sync Knowledge Base in the UI.
• Bedrock KB reads documents from S3 (StartIngestionJob),
• Bedrock KB chunks the content,
• Bedrock KB generates embeddings using the configured embedding model (Titan Text Embeddings v2),
• and stores the resulting vectors and document information in Amazon OpenSearch Serverless for vector search.

At query/retrieval time;

• User types a question/query into the Streamlit chat input.
• Bedrock KB converts the user's question into an embedding,
• Bedrock KB retrieves the most relevant chunks from OpenSearch Serverless vector database,
• and uses the retrieved context with the configured generation model to produce the final answer and source citations.

If you have read this far, we can go deeper 😊

Vector Database Configuration

For the vector database, we configure Amazon OpenSearch Serverless with a knn_vector field to store document embeddings and enable KNN (k-Nearest Neighbors) similarity search.

The index uses HNSW (Hierarchical Navigable Small World) as the approximate nearest-neighbor algorithm, with FAISS as the underlying vector search engine, allowing efficient similarity search across a large number of embeddings.

We also keep the original text and metadata in the index, which allows retrieved chunks to be returned with their source information and enables metadata filtering when more precise retrieval is required.

In addition to vector search, OpenSearch supports traditional lexical search approaches such as BM25 and keyword search, which can be useful when exact terms, names, identifiers, or keywords are more important than semantic similarity.

For vector similarity, cosine similarity measures the angle between two embedding vectors rather than their magnitude:

• vectors pointing in similar directions have a score closer to 1,
• while unrelated or opposite directions produce lower scores.

In our configuration, the OpenSearch index uses knn_vector, HNSW, FAISS, and L2 distance for the vector search, alongside text and metadata fields.

Code GitHub Link: Project on GitHub

Whether you're exploring to build your own system, this will give you a clear, practical starting point 😉

Table of Contents

• Terraform: Bedrock KnowledgeBase, OpenSearch Serverless, S3, IAM
• Streamlit App: Digesting & Querying
• All Code & Demo
• Conclusion
• References

---

Terraform: Bedrock KnowledgeBase, OpenSearch Serverless, S3, IAM

Terraform providers and version:

terraform {
  required_version = ">= 1.5.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    opensearch = {
      source  = "opensearch-project/opensearch"
      version = "~> 2.3"
    }
    time = {
      source  = "hashicorp/time"
      version = "~> 0.11"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

data "aws_caller_identity" "current" {}

IAM Assume Role, Permission Policy for S3 Bucket Read, Invoking Embedding Model, OpenSearch Serverless Access:

data "aws_iam_policy_document" "bedrock_kb_trust" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["bedrock.amazonaws.com"]
    }
    condition {
      test     = "StringEquals"
      variable = "aws:SourceAccount"
      values   = [data.aws_caller_identity.current.account_id]
    }
  }
}

resource "aws_iam_role" "bedrock_kb" {
  name               = "${var.project_name}-kb-role"
  assume_role_policy = data.aws_iam_policy_document.bedrock_kb_trust.json
}

data "aws_iam_policy_document" "bedrock_kb_permissions" {
  statement {
    sid       = "S3Read"
    actions   = ["s3:GetObject", "s3:ListBucket"]
    resources = [aws_s3_bucket.docs.arn, "${aws_s3_bucket.docs.arn}/*"]
  }

  statement {
    sid       = "InvokeEmbeddingModel"
    actions   = ["bedrock:InvokeModel"]
    resources = ["arn:aws:bedrock:${var.aws_region}::foundation-model/${var.embedding_model_id}"]
  }

  statement {
    sid       = "OpenSearchServerlessAccess"
    actions   = ["aoss:APIAccessAll"]
    resources = [aws_opensearchserverless_collection.kb.arn]
  }
}

resource "aws_iam_role_policy" "bedrock_kb" {
  name   = "${var.project_name}-kb-policy"
  role   = aws_iam_role.bedrock_kb.id
  policy = data.aws_iam_policy_document.bedrock_kb_permissions.json
}

S3 Bucket, Bucket Policy:

resource "aws_s3_bucket" "docs" {
  bucket = var.docs_bucket_name != null ? var.docs_bucket_name : "${var.project_name}-docs-${data.aws_caller_identity.current.account_id}"
}

resource "aws_s3_bucket_versioning" "docs" {
  bucket = aws_s3_bucket.docs.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_public_access_block" "docs" {
  bucket                  = aws_s3_bucket.docs.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

Bedrock KnowledgeBase for Vector Database Config (embedding model, embedding dimensions, opensearch serverless collection, chunking), Data Source (S3 Bucket, Vector Ingestion, Chunking Strategy):

locals {
  generation_model_arn = "arn:aws:bedrock:${var.aws_region}:${data.aws_caller_identity.current.account_id}:inference-profile/${var.generation_model_id}"
  embedding_model_arn  = "arn:aws:bedrock:${var.aws_region}::foundation-model/${var.embedding_model_id}"
}

resource "aws_bedrockagent_knowledge_base" "this" {
  name     = "${var.project_name}-kb"
  role_arn = aws_iam_role.bedrock_kb.arn

  knowledge_base_configuration {
    type = "VECTOR"
    vector_knowledge_base_configuration {
      embedding_model_arn = local.embedding_model_arn
      embedding_model_configuration {
        bedrock_embedding_model_configuration {
          dimensions = var.embedding_dimension
        }
      }
    }
  }

  storage_configuration {
    type = "OPENSEARCH_SERVERLESS"
    opensearch_serverless_configuration {
      collection_arn    = aws_opensearchserverless_collection.kb.arn
      vector_index_name = opensearch_index.kb_vector_index.name
      field_mapping {
        vector_field   = "bedrock-knowledge-base-default-vector"
        text_field     = "AMAZON_BEDROCK_TEXT_CHUNK"
        metadata_field = "AMAZON_BEDROCK_METADATA"
      }
    }
  }

  depends_on = [aws_iam_role_policy.bedrock_kb, time_sleep.wait_for_index]
}

resource "aws_bedrockagent_data_source" "docs" {
  knowledge_base_id = aws_bedrockagent_knowledge_base.this.id
  name              = "${var.project_name}-docs-source"

  data_source_configuration {
    type = "S3"
    s3_configuration {
      bucket_arn = aws_s3_bucket.docs.arn
    }
  }

  vector_ingestion_configuration {
    chunking_configuration {
      chunking_strategy = "FIXED_SIZE"
      fixed_size_chunking_configuration {
        max_tokens         = var.chunk_max_tokens
        overlap_percentage = var.chunk_overlap_percentage
      }
    }
  }
}

OpenSearch Serverless Config for security (encryption, network), VectorDB config (embedding dimension, hnsw, faiss, knn_vector):

resource "aws_opensearchserverless_security_policy" "encryption" {
  name = "${var.project_name}-encrypt"
  type = "encryption"
  policy = jsonencode({
    Rules = [{
      ResourceType = "collection"
      Resource     = ["collection/${var.project_name}-kb"]
    }]
    AWSOwnedKey = true
  })
}

resource "aws_opensearchserverless_security_policy" "network" {
  name = "${var.project_name}-network"
  type = "network"
  policy = jsonencode([{
    Rules = [{
      ResourceType = "collection"
      Resource     = ["collection/${var.project_name}-kb"]
    }]
    AllowFromPublic = true
  }])
}

resource "aws_opensearchserverless_collection" "kb" {
  name = "${var.project_name}-kb"
  type = "VECTORSEARCH"

  depends_on = [
    aws_opensearchserverless_security_policy.encryption,
    aws_opensearchserverless_security_policy.network,
  ]
}

resource "aws_opensearchserverless_access_policy" "kb" {
  name = "${var.project_name}-access"
  type = "data"
  policy = jsonencode([{
    Rules = [
      {
        ResourceType = "collection"
        Resource     = ["collection/${var.project_name}-kb"]
        Permission   = ["aoss:*"]
      },
      {
        ResourceType = "index"
        Resource     = ["index/${var.project_name}-kb/*"]
        Permission   = ["aoss:*"]
      }
    ]
    Principal = [
      aws_iam_role.bedrock_kb.arn,
      var.admin_principal_arn != null ? var.admin_principal_arn : data.aws_caller_identity.current.arn,
    ]
  }])
}

resource "time_sleep" "wait_for_access_policy" {
  depends_on      = [aws_opensearchserverless_access_policy.kb]
  create_duration = "60s"
}

provider "opensearch" {
  url               = aws_opensearchserverless_collection.kb.collection_endpoint
  healthcheck       = false
  aws_region        = var.aws_region
  sign_aws_requests = true
}

resource "opensearch_index" "kb_vector_index" {
  name      = "${var.project_name}-index"
  index_knn = true

  mappings = jsonencode({
    properties = {
      "bedrock-knowledge-base-default-vector" = {
        type      = "knn_vector"
        dimension = var.embedding_dimension
        method = {
          name       = "hnsw"
          engine     = "faiss"
          space_type = "l2"
        }
      }
      "AMAZON_BEDROCK_TEXT_CHUNK" = { type = "text" }
      "AMAZON_BEDROCK_METADATA"   = { type = "text", index = false }
    }
  })

  depends_on = [time_sleep.wait_for_access_policy]
}

resource "time_sleep" "wait_for_index" {
  depends_on      = [opensearch_index.kb_vector_index]
  create_duration = "30s"
}

Also, variables.tf, output.tf implemented, you can reach from => Terraform Codes. We can run these TF files with terraform plan, apply, destroy.

---

Streamlit App: Digest & Querying

App needed to upload files to S3, synch with RAG, and querying on the RAG.

App.py (chat history, uploading docs to s3 bucket, triggering ingestion on OpenSearch, querying/asking questions):

import streamlit as st

from utils.bedrock_ops import get_ingestion_job_status, query, start_ingestion_job
from utils.s3_ops import upload_file

st.set_page_config(page_title="Simple RAG", layout="wide")

if "chat_history" not in st.session_state:
    st.session_state.chat_history = []
if "ingestion_job_id" not in st.session_state:
    st.session_state.ingestion_job_id = None

with st.sidebar:
    st.header("Chat History")
    for turn in st.session_state.chat_history:
        st.markdown(f"**{turn['role'].capitalize()}:** {turn['content']}")

st.title("Simple RAG")

st.subheader("Upload documents")
uploaded_files = st.file_uploader("Choose file(s)", accept_multiple_files=True)
if uploaded_files and st.button("Upload to S3"):
    for uploaded_file in uploaded_files:
        try:
            key = upload_file(uploaded_file, uploaded_file.name)
            st.success(f"Uploaded '{key}' to S3.")
        except Exception as e:
            st.error(f"Upload failed for '{uploaded_file.name}': {e}")

st.subheader("Sync Knowledge Base")
if st.button("Sync Knowledge Base"):
    try:
        st.session_state.ingestion_job_id = start_ingestion_job()
        st.info(f"Ingestion job started: {st.session_state.ingestion_job_id}")
    except Exception as e:
        st.error(f"Sync failed: {e}")

if st.session_state.ingestion_job_id:
    try:
        job_info = get_ingestion_job_status(st.session_state.ingestion_job_id)
        st.write(f"Last ingestion job status: **{job_info['status']}**")
        if job_info["status"] == "FAILED" and job_info["failure_reasons"]:
            st.error("Ingestion failed: " + "; ".join(job_info["failure_reasons"]))
    except Exception as e:
        st.error(f"Could not fetch job status: {e}")

st.subheader("Ask a question")
question = st.chat_input("Ask a question about your documents")
if question:
    st.session_state.chat_history.append({"role": "user", "content": question})
    try:
        result = query(question)
        answer = result["answer"]
        if result["citations"]:
            answer += "\n\nSources:\n" + "\n".join(f"- {c}" for c in result["citations"])
        else:
            answer += "\n\n_No source documents were found for this question — make sure you've uploaded and synced documents._"
    except Exception as e:
        answer = f"Error: {e}"
    st.session_state.chat_history.append({"role": "assistant", "content": answer})

for turn in st.session_state.chat_history:
    with st.chat_message(turn["role"]):
        st.markdown(turn["content"])

To get from terraform output to config .env:

AWS_REGION=eu-central-1
DOCS_BUCKET_NAME=simple-rag-docs-xxx
KNOWLEDGE_BASE_ID=STCXx
DATA_SOURCE_ID=PSGxx
GENERATION_MODEL_ARN=arn:aws:bedrock:eu-central-1:xxx:inference-profile/eu.anthropic.claude-sonnet-4-5-20250929-v1:0

Clients for bedrock-agents for ingestion_job, querying; s3; aws_clients.py:

import os

import boto3
from dotenv import load_dotenv

load_dotenv()


def get_config() -> dict:
    return {
        "region": os.environ["AWS_REGION"],
        "docs_bucket": os.environ["DOCS_BUCKET_NAME"],
        "knowledge_base_id": os.environ["KNOWLEDGE_BASE_ID"],
        "data_source_id": os.environ["DATA_SOURCE_ID"],
        "generation_model_arn": os.environ["GENERATION_MODEL_ARN"],
    }

def get_s3_client():
    return boto3.client("s3", region_name=get_config()["region"])

def get_bedrock_agent_client():
    return boto3.client("bedrock-agent", region_name=get_config()["region"])


def get_bedrock_agent_runtime_client():
    return boto3.client("bedrock-agent-runtime", region_name=get_config()["region"])

Start_ingestion_job, get_ingestion_job_status, querying; bedrock-ops.py:

from utils.aws_clients import get_bedrock_agent_client, get_bedrock_agent_runtime_client, get_config


def start_ingestion_job() -> str:
    cfg = get_config()
    resp = get_bedrock_agent_client().start_ingestion_job(
        knowledgeBaseId=cfg["knowledge_base_id"],
        dataSourceId=cfg["data_source_id"],
    )
    return resp["ingestionJob"]["ingestionJobId"]


def get_ingestion_job_status(job_id: str) -> dict:
    cfg = get_config()
    resp = get_bedrock_agent_client().get_ingestion_job(
        knowledgeBaseId=cfg["knowledge_base_id"],
        dataSourceId=cfg["data_source_id"],
        ingestionJobId=job_id,
    )
    job = resp["ingestionJob"]
    return {"status": job["status"], "failure_reasons": job.get("failureReasons", [])}


def query(question: str) -> dict:
    cfg = get_config()
    resp = get_bedrock_agent_runtime_client().retrieve_and_generate(
        input={"text": question},
        retrieveAndGenerateConfiguration={
            "type": "KNOWLEDGE_BASE",
            "knowledgeBaseConfiguration": {
                "knowledgeBaseId": cfg["knowledge_base_id"],
                "modelArn": cfg["generation_model_arn"],
            },
        },
    )

    citations = []
    for citation in resp.get("citations", []):
        for ref in citation.get("retrievedReferences", []):
            uri = ref.get("location", {}).get("s3Location", {}).get("uri")
            if uri:
                citations.append(uri)

    return {"answer": resp["output"]["text"], "citations": citations}

S3 upload jobs on S3 bucket; s3_ops.py:

from utils.aws_clients import get_config, get_s3_client

def upload_file(file_obj, filename: str) -> str:
    cfg = get_config()
    key = filename
    get_s3_client().upload_fileobj(file_obj, cfg["docs_bucket"], key)
    return key

---

All Code & Demo

GitHub Link: Project on GitHub

Run:

cd terraform
terraform init
terraform plan
terraform apply
Apply complete! Resources: 14 added, 0 changed, 0 destroyed.

Outputs:

aws_region = "eu-central-1"
data_source_id = "PSGxx"
docs_bucket_name = "simple-rag-docs-xx"
generation_model_arn = "arn:aws:bedrock:eu-central-1:xxx:inference-profile/eu.anthropic.claude-sonnet-4-5-20250929-v1:0"
knowledge_base_id = "STCxx"
## copy above settings in app/.env

cd app
streamlit run app.py
2026-09-xx Uvicorn server started on :::8502
  You can now view your Streamlit app in your browser.
  Local URL: http://localhost:8502
  Network URL: http://172.24.55.241:8502

## remove it after using, with destroy
cd terraform
terraform destroy

All images with better resolution

After applying the TF, view/copy outputs to link with app.py:

Test Claude Sonnet before running app to be sure that you have access to Claude models:

Uploading files (popular latest GenAI papers under docs/pdf):

Uploaded files under S3 Bucket:

Provisioned OpenSearch Serverless:

Monitoring OpenSearch Serverless Details:

Querying & Retrieval Data:

Before removing infra, delete objects from S3 files:

---

Conclusion

In this post, we mentioned:

• how to create RAG using Bedrock KnowledgeBase, S3, OpenSearch Serverless with Terraform,
• how to digest, query using simple streamlit app,
• how to enable with permission roles,
• how to use AWS Claude Sonnet.

If you found the tutorial interesting, I’d love to hear your thoughts in the blog post comments. Feel free to share your reactions or leave a comment. I truly value your input and engagement 😉

For other posts 👉 https://dev.to/omerberatsezer 🧐

References

• https://aws.amazon.com/bedrock
• https://github.com/omerbsezer/Fast-LLM-Agent-MCP/

Your comments 🤔
I’d love to hear your experience and know-how.

• Do you know the detail of RAG that handled by Bedrock Knowledge Base
• How do you implement your RAG in production (e.g. Bedrock, LlamaIndex, LangChain, etc.)?

Drop your thoughts, feedback, or improvements in the comments, always curious to learn from different approaches. ☺️