November 2026 | ~18 min read
The Question Nobody Could Answer
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.
The Numbers
┌──────────────────────────────────────────────────────────────────┐
│ 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 │
└──────────────────────────────────┴───────────────────────────────┘
Table of Contents
- Why RAG Over Fine-Tuning
- Architecture Overview
- OpenSearch Serverless Collection
- Bedrock Knowledge Base
- Query API with Lambda
- Observability and Cost Control
- Document Ingestion Pipeline
- Lessons Learned
- Action Items Checklist
Why RAG Over Fine-Tuning
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:
- Documents change frequently. Internal runbooks, ADRs, and post-mortems update weekly. Fine-tuning means retraining every time a doc changes. RAG means dropping a new file in S3.
- You need citations. "According to the post-mortem from March 2026..." is the difference between a useful answer and a hallucination you can't verify.
- Cost matters. Fine-tuning Claude or Llama on 2,400 documents costs real money per training run. RAG costs pennies per ingestion.
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.
Architecture Overview
┌─────────────────────────────────────────────────────────────────────────────┐
│ 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?
- Pinecone: Great product, but adds a third-party dependency and egress costs. Bedrock Knowledge Bases have native AOSS integration — no custom retrieval code needed.
- pgvector (Aurora/RDS): Solid for small datasets. Falls behind at scale — no built-in ANN indexing tuning, and you're paying for a full database instance.
- S3 Vectors: New and promising, but Bedrock Knowledge Bases don't support it as a storage backend yet. When they do, it'll be the cheapest option.
- OpenSearch Serverless: Native Bedrock integration, managed scaling, no cluster to babysit. The minimum cost stings ($350/month for 2 OCUs), but you're trading ops burden for dollars.
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 Collection
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.
Terraform Configuration
# 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 = trueon the network policy: For a production deployment, you'd scope this to a VPC endpoint. I'm keeping it public here for simplicity — the data access policy controls who can actually read/write. -
The
depends_onblock is required. Without it, Terraform tries to create the collection before the encryption policy exists, and the API returns a cryptic error about missing encryption configuration. - Both the Bedrock role and your own IAM identity need access. The Bedrock role writes embeddings during ingestion. Your identity is needed to create the vector index.
Creating the Vector Index
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:
# 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).
Bedrock Knowledge Base
The knowledge base ties everything together: it connects S3 (documents) to the embedding model (Titan) to the vector store (OpenSearch Serverless).
IAM Role
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 and Data Source
# 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
}
}
}
}
Chunking Strategy: Why Fixed-Size at 512 Tokens
Bedrock Knowledge Bases support three chunking strategies:
- Default chunking: Bedrock decides. Uses ~300 token chunks with automatic overlap. Fine for getting started, terrible for control.
- Fixed-size chunking: You set the chunk size and overlap. Predictable, debuggable, and works well for mixed document types.
- Hierarchical chunking: Parent and child chunks. Better semantic coherence, but more complex to tune and debug.
- Semantic chunking: Splits on topic boundaries using embedding similarity. Best results in theory, but slower ingestion and harder to predict costs.
I chose fixed-size at 512 tokens with 20% overlap for three reasons:
- Predictable cost. 2,400 documents at ~2,000 tokens each = ~4,800 chunks. Easy to estimate embedding costs.
- Debuggable retrieval. When an answer is wrong, I can trace it to a specific chunk. With semantic chunking, chunk boundaries are non-deterministic.
- Good enough. After testing all four strategies, fixed-size at 512 tokens produced the best balance of answer quality and retrieval precision for our document types (markdown, HTML, PDF).
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.
Triggering Ingestion
After deploying the infrastructure and uploading documents to S3, trigger the initial sync:
#!/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.
Query API with Lambda
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.
Lambda Function
# 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": Combines vector similarity search with keyword matching. Pure vector search misses exact terms (error codes, service names). Hybrid catches both. -
numberOfResults: 5: Retrieves the top 5 chunks. More than 5 adds noise without improving answer quality. Fewer than 3 misses context. - Input validation: The 1,000-character limit prevents prompt injection via absurdly long inputs. The empty-check prevents wasting Bedrock invocations.
- Citation extraction: We pull the S3 filename and a 200-char excerpt from each retrieved chunk. This gives the user enough to verify the answer without dumping raw chunks.
Terraform for Lambda and API Gateway
# 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"
}
Example Query
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.
Observability and Cost Control
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.
CloudWatch Dashboard
# 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"
}
}
]
})
}
Alarms
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 Model Invocation Logging
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?"
Cost Optimization
The $350/month OpenSearch Serverless minimum is the elephant in the room. Here's how to minimize everything else:
- AOSS OCUs: You're stuck with the 2-OCU minimum (1 indexing + 1 search). For a low-traffic knowledge base, this is the floor. If you're running multiple knowledge bases, they can share a collection to amortize the cost.
- Cache frequent queries: Add a DynamoDB or ElastiCache layer for repeated questions. At 500 queries/day, if 30% are repeats, you save ~$5/month in Bedrock costs. Not huge, but the latency improvement (from 1.8s to ~50ms for cache hits) is significant.
- Right-size the model: For simpler questions, Claude 3 Haiku at $0.25/$1.25 per million tokens is 60x cheaper than Sonnet. You could route simple queries to Haiku and only use Sonnet for complex multi-document synthesis.
- Ingestion scheduling: Don't run continuous sync. Batch ingestion during off-hours to avoid indexing OCU spikes.
Document Ingestion Pipeline
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.
Architecture
S3 PutObject Event → EventBridge Rule → Lambda → Bedrock StartIngestionJob
Terraform
# 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
}
Ingestion Trigger Lambda
# 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.
Supported Document Formats
Bedrock Knowledge Bases natively handle:
- Markdown (.md) — Best results. Clean structure, headings become natural chunk boundaries.
- PDF (.pdf) — Good results for text-heavy docs. Tables and images are extracted but can lose formatting.
- HTML (.html) — Good results. Tags are stripped, content is extracted cleanly.
- Plain text (.txt) — Works, but no structural hints for the chunker.
- Word documents (.docx) — Supported, but convert to Markdown for better chunking if you can.
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.
Lessons Learned
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.
Action Items Checklist
Phase 1: Foundation (Week 1)
- [ ] Create S3 bucket for documents
- [ ] Deploy OpenSearch Serverless collection with encryption, network, and access policies
- [ ] Run Python script to create vector index
- [ ] Deploy Bedrock Knowledge Base with Titan Embed v2
- [ ] Configure S3 data source with fixed-size chunking (512 tokens, 20% overlap)
- [ ] Upload initial document set to S3
- [ ] Run first ingestion job and verify documents are indexed
Phase 2: Query API (Week 2)
- [ ] Deploy Lambda query handler with RetrieveAndGenerate
- [ ] Set up API Gateway with POST /query endpoint
- [ ] Test with sample queries, verify citations are returned
- [ ] Tune prompt template for your team's question patterns
- [ ] Add API key or IAM auth to API Gateway (don't leave it open)
Phase 3: Production Hardening (Week 3)
- [ ] Deploy CloudWatch dashboard with latency, errors, and token metrics
- [ ] Set up alarms for p99 latency and error rate
- [ ] Enable Bedrock model invocation logging to S3
- [ ] Deploy event-driven ingestion pipeline (S3 → EventBridge → Lambda)
- [ ] Set up Confluence export automation (if applicable)
- [ ] Review IAM policies — scope down
Resource: "*"to specific ARNs - [ ] Load test at expected query volume to verify latency targets
- [ ] Set up AWS Budget alert for the RAG pipeline resources
Top comments (0)