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

> Source: <https://dev.to/garrett_yan/building-a-production-rag-pipeline-on-aws-bedrock-knowledge-bases-opensearch-serverless-and-1gpd>
> Published: 2026-11-02 21:35:17+00:00

*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