Building a Production RAG Pipeline on AWS: Bedrock Knowledge Bases, OpenSearch Serverless, and Terraform

A developer built a production RAG pipeline on AWS that indexes 2,400+ internal documents (Confluence pages, runbooks, post-mortems, ADRs) to answer engineering questions with source citations. The stack uses Amazon Titan Text Embed v2 for embeddings, OpenSearch Serverless for retrieval, Claude 3.5 Sonnet via Bedrock for generation, and a Lambda query API behind API Gateway, all provisioned entirely through Terraform. The pipeline reports p50 latency of 1.8 seconds, p99 of 4.2 seconds, and roughly $380 monthly AWS cost at about 500 queries per day.

November 2026 | ~18 min read Every engineering org has the same problem. New engineer joins, asks "how does the deployment pipeline handle database migrations?" The answer exists — buried in a Confluence page from 2023, a Slack thread from last April, and a post-mortem that three people remember reading. Nobody can find it. So the question gets asked again. And again. I built a production-ready RAG pipeline on AWS to solve this. Not a notebook demo. Not a "just call the API" tutorial. A real pipeline — document ingestion from S3, vector embeddings via Amazon Titan, retrieval from OpenSearch Serverless, generation via Claude 3.5 Sonnet through Bedrock, a Lambda query API behind API Gateway, and CloudWatch observability across the whole stack. All deployed with Terraform. Zero console clicks. The goal: any engineer can hit an API endpoint or a Slack bot wired to it with a natural language question and get an accurate answer with citations back to the source documents. ┌──────────────────────────────────────────────────────────────────┐ │ PRODUCTION RAG PIPELINE │ ├──────────────────────────────────┬───────────────────────────────┤ │ Metric │ Value │ ├──────────────────────────────────┼───────────────────────────────┤ │ Documents indexed │ 2,400+ Confluence, runbooks, │ │ │ post-mortems, ADRs │ │ Avg query latency p50 │ 1.8 seconds │ │ Avg query latency p99 │ 4.2 seconds │ │ Monthly AWS cost │ ~$380 at ~500 queries/day │ │ Embedding model │ Amazon Titan Text Embed v2 │ │ Generation model │ Claude 3.5 Sonnet via Bedrock │ │ Infrastructure │ 100% Terraform, zero console │ └──────────────────────────────────┴───────────────────────────────┘ Before writing any Terraform, I had to answer the first question: why RAG instead of fine-tuning a model on our internal docs? ┌──────────────────────────┬───────────────────────┬───────────────────────┐ │ Factor │ RAG │ Fine-Tuning │ ├──────────────────────────┼───────────────────────┼───────────────────────┤ │ Data freshness │ Real-time re-ingest │ Stale retrain cycle │ │ Citation support │ Native source docs │ None │ │ Cost to update │ ~$0 S3 sync │ $100s–$1000s/retrain │ │ Hallucination control │ Grounded in sources │ No guarantees │ │ Setup complexity │ Medium │ High │ │ Latency │ Higher ~2s │ Lower ~500ms │ │ Domain-specific language │ Depends on retrieval │ Strong │ │ Data privacy │ Data stays in your VPC│ Sent to training API │ └──────────────────────────┴───────────────────────┴───────────────────────┘ RAG wins when: Fine-tuning wins when you need specialized domain language medical terminology, legal jargon or sub-second latency. For an internal knowledge base where accuracy and freshness matter more than speed, RAG is the clear choice. ┌─────────────────────────────────────────────────────────────────────────────┐ │ PRODUCTION RAG PIPELINE │ │ │ │ ┌──────────┐ ┌─────────────────┐ ┌──────────────┐ │ │ │ │ │ Bedrock Data │ │ Amazon │ │ │ │ S3 │───▶│ Source │───▶│ Titan │ │ │ │ docs │ │ chunking │ │ Embed v2 │ │ │ └──────────┘ └─────────────────┘ └──────┬───────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────┐ │ │ │ OpenSearch │ │ │ │ Serverless │ │ │ │ vector store │ │ │ └────────┬───────────┘ │ │ │ │ │ ┌──────────┐ ┌─────────────────┐ │ │ │ │ API │ │ Lambda │ │ │ │ │ Gateway │───▶│ query API │───▶ Bedrock KB ───▶ Claude 3.5 │ │ └──────────┘ └─────────────────┘ retrieve + Sonnet │ │ generate answer │ │ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │ CloudWatch: latency, tokens, errors, cost anomaly alarms │ │ │ └─────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ Why OpenSearch Serverless over the alternatives? Component cost breakdown: ┌──────────────────────────────────┬───────────────────────────────┐ │ Component │ Monthly Cost │ ├──────────────────────────────────┼───────────────────────────────┤ │ OpenSearch Serverless 2 OCUs │ ~$350 │ │ Bedrock - Titan Embed v2 │ ~$4 ingestion │ │ Bedrock - Claude 3.5 Sonnet │ ~$18 500 queries/day │ │ Lambda │ ~$2 │ │ API Gateway │ ~$2 │ │ S3 document storage │ <$1 │ │ CloudWatch │ ~$3 │ ├──────────────────────────────────┼───────────────────────────────┤ │ Total │ ~$380/month │ └──────────────────────────────────┴───────────────────────────────┘ Yes, OpenSearch Serverless is 92% of the bill. We'll talk about that. OpenSearch Serverless for vector search requires three things: an encryption policy, a network policy, and a data access policy. Without all three, the collection won't create. This trips up everyone the first time. opensearch-serverless.tf resource "aws opensearchserverless security policy" "encryption" { name = "rag-kb-encryption" type = "encryption" policy = jsonencode { Rules = { Resource = "collection/rag-knowledge-base" ResourceType = "collection" } AWSOwnedKey = true } } resource "aws opensearchserverless security policy" "network" { name = "rag-kb-network" type = "network" policy = jsonencode { Rules = { Resource = "collection/rag-knowledge-base" ResourceType = "collection" }, { Resource = "collection/rag-knowledge-base" ResourceType = "dashboard" } AllowFromPublic = true } } resource "aws opensearchserverless collection" "kb" { name = "rag-knowledge-base" description = "Vector store for RAG knowledge base" type = "VECTORSEARCH" depends on = aws opensearchserverless security policy.encryption, aws opensearchserverless security policy.network, } resource "aws opensearchserverless access policy" "kb" { name = "rag-kb-access" type = "data" policy = jsonencode { Rules = { Resource = "collection/rag-knowledge-base" ResourceType = "collection" Permission = "aoss:CreateCollectionItems", "aoss:DeleteCollectionItems", "aoss:UpdateCollectionItems", "aoss:DescribeCollectionItems" }, { Resource = "index/rag-knowledge-base/ " ResourceType = "index" Permission = "aoss:CreateIndex", "aoss:DeleteIndex", "aoss:UpdateIndex", "aoss:DescribeIndex", "aoss:ReadDocument", "aoss:WriteDocument" } Principal = aws iam role.bedrock kb.arn, data.aws caller identity.current.arn } } A few things to note: AllowFromPublic = true depends on block is required. Bedrock Knowledge Bases require a specific index schema. You can't create this index through Terraform — the OpenSearch Serverless API doesn't expose index management. Use a Python script: python scripts/create vector index.py import boto3 import json from opensearchpy import OpenSearch, RequestsAWSV4SignerAuth, AWSV4SignerAuth from requests aws4auth import AWS4Auth COLLECTION ENDPOINT = "https://your-collection-id.us-east-1.aoss.amazonaws.com" INDEX NAME = "bedrock-knowledge-base-default-index" REGION = "us-east-1" credentials = boto3.Session .get credentials auth = AWS4Auth credentials.access key, credentials.secret key, REGION, "aoss", session token=credentials.token, client = OpenSearch hosts= {"host": COLLECTION ENDPOINT.replace "https://", "" , "port": 443} , http auth=auth, use ssl=True, verify certs=True, connection class="RequestsHttpConnection", index body = { "settings": { "index": { "knn": True, "knn.algo param.ef search": 512, } }, "mappings": { "properties": { "bedrock-knowledge-base-default-vector": { "type": "knn vector", "dimension": 1024, "method": { "engine": "faiss", "space type": "l2", "name": "hnsw", "parameters": { "ef construction": 512, "m": 16, }, }, }, "AMAZON BEDROCK METADATA": { "type": "text", "index": False, }, "AMAZON BEDROCK TEXT CHUNK": { "type": "text", }, } }, } response = client.indices.create index=INDEX NAME, body=index body print json.dumps response, indent=2 Why the field names matter: Bedrock Knowledge Bases expect specific field names — bedrock-knowledge-base-default-vector , AMAZON BEDROCK TEXT CHUNK , and AMAZON BEDROCK METADATA . If you use custom field names, you'll need to specify the mapping in the knowledge base configuration. Using the defaults saves a layer of configuration. Dimension 1024 matches Amazon Titan Text Embeddings v2. If you switch to a different embedding model, this number changes Cohere Embed = 1024, Titan v1 = 1536 . The knowledge base ties everything together: it connects S3 documents to the embedding model Titan to the vector store OpenSearch Serverless . The Bedrock service needs permissions to read from S3, generate embeddings, and write to OpenSearch Serverless: iam.tf data "aws caller identity" "current" {} resource "aws iam role" "bedrock kb" { name = "bedrock-knowledge-base-role" assume role policy = jsonencode { Version = "2012-10-17" Statement = { Action = "sts:AssumeRole" Effect = "Allow" Principal = { Service = "bedrock.amazonaws.com" } Condition = { StringEquals = { "aws:SourceAccount" = data.aws caller identity.current.account id } } } } } resource "aws iam role policy" "bedrock kb" { name = "bedrock-knowledge-base-policy" role = aws iam role.bedrock kb.id policy = jsonencode { Version = "2012-10-17" Statement = { Sid = "BedrockInvokeModel" Effect = "Allow" Action = "bedrock:InvokeModel" Resource = "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0" }, { Sid = "S3ReadDocuments" Effect = "Allow" Action = "s3:GetObject", "s3:ListBucket" Resource = aws s3 bucket.documents.arn, "${aws s3 bucket.documents.arn}/ " }, { Sid = "OpenSearchServerlessAccess" Effect = "Allow" Action = "aoss:APIAccessAll" Resource = aws opensearchserverless collection.kb.arn } } } The aws:SourceAccount condition in the trust policy prevents confused deputy attacks — without it, any AWS account could potentially assume your Bedrock role. knowledge-base.tf resource "aws s3 bucket" "documents" { bucket = "rag-knowledge-base-documents-${data.aws caller identity.current.account id}" } resource "aws s3 bucket versioning" "documents" { bucket = aws s3 bucket.documents.id versioning configuration { status = "Enabled" } } resource "aws bedrockagent knowledge base" "main" { name = "internal-knowledge-base" role arn = aws iam role.bedrock kb.arn knowledge base configuration { type = "VECTOR" vector knowledge base configuration { embedding model arn = "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0" } } storage configuration { type = "OPENSEARCH SERVERLESS" opensearch serverless configuration { collection arn = aws opensearchserverless collection.kb.arn vector index name = "bedrock-knowledge-base-default-index" field mapping { vector field = "bedrock-knowledge-base-default-vector" text field = "AMAZON BEDROCK TEXT CHUNK" metadata field = "AMAZON BEDROCK METADATA" } } } } resource "aws bedrockagent data source" "s3" { name = "s3-documents" knowledge base id = aws bedrockagent knowledge base.main.id data deletion policy = "RETAIN" data source configuration { type = "S3" s3 configuration { bucket arn = aws s3 bucket.documents.arn } } vector ingestion configuration { chunking configuration { chunking strategy = "FIXED SIZE" fixed size chunking configuration { max tokens = 512 overlap percentage = 20 } } } } Bedrock Knowledge Bases support three chunking strategies: I chose fixed-size at 512 tokens with 20% overlap for three reasons: The 20% overlap 102 tokens ensures that if a concept spans a chunk boundary, at least part of it appears in both chunks. Without overlap, you get answers that start mid-sentence. After deploying the infrastructure and uploading documents to S3, trigger the initial sync: bash /bin/bash scripts/trigger ingestion.sh KNOWLEDGE BASE ID=$ terraform output -raw knowledge base id DATA SOURCE ID=$ terraform output -raw data source id echo "Starting ingestion job..." INGESTION JOB=$ aws bedrock-agent start-ingestion-job \ --knowledge-base-id "$KNOWLEDGE BASE ID" \ --data-source-id "$DATA SOURCE ID" \ --output json JOB ID=$ echo "$INGESTION JOB" | jq -r '.ingestionJob.ingestionJobId' echo "Job ID: $JOB ID" Poll until complete while true; do STATUS=$ aws bedrock-agent get-ingestion-job \ --knowledge-base-id "$KNOWLEDGE BASE ID" \ --data-source-id "$DATA SOURCE ID" \ --ingestion-job-id "$JOB ID" \ --query 'ingestionJob.status' \ --output text echo "Status: $STATUS" if "$STATUS" = "COMPLETE" || "$STATUS" = "FAILED" ; then break fi sleep 10 done Show statistics aws bedrock-agent get-ingestion-job \ --knowledge-base-id "$KNOWLEDGE BASE ID" \ --data-source-id "$DATA SOURCE ID" \ --ingestion-job-id "$JOB ID" \ --query 'ingestionJob.statistics' For 2,400 documents, initial ingestion takes about 15 minutes. Subsequent incremental syncs process only changed/new files and finish in under a minute. The query layer is a Lambda function behind API Gateway. It takes a natural language question, calls Bedrock's RetrieveAndGenerate API, and returns a structured response with citations. python lambda/query handler.py import json import os import logging import boto3 logger = logging.getLogger logger.setLevel logging.INFO bedrock agent = boto3.client "bedrock-agent-runtime" KNOWLEDGE BASE ID = os.environ "KNOWLEDGE BASE ID" MODEL ARN = os.environ.get "MODEL ARN", "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0", SYSTEM PROMPT = """You are an internal knowledge assistant for an engineering team. Answer questions using ONLY the provided search results. If the search results don't contain enough information to answer the question, say so clearly — do not make up information. Rules: - Cite your sources by referencing the document title and section - If multiple documents provide relevant information, synthesize them - For code examples, include the full snippet from the source document - If the question is ambiguous, ask for clarification - Keep answers concise but complete""" def handler event, context : try: body = json.loads event.get "body", "{}" question = body.get "question", "" .strip if not question: return response 400, {"error": "Missing 'question' in request body"} if len question 1000: return response 400, {"error": "Question exceeds 1000 character limit"} logger.info f"Query: {question :100 }..." result = bedrock agent.retrieve and generate input={"text": question}, retrieveAndGenerateConfiguration={ "type": "KNOWLEDGE BASE", "knowledgeBaseConfiguration": { "knowledgeBaseId": KNOWLEDGE BASE ID, "modelArn": MODEL ARN, "generationConfiguration": { "promptTemplate": { "textPromptTemplate": f"{SYSTEM PROMPT}\n\n" "Search results:\n$search results$\n\n" "Question: $query$\n\n" "Answer:" } }, "retrievalConfiguration": { "vectorSearchConfiguration": { "numberOfResults": 5, "overrideSearchType": "HYBRID", } }, }, }, answer = result "output" "text" citations = for citation in result.get "citations", : for ref in citation.get "retrievedReferences", : location = ref.get "location", {} s3 uri = location.get "s3Location", {} .get "uri", "" content text = ref.get "content", {} .get "text", "" :200 citations.append { "source": s3 uri.split "/" -1 if s3 uri else "unknown", "excerpt": content text, } return response 200, { "answer": answer, "citations": citations, "query": question, }, except bedrock agent.exceptions.ThrottlingException: logger.warning "Bedrock throttled" return response 429, {"error": "Rate limited. Try again in a few seconds."} except Exception as e: logger.error f"Query failed: {str e }", exc info=True return response 500, {"error": "Internal server error"} def response status code, body : return { "statusCode": status code, "headers": { "Content-Type": "application/json", "Access-Control-Allow-Origin": " ", }, "body": json.dumps body , } Key design decisions: overrideSearchType: "HYBRID" numberOfResults: 5 lambda.tf data "archive file" "query handler" { type = "zip" source file = "${path.module}/lambda/query handler.py" output path = "${path.module}/build/query handler.zip" } resource "aws lambda function" "query" { function name = "rag-query-handler" filename = data.archive file.query handler.output path source code hash = data.archive file.query handler.output base64sha256 handler = "query handler.handler" runtime = "python3.12" timeout = 30 memory size = 256 role = aws iam role.lambda query.arn environment { variables = { KNOWLEDGE BASE ID = aws bedrockagent knowledge base.main.id MODEL ARN = "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0" } } } resource "aws iam role" "lambda query" { name = "rag-query-lambda-role" assume role policy = jsonencode { Version = "2012-10-17" Statement = { Action = "sts:AssumeRole" Effect = "Allow" Principal = { Service = "lambda.amazonaws.com" } } } } resource "aws iam role policy" "lambda query" { name = "rag-query-lambda-policy" role = aws iam role.lambda query.id policy = jsonencode { Version = "2012-10-17" Statement = { Sid = "BedrockRetrieveAndGenerate" Effect = "Allow" Action = "bedrock:RetrieveAndGenerate", "bedrock:Retrieve", "bedrock:InvokeModel" Resource = " " }, { Sid = "CloudWatchLogs" Effect = "Allow" Action = "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" Resource = "arn:aws:logs: : : " } } } API Gateway resource "aws api gateway rest api" "rag" { name = "rag-query-api" description = "RAG Knowledge Base Query API" endpoint configuration { types = "REGIONAL" } } resource "aws api gateway resource" "query" { rest api id = aws api gateway rest api.rag.id parent id = aws api gateway rest api.rag.root resource id path part = "query" } resource "aws api gateway method" "query post" { rest api id = aws api gateway rest api.rag.id resource id = aws api gateway resource.query.id http method = "POST" authorization = "NONE" } resource "aws api gateway integration" "query lambda" { rest api id = aws api gateway rest api.rag.id resource id = aws api gateway resource.query.id http method = aws api gateway method.query post.http method integration http method = "POST" type = "AWS PROXY" uri = aws lambda function.query.invoke arn } resource "aws lambda permission" "api gateway" { statement id = "AllowAPIGatewayInvoke" action = "lambda:InvokeFunction" function name = aws lambda function.query.function name principal = "apigateway.amazonaws.com" source arn = "${aws api gateway rest api.rag.execution arn}/ / " } resource "aws api gateway deployment" "rag" { rest api id = aws api gateway rest api.rag.id depends on = aws api gateway integration.query lambda, lifecycle { create before destroy = true } } resource "aws api gateway stage" "prod" { rest api id = aws api gateway rest api.rag.id deployment id = aws api gateway deployment.rag.id stage name = "prod" } curl -X POST https://your-api-id.execute-api.us-east-1.amazonaws.com/prod/query \ -H "Content-Type: application/json" \ -d '{"question": "How does our deployment pipeline handle database migrations?"}' Response: { "answer": "The deployment pipeline handles database migrations through a two-phase process. First, the CI/CD pipeline runs alembic upgrade head in a pre-deploy Lambda that executes against the staging database. If migrations succeed, the pipeline proceeds to the blue-green deployment swap. Rollback migrations are required for every forward migration — the pipeline validates that alembic downgrade works before approving the PR.\n\nFor breaking schema changes, the team uses the expand-contract pattern: deploy the new schema alongside the old one, migrate data, then remove the old schema in a subsequent release.", "citations": { "source": "deployment-pipeline-runbook.md", "excerpt": "Database migrations run as a pre-deploy step using Alembic. The migration Lambda has a 5-minute timeout and connects via RDS Proxy..." }, { "source": "adr-024-schema-migration-strategy.md", "excerpt": "We adopt the expand-contract pattern for all breaking schema changes to enable zero-downtime deployments..." } , "query": "How does our deployment pipeline handle database migrations?" } That's the power of RAG with citations. The answer synthesizes two source documents and tells you exactly where to go for the full details. A RAG pipeline without observability is a black box. You don't know if answers are degrading, costs are spiking, or queries are failing silently. observability.tf resource "aws cloudwatch dashboard" "rag" { dashboard name = "rag-pipeline" dashboard body = jsonencode { widgets = { type = "metric" x = 0 y = 0 width = 12 height = 6 properties = { title = "Query Latency ms " metrics = "AWS/Lambda", "Duration", "FunctionName", "rag-query-handler", { stat = "p50" } , "AWS/Lambda", "Duration", "FunctionName", "rag-query-handler", { stat = "p99" } , period = 300 view = "timeSeries" } }, { type = "metric" x = 12 y = 0 width = 12 height = 6 properties = { title = "Query Volume & Errors" metrics = "AWS/Lambda", "Invocations", "FunctionName", "rag-query-handler", { stat = "Sum" } , "AWS/Lambda", "Errors", "FunctionName", "rag-query-handler", { stat = "Sum" } , "AWS/Lambda", "Throttles", "FunctionName", "rag-query-handler", { stat = "Sum" } , period = 300 view = "timeSeries" } }, { type = "metric" x = 0 y = 6 width = 12 height = 6 properties = { title = "Bedrock Token Usage" metrics = "AWS/Bedrock", "InputTokenCount", "ModelId", "anthropic.claude-3-5-sonnet-20241022-v2:0", { stat = "Sum" } , "AWS/Bedrock", "OutputTokenCount", "ModelId", "anthropic.claude-3-5-sonnet-20241022-v2:0", { stat = "Sum" } , period = 3600 view = "timeSeries" } }, { type = "metric" x = 12 y = 6 width = 12 height = 6 properties = { title = "OpenSearch Serverless" metrics = "AWS/AOSS", "SearchOCU", "CollectionId", aws opensearchserverless collection.kb.id , "AWS/AOSS", "IndexingOCU", "CollectionId", aws opensearchserverless collection.kb.id , period = 3600 view = "timeSeries" } } } } resource "aws cloudwatch metric alarm" "high latency" { alarm name = "rag-query-high-latency" comparison operator = "GreaterThanThreshold" evaluation periods = 3 metric name = "Duration" namespace = "AWS/Lambda" period = 300 statistic = "p99" threshold = 10000 alarm description = "RAG query p99 latency exceeded 10 seconds for 15 minutes" alarm actions = aws sns topic.alerts.arn dimensions = { FunctionName = aws lambda function.query.function name } } resource "aws cloudwatch metric alarm" "high error rate" { alarm name = "rag-query-high-error-rate" comparison operator = "GreaterThanThreshold" evaluation periods = 2 threshold = 5 metric query { id = "error rate" expression = " errors / invocations 100" label = "Error Rate %" return data = true } metric query { id = "errors" metric { metric name = "Errors" namespace = "AWS/Lambda" period = 300 stat = "Sum" dimensions = { FunctionName = aws lambda function.query.function name } } } metric query { id = "invocations" metric { metric name = "Invocations" namespace = "AWS/Lambda" period = 300 stat = "Sum" dimensions = { FunctionName = aws lambda function.query.function name } } } alarm description = "RAG query error rate exceeded 5% for 10 minutes" alarm actions = aws sns topic.alerts.arn } resource "aws sns topic" "alerts" { name = "rag-pipeline-alerts" } Bedrock can log every model invocation — input, output, latency, token counts — to S3 and/or CloudWatch. This is essential for debugging bad answers and tracking costs: resource "aws s3 bucket" "bedrock logs" { bucket = "rag-bedrock-invocation-logs-${data.aws caller identity.current.account id}" } resource "aws bedrock model invocation logging configuration" "main" { logging config { embedding data delivery enabled = true s3 config { bucket name = aws s3 bucket.bedrock logs.id key prefix = "invocation-logs/" } } } With invocation logging enabled, you can query the S3 logs with Athena to answer questions like "which queries had the highest token counts this week?" or "what questions produced empty retrieval results?" The $350/month OpenSearch Serverless minimum is the elephant in the room. Here's how to minimize everything else: The initial sync is manual — upload docs to S3, run the ingestion script. For production, you want an event-driven pipeline that automatically ingests new or updated documents. S3 PutObject Event → EventBridge Rule → Lambda → Bedrock StartIngestionJob ingestion-pipeline.tf resource "aws lambda function" "ingest trigger" { function name = "rag-ingest-trigger" filename = data.archive file.ingest trigger.output path source code hash = data.archive file.ingest trigger.output base64sha256 handler = "ingest trigger.handler" runtime = "python3.12" timeout = 60 memory size = 128 role = aws iam role.lambda ingest.arn environment { variables = { KNOWLEDGE BASE ID = aws bedrockagent knowledge base.main.id DATA SOURCE ID = aws bedrockagent data source.s3.data source id } } } resource "aws iam role" "lambda ingest" { name = "rag-ingest-lambda-role" assume role policy = jsonencode { Version = "2012-10-17" Statement = { Action = "sts:AssumeRole" Effect = "Allow" Principal = { Service = "lambda.amazonaws.com" } } } } resource "aws iam role policy" "lambda ingest" { name = "rag-ingest-lambda-policy" role = aws iam role.lambda ingest.id policy = jsonencode { Version = "2012-10-17" Statement = { Effect = "Allow" Action = "bedrock:StartIngestionJob" Resource = aws bedrockagent knowledge base.main.arn }, { Effect = "Allow" Action = "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" Resource = "arn:aws:logs: : : " } } } EventBridge rule for S3 events resource "aws s3 bucket notification" "documents" { bucket = aws s3 bucket.documents.id eventbridge = true } resource "aws cloudwatch event rule" "s3 upload" { name = "rag-document-upload" description = "Trigger ingestion when documents are uploaded to S3" event pattern = jsonencode { source = "aws.s3" detail-type = "Object Created" detail = { bucket = { name = aws s3 bucket.documents.id } object = { key = { suffix = ".md" }, { suffix = ".pdf" }, { suffix = ".html" }, { suffix = ".txt" } } } } } resource "aws cloudwatch event target" "ingest lambda" { rule = aws cloudwatch event rule.s3 upload.name arn = aws lambda function.ingest trigger.arn } resource "aws lambda permission" "eventbridge" { statement id = "AllowEventBridgeInvoke" action = "lambda:InvokeFunction" function name = aws lambda function.ingest trigger.function name principal = "events.amazonaws.com" source arn = aws cloudwatch event rule.s3 upload.arn } python lambda/ingest trigger.py import json import os import logging import time import boto3 logger = logging.getLogger logger.setLevel logging.INFO bedrock agent = boto3.client "bedrock-agent" KNOWLEDGE BASE ID = os.environ "KNOWLEDGE BASE ID" DATA SOURCE ID = os.environ "DATA SOURCE ID" Debounce: don't start a new ingestion if one is already running def handler event, context : logger.info f"S3 event received: {json.dumps event }" Check if an ingestion job is already running jobs = bedrock agent.list ingestion jobs knowledgeBaseId=KNOWLEDGE BASE ID, dataSourceId=DATA SOURCE ID, sortBy={"attribute": "STARTED AT", "order": "DESCENDING"}, maxResults=1, if jobs.get "ingestionJobSummaries" : latest = jobs "ingestionJobSummaries" 0 if latest "status" in "STARTING", "IN PROGRESS" : logger.info f"Ingestion job {latest 'ingestionJobId' } already running. Skipping." return {"statusCode": 200, "body": "Ingestion already in progress"} Start new ingestion response = bedrock agent.start ingestion job knowledgeBaseId=KNOWLEDGE BASE ID, dataSourceId=DATA SOURCE ID, job id = response "ingestionJob" "ingestionJobId" logger.info f"Started ingestion job: {job id}" return {"statusCode": 200, "body": json.dumps {"ingestionJobId": job id} } The debounce check is important. If someone uploads 50 documents in quick succession, you don't want 50 ingestion jobs — one sync will pick up all the new files. The Lambda checks for a running job before starting a new one. Bedrock Knowledge Bases natively handle: For Confluence, I export pages as HTML Confluence's REST API supports this and drop them in S3. A small Lambda handles the export on a nightly schedule. 1. OpenSearch Serverless minimum cost is the biggest surprise. $350/month for 2 OCUs before you index a single document. This is the minimum — you can't go lower. If your RAG pipeline is for a small team or a proof of concept, this cost is hard to justify. Watch for S3 Vectors support in Bedrock Knowledge Bases — when it ships, it'll drop the storage cost to pennies. 2. Chunking strategy matters more than model choice. I spent a week trying different generation models Haiku, Sonnet, Llama before realizing the answer quality was determined by what the retriever returned, not which model summarized it. Bad chunks → bad retrieval → bad answers, regardless of model. Fix your chunking before upgrading your model. 3. Bedrock Knowledge Bases handle 80% of the RAG plumbing — don't rebuild it. The temptation to build a custom pipeline with LangChain or LlamaIndex is strong. Resist it unless you have a specific reason. Bedrock Knowledge Bases handle document parsing, chunking, embedding, indexing, retrieval, and citation generation out of the box. That's months of engineering you're not doing. 4. Citation quality depends on metadata, not just content. The difference between "According to a document..." and "According to the Q3 2026 Post-Mortem for the Payment Service last updated October 15 " is metadata. Store descriptive filenames and folder structures in S3. The metadata is included in the vector store and returned with citations. 5. Start with Bedrock's managed RAG, escape to custom only when you hit limits. You'll hit limits — Bedrock Knowledge Bases don't support custom reranking, multi-modal retrieval, or query decomposition out of the box. But you'll discover which limits matter by running the managed version first. Don't prematurely optimize. Resource: " " to specific ARNs