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

Implementing RAG (Document Ingestion, Document Retrieval Process) using Bedrock KnowledgeBase, OpenSearch, S3 with Terraform; Streamlit App to upload files, to trigger ingestion, to query & retrieval data with source citations
Ömer Berat Sezer

AI & Platform Engineer
Published Sep 14, 2026
Last modified Sep 14, 2026
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
Bash
- Upload documents → Amazon S3
- Split documents → Chunks
- Generate embeddings → Embedding Model
- Store embeddings + text + metadata → Vector Database
Query & Retrieval
Bash
- 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 ConfigurationFor 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 😉
﻿
Terraform: Bedrock KnowledgeBase, OpenSearch Serverless, S3, IAM Terraform providers and version:
Terraform
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:
tf
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:
tf
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):
tf
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):
tf
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):
Python
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 :
Python
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 :
Python
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 :
Python
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 :
Python
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:
Bash
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:
﻿
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 infrastructure using terraform destroy, 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. ☺️