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.
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:
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:
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.
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 up documents to S3, trigger the initial sync:
#!/bin/bash
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"
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
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.
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
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:*:*:*"
}
]
})
}
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.
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
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:*:*:*"
}
]
})
}
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
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"]
def handler(event, context):
logger.info(f"S3 event received: {json.dumps(event)}")
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"}
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