Deploy a RAG Customer Support Agent

Build a production RAG chatbot on AWS with LangGraph, pgvector, and Bedrock.

Introduction

30 Second Summary

When you message a company's support chat and get an instant, helpful answer pulled from their own knowledge base, there's usually an AI agent working behind the scenes. That agent needs real architecture to retrieve the right information, remember your conversation, and stay within safe boundaries.

In this project, you will build and deploy a production-grade RAG customer support agent on AWS. You'll wire up the full stack from document ingestion to real-time inference to monitoring.

What You'll Build

A fully deployed AI customer support system on AWS that ingests documents into a vector database, retrieves grounded answers using Claude Haiku 4.5 on Bedrock, and orchestrates the entire flow with a LangGraph agent.

By the end of this project, you'll have:

A RAG pipeline that ingests company documents into Aurora Serverless v2 (pgvector), embeds them with Titan Embeddings v2, and retrieves relevant context at query time.
A LangGraph agent with tool calling, persistent conversation memory via DynamoDB, and Bedrock Guardrails for PII filtering and topic boundaries.
A complete infrastructure-as-code setup with Terraform that provisions and tears down the entire stack in minutes, plus a CloudWatch dashboard for production monitoring.
Secret Mission: Add a human-in-the-loop approval workflow where sensitive actions pause and wait for approval before continuing.

Are there any prerequisites?

You should have Python experience and a basic familiarity with AWS (an existing account is enough). An understanding of REST APIs will also help you follow along with the API deployment steps.

Set Up AWS Environment & Tooling

Before you can build a production AI system on AWS, you need the right tools connected to the right cloud services. Your local machine needs to communicate with AWS to provision infrastructure and call AI models.

In this step, you'll install the AWS CLI and Terraform, configure your credentials, enable access to the AI models you'll use, and set up your Python project with all required dependencies.

In this step, get ready to:

Install and configure the AWS CLI with credentials.
Install Terraform for infrastructure-as-code deployments.
Enable Bedrock model access and set up your Python project.

Create your project and configure the AWS CLI

You need a project folder, a Python virtual environment, and all the libraries your agent will depend on. You'll also configure the AWS CLI so your terminal can talk to AWS.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir rag-support-agent
cd rag-support-agent

Create a Python virtual environment and activate it by running:

python3 -m venv venv
source venv/bin/activate

What about Windows?

On Windows, activate the virtual environment with venv\Scripts\activate instead.

Create the requirements.txt file that lists all the Python packages your project needs. Create a new file called requirements.txt in the rag-support-agent folder and paste the following:

boto3
langchain
langchain-aws
langchain-community
langgraph
fastapi
mangum
psycopg2-binary
pgvector
pydantic
uvicorn

What are these packages?

boto3 is the AWS SDK for Python. langchain and langgraph handle AI agent orchestration. fastapi serves your API. psycopg2-binary and pgvector connect to your PostgreSQL vector database.

mangum adapts FastAPI to run inside AWS Lambda. pydantic handles data validation. uvicorn is used for local development testing.

Install all the dependencies by running:

pip install -r requirements.txt

Wait for all packages to download and install. This may take 1-2 minutes.
Confirm the install succeeded by running:

pip list | grep langchain

You should see langchain, langchain-aws, and langchain-community in the output.

Seeing install errors?

If you see errors about psycopg2-binary, make sure you're using Python 3.12 and that your virtual environment is activated (you should see (venv) in your terminal prompt).

If pip itself isn't found, try pip3 instead.

help me debug Python package installation errors

Now check whether the AWS CLI is already installed. Run this command:

aws --version

✔️ I see aws-cli/2.x.x

The AWS CLI v2 is installed and ready. You need version 2 for Bedrock support. Move on to configuring your credentials below.

ⓧ Command not found or I see aws-cli/1.x.x

You need AWS CLI version 2. Follow the official installation guide for your operating system:

AWS CLI v2 installation guide

After installing, close and reopen your terminal, then run aws --version again to confirm you see aws-cli/2.x.x in the output.

Configure the AWS CLI with your credentials by running:

aws configure

The CLI will prompt you for four values. Enter them one at a time:

Enter your AWS Access Key ID when prompted.
Enter your AWS Secret Access Key when prompted.
Enter us-east-1 for the Default region name.
Enter json for the Default output format.

Where do I find my access keys?

In the AWS Console, go to IAM and find your user under **Users**. Select the **Security credentials** tab and create an access key if you don't have one already. Copy the Access Key ID and Secret Access Key.

Never share your access keys or commit them to version control.

Verify your configuration works by running:

aws sts get-caller-identity

You should see a JSON response with your account ID, user ARN, and user ID. This confirms your CLI can authenticate with AWS.

Seeing InvalidClientTokenId?

Double-check that you pasted the Access Key ID and Secret Access Key correctly. Run aws configure again to re-enter them.

help me fix AWS CLI authentication errors

Install Terraform and verify

Terraform is an Infrastructure as Code tool that lets you define your entire AWS stack in configuration files. You'll use it to provision and tear down all resources in this project.

Check whether Terraform is already installed by running:

terraform --version

✔️ I see a version number

Terraform is installed. Any recent version (1.5+) works for this project. Move on to the next substep.

ⓧ Command not found

Install Terraform using one of these methods:

macOS

Install via Homebrew by running these commands:

brew tap hashicorp/tap
brew install hashicorp/tap/terraform

Windows

Download the Terraform binary from the official install page:
Terraform downloads
Unzip the downloaded file and add the terraform.exe to a folder in your system PATH.

After installing, close and reopen your terminal, then run terraform --version again to confirm the install succeeded.

Terraform still not found?

Make sure you opened a new terminal session after installing. On macOS, if Homebrew isn't installed, visit brew.sh to install it first.

help me troubleshoot Terraform installation

Enable Bedrock model access

Amazon Bedrock provides access to foundation models as a managed service. You need to enable access to the two specific models this project uses: Claude Haiku 4.5 for text generation and Titan Embeddings v2 for vectorization.

Why two models?

A production RAG system uses an embedding model to convert documents into vectors for storage and search. It uses a separate generation model to read the retrieved context and produce human-readable answers.

Claude Haiku 4.5 provides the best cost-to-quality ratio for customer support. Titan Embeddings v2 produces 1024-dimensional vectors optimized for similarity search.

Open the AWS Management Console in your browser and make sure you are in the US East (N. Virginia) region (us-east-1). You can change regions by clicking the region name in the top-right corner of the console.
Search for Bedrock in the console search bar and select Amazon Bedrock.
In the left navigation pane, under Bedrock configurations, choose Model access.
Choose Enable specific models.
In the model list, find and select Claude 3.5 Haiku (from Anthropic) and Titan Text Embeddings V2 (from Amazon).
Click Next.

Anthropic use case form

If this is your first time enabling an Anthropic model, AWS will ask you to submit a brief use case description. Fill out the form describing that you are building a customer support chatbot for learning purposes, then submit.

Review the terms on the confirmation page, then click Submit.
Wait for the status to change to Access granted for both models. This usually takes a few minutes.

Why us-east-1?

The region you enable model access from is the only region where you can use those models. This project uses us-east-1 because it has the broadest model availability for Bedrock.

Verify model access is working by running this command in your terminal:

aws bedrock list-foundation-models --region us-east-1 --query "modelSummaries[?modelId=='amazon.titan-embed-text-v2:0'].modelId" --output text

You should see amazon.titan-embed-text-v2:0 in the output. This confirms your CLI has access to the Titan Embeddings model.

Seeing an empty response?

Model access can take a few minutes to propagate. Wait 2-3 minutes and try again. Also confirm you enabled the models in us-east-1 specifically, not another region.

help me troubleshoot Bedrock model access

✔️ Awesome, I've got everything!

Great. Double check that you've saved requirements.txt and that all packages installed without errors.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt file:

boto3
langchain
langchain-aws
langchain-community
langgraph
fastapi
mangum
psycopg2-binary
pgvector
pydantic
uvicorn

Your environment is fully configured. Next up, you'll write Terraform configuration to provision the entire AWS infrastructure stack in one command.

Deploy Infrastructure with Terraform

Your environment is ready. Now it is time to provision the actual AWS infrastructure your agent needs to run.

Production AI systems need multiple services working together. Your RAG agent needs a vector database for document retrieval, a conversation memory store, object storage for documents, and an API layer to expose it all. Terraform lets you define this entire stack as code and deploy it with a single command.

In this step, get ready to:

Write a Terraform configuration defining VPC networking, Aurora Serverless v2, DynamoDB, S3, IAM, and API Gateway.
Provision the full stack with a single command.
Verify all resources are created and capture their connection details.

Write the Terraform configuration

Your Terraform config will live in a dedicated terraform/ directory inside your project. Each file handles a different concern: variables for inputs, the main config for resources, and outputs for connection details.

Create the terraform directory and move into it by running these commands:

mkdir terraform
cd terraform

Confirm you are inside the terraform/ directory by running:

ls

The directory should be empty since you just created it.
Create the variables.tf file to define your input variables by running:

touch variables.tf

Open variables.tf in your editor and paste the following code:

variable "aws_region" {
  default = "us-east-1"
}

variable "db_password" {
  description = "Aurora master password"
  sensitive   = true
}

What does this code do?

aws_region defaults to us-east-1 so you do not need to pass it every time you run Terraform.
db_password is marked sensitive so Terraform hides it from terminal output. Terraform will prompt you for this value during apply.

Save variables.tf.

Now for the main configuration. You will build main.tf in three logical sections: networking, data stores, and access control.

Create main.tf by running:

touch main.tf

Open main.tf and paste the provider configuration and networking layer:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

data "aws_caller_identity" "current" {}

resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true
  tags = { Name = "rag-support-vpc" }
}

data "aws_availability_zones" "available" {}

resource "aws_subnet" "private" {
  count             = 2
  vpc_id            = aws_vpc.main.id
  cidr_block        = cidrsubnet(aws_vpc.main.cidr_block, 8, count.index)
  availability_zone = data.aws_availability_zones.available.names[count.index]
  tags = { Name = "rag-support-private-${count.index}" }
}

resource "aws_security_group" "aurora" {
  name   = "rag-aurora-sg"
  vpc_id = aws_vpc.main.id
  ingress {
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/16"]
  }
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_security_group" "lambda" {
  name   = "rag-lambda-sg"
  vpc_id = aws_vpc.main.id
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

What does this networking layer do?

The VPC creates an isolated network (10.0.0.0/16) for all your resources.
Two private subnets span different availability zones. Aurora requires at least two for high availability.
The Aurora security group only allows PostgreSQL traffic (port 5432) from within the VPC. The Lambda security group allows outbound traffic so your functions can reach Bedrock and other AWS services.

Below the networking code in main.tf, paste the database and storage resources:

resource "aws_rds_cluster" "aurora" {
  cluster_identifier     = "rag-support-cluster"
  engine                 = "aurora-postgresql"
  engine_version         = "15.4"
  database_name          = "ragdb"
  master_username        = "ragadmin"
  master_password        = var.db_password
  db_subnet_group_name   = aws_db_subnet_group.aurora.name
  vpc_security_group_ids = [aws_security_group.aurora.id]
  serverlessv2_scaling_configuration {
    min_capacity = 0.5
    max_capacity = 4.0
  }
  skip_final_snapshot = true
}

resource "aws_rds_cluster_instance" "aurora" {
  cluster_identifier = aws_rds_cluster.aurora.id
  instance_class     = "db.serverless"
  engine             = aws_rds_cluster.aurora.engine
  engine_version     = aws_rds_cluster.aurora.engine_version
}

resource "aws_db_subnet_group" "aurora" {
  name       = "rag-aurora-subnet-group"
  subnet_ids = aws_subnet.private[*].id
}

resource "aws_dynamodb_table" "conversation_memory" {
  name         = "rag-conversation-memory"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "session_id"
  range_key    = "id"
  attribute {
    name = "session_id"
    type = "S"
  }
  attribute {
    name = "id"
    type = "S"
  }
  ttl {
    attribute_name = "expires_at"
    enabled        = true
  }
}

resource "aws_s3_bucket" "documents" {
  bucket = "rag-support-documents-${data.aws_caller_identity.current.account_id}"
}

What does this data layer do?

Aurora Serverless v2 is your vector database. It runs PostgreSQL 15 with pgvector for storing document embeddings. The serverlessv2_scaling_configuration block sets it to scale between 0.5 and 4 ACUs (Aurora Capacity Units) based on demand.

DynamoDB stores conversation memory so your agent remembers past interactions across sessions. The TTL on expires_at auto-deletes old conversations.

The S3 bucket holds the source documents your agent retrieves from. The bucket name includes your AWS account ID to guarantee global uniqueness.

Below the storage resources in main.tf, paste the IAM and API Gateway configuration:

resource "aws_iam_role" "lambda_exec" {
  name = "rag-lambda-exec-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_policy" {
  name = "rag-lambda-policy"
  role = aws_iam_role.lambda_exec.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream", "bedrock:ApplyGuardrail"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Query", "dynamodb:Scan"]
        Resource = aws_dynamodb_table.conversation_memory.arn
      },
      {
        Effect = "Allow"
        Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
        Resource = [aws_s3_bucket.documents.arn, "${aws_s3_bucket.documents.arn}/*"]
      },
      {
        Effect = "Allow"
        Action = ["rds-data:ExecuteStatement", "rds-data:BatchExecuteStatement"]
        Resource = aws_rds_cluster.aurora.arn
      },
      {
        Effect = "Allow"
        Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
        Resource = "arn:aws:logs:*:*:*"
      },
      {
        Effect = "Allow"
        Action = ["ec2:CreateNetworkInterface", "ec2:DescribeNetworkInterfaces", "ec2:DeleteNetworkInterface"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["cloudwatch:PutMetricData"]
        Resource = "*"
      }
    ]
  })
}

resource "aws_api_gateway_rest_api" "main" {
  name = "rag-support-api"
}

What does this access layer do?

The IAM role gives your Lambda functions exactly the permissions they need. It can invoke Bedrock models, read and write to DynamoDB and S3, query Aurora, manage VPC network interfaces, and emit CloudWatch logs and metrics.

API Gateway creates the REST API endpoint that external users will call to chat with your agent. You will wire Lambda functions to it in a later step.

Save main.tf.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.tf with all three sections (networking, data stores, access control).

ⓧ I'd like to double check the full code

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

data "aws_caller_identity" "current" {}

resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true
  tags = { Name = "rag-support-vpc" }
}

data "aws_availability_zones" "available" {}

resource "aws_subnet" "private" {
  count             = 2
  vpc_id            = aws_vpc.main.id
  cidr_block        = cidrsubnet(aws_vpc.main.cidr_block, 8, count.index)
  availability_zone = data.aws_availability_zones.available.names[count.index]
  tags = { Name = "rag-support-private-${count.index}" }
}

resource "aws_security_group" "aurora" {
  name   = "rag-aurora-sg"
  vpc_id = aws_vpc.main.id
  ingress {
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/16"]
  }
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_security_group" "lambda" {
  name   = "rag-lambda-sg"
  vpc_id = aws_vpc.main.id
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_rds_cluster" "aurora" {
  cluster_identifier     = "rag-support-cluster"
  engine                 = "aurora-postgresql"
  engine_version         = "15.4"
  database_name          = "ragdb"
  master_username        = "ragadmin"
  master_password        = var.db_password
  db_subnet_group_name   = aws_db_subnet_group.aurora.name
  vpc_security_group_ids = [aws_security_group.aurora.id]
  serverlessv2_scaling_configuration {
    min_capacity = 0.5
    max_capacity = 4.0
  }
  skip_final_snapshot = true
}

resource "aws_rds_cluster_instance" "aurora" {
  cluster_identifier = aws_rds_cluster.aurora.id
  instance_class     = "db.serverless"
  engine             = aws_rds_cluster.aurora.engine
  engine_version     = aws_rds_cluster.aurora.engine_version
}

resource "aws_db_subnet_group" "aurora" {
  name       = "rag-aurora-subnet-group"
  subnet_ids = aws_subnet.private[*].id
}

resource "aws_dynamodb_table" "conversation_memory" {
  name         = "rag-conversation-memory"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "session_id"
  range_key    = "id"
  attribute {
    name = "session_id"
    type = "S"
  }
  attribute {
    name = "id"
    type = "S"
  }
  ttl {
    attribute_name = "expires_at"
    enabled        = true
  }
}

resource "aws_s3_bucket" "documents" {
  bucket = "rag-support-documents-${data.aws_caller_identity.current.account_id}"
}

resource "aws_iam_role" "lambda_exec" {
  name = "rag-lambda-exec-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_policy" {
  name = "rag-lambda-policy"
  role = aws_iam_role.lambda_exec.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream", "bedrock:ApplyGuardrail"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Query", "dynamodb:Scan"]
        Resource = aws_dynamodb_table.conversation_memory.arn
      },
      {
        Effect = "Allow"
        Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
        Resource = [aws_s3_bucket.documents.arn, "${aws_s3_bucket.documents.arn}/*"]
      },
      {
        Effect = "Allow"
        Action = ["rds-data:ExecuteStatement", "rds-data:BatchExecuteStatement"]
        Resource = aws_rds_cluster.aurora.arn
      },
      {
        Effect = "Allow"
        Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
        Resource = "arn:aws:logs:*:*:*"
      },
      {
        Effect = "Allow"
        Action = ["ec2:CreateNetworkInterface", "ec2:DescribeNetworkInterfaces", "ec2:DeleteNetworkInterface"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["cloudwatch:PutMetricData"]
        Resource = "*"
      }
    ]
  })
}

resource "aws_api_gateway_rest_api" "main" {
  name = "rag-support-api"
}

Finally, create the outputs file so Terraform captures the connection details you will need in later steps.

Create outputs.tf by running:

touch outputs.tf

Open outputs.tf and paste the following:

output "aurora_endpoint" {
  value = aws_rds_cluster.aurora.endpoint
}

output "s3_bucket_name" {
  value = aws_s3_bucket.documents.bucket
}

output "dynamodb_table_name" {
  value = aws_dynamodb_table.conversation_memory.name
}

output "api_gateway_id" {
  value = aws_api_gateway_rest_api.main.id
}

What does this code do?

Outputs extract key values from the resources Terraform creates. After terraform apply finishes, you can run terraform output to see the Aurora endpoint, S3 bucket name, DynamoDB table name, and API Gateway ID without digging through the AWS Console.

Save outputs.tf.

Seeing a syntax error when you save?

Check for mismatched curly braces {} or square brackets [] in main.tf. Every opening brace needs a matching close.
Make sure each resource block has exactly two strings after it (the type and the name), like resource "aws_vpc" "main".
If your editor shows red on the jsonencode blocks, that is likely a display-only issue. The file is valid HCL.

Help me debug a Terraform syntax error in my main.tf file

Provision the infrastructure

With all three files in place, you can initialize Terraform and deploy the stack. Terraform will download the AWS provider plugin, plan the changes, and create all resources in the correct dependency order.

Make sure you are inside the terraform/ directory, then initialize the workspace by running:

terraform init

You should see output ending with a confirmation that the AWS provider was downloaded and Terraform is ready.

Seeing a provider error during init?

Check that your required_providers block in main.tf has the correct source hashicorp/aws and version constraint ~> 5.0.
Make sure you have internet connectivity. Terraform needs to download the provider binary from the registry.

Help me troubleshoot a terraform init failure

Deploy all resources by running:

terraform apply

What happens during terraform apply?

Terraform will show you an execution plan listing every resource it intends to create. It then prompts you to confirm by typing yes.

It will also prompt you for the db_password variable. Enter a strong password (at least 8 characters, mix of letters and numbers). Remember this password for later steps.

The Aurora cluster takes the longest to create (typically 5-10 minutes). You will see progress messages while it provisions.

When prompted for the database password, type a strong password and press Enter.
When asked if you want to perform these actions, type yes and press Enter.
Wait for the apply to finish. You should see a completion message showing the number of resources added.

Apply failed with a permissions error?

Verify your AWS CLI credentials have administrator access or at minimum the permissions to create VPC, RDS, DynamoDB, S3, IAM, and API Gateway resources.
Check that your configured region is us-east-1. Some resources may fail if your CLI is pointed at a different region than what variables.tf specifies.
If you see an error about the Aurora engine version not being available, confirm your account supports Aurora PostgreSQL 15.4 in us-east-1.

Help me debug a failed terraform apply

Verify all resources are created

Once the apply completes, confirm that Terraform created everything and captured the connection details you need.

View your resource outputs by running:

terraform output

You should see four values printed: aurora_endpoint, s3_bucket_name, dynamodb_table_name, and api_gateway_id.

What should I see?

The output shows the connection details for each provisioned resource. Your Aurora endpoint will be a hostname like rag-support-cluster.cluster-xxxx.us-east-1.rds.amazonaws.com. The S3 bucket name includes your AWS account ID. The DynamoDB table name is rag-conversation-memory. The API Gateway ID is a short alphanumeric string.

These values confirm all resources were provisioned successfully. You will reference them in later steps when wiring up the agent code.

Missing outputs or empty values?

If terraform output shows nothing, confirm outputs.tf is saved in the same terraform/ directory as main.tf.
If a value is empty, the corresponding resource may not have finished creating. Run terraform apply again to check for any remaining issues.

Help me figure out why terraform output shows empty values

Your entire AWS infrastructure is now live. Next up, you will build the document ingestion pipeline that takes files from S3, chunks them, embeds them with Titan Embeddings v2, and stores the vectors in Aurora pgvector.

Build the Document Ingestion Pipeline

Your infrastructure is provisioned and ready. Now you need a way to get documents into the system so the agent can retrieve them later.

In production RAG systems, the ingestion path and query path are strictly separated. The ingestion path runs offline, processing documents into vector embeddings that get stored in the database. In this step, you will build that ingestion path as a Lambda function that automatically triggers whenever a document lands in your S3 bucket.

In this step, get ready to:

Create a Python Lambda that chunks, embeds, and stores documents in Aurora pgvector.
Deploy the Lambda with Terraform and wire up S3 event notifications.
Upload sample documents and verify embeddings are stored correctly.

Create the ingestion Lambda function

The ingestion Lambda handles three responsibilities: splitting documents into overlapping chunks, generating vector embeddings for each chunk using Titan Embeddings v2, and upserting those vectors into Aurora pgvector with HNSW indexing.

Create the lambda/ directory by running this command:

mkdir lambda

Create a new file called ingestion.py inside the lambda/ folder.
Add the imports, configuration, database connection, and schema initialization by pasting this code into lambda/ingestion.py:

import json
import os
import boto3
import psycopg2
import logging
import time
from typing import List

logger = logging.getLogger()
logger.setLevel(logging.INFO)

bedrock = boto3.client('bedrock-runtime', region_name=os.environ['BEDROCK_REGION'])

AURORA_HOST = os.environ['AURORA_HOST']
AURORA_DB = os.environ['AURORA_DB']
AURORA_USER = os.environ['AURORA_USER']
AURORA_PASSWORD = os.environ['AURORA_PASSWORD']
EMBEDDING_MODEL_ID = 'amazon.titan-embed-text-v2:0'
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100


def get_db_connection():
    return psycopg2.connect(
        host=AURORA_HOST,
        database=AURORA_DB,
        user=AURORA_USER,
        password=AURORA_PASSWORD
    )


def init_schema(conn):
    with conn.cursor() as cur:
        cur.execute('CREATE EXTENSION IF NOT EXISTS vector;')
        cur.execute('''
            CREATE TABLE IF NOT EXISTS document_chunks (
                id SERIAL PRIMARY KEY,
                doc_id TEXT NOT NULL,
                chunk_index INTEGER NOT NULL,
                content TEXT NOT NULL,
                embedding vector(1024),
                metadata JSONB,
                created_at TIMESTAMP DEFAULT NOW()
            );
        ''')
        cur.execute('''
            CREATE INDEX IF NOT EXISTS document_chunks_embedding_idx
            ON document_chunks USING hnsw (embedding vector_cosine_ops)
            WITH (m = 16, ef_construction = 64);
        ''')
        conn.commit()

What does this code do?

The configuration section reads database credentials and the Bedrock region from environment variables. These get injected by Lambda at runtime via the Terraform configuration you will add shortly.

The init_schema function enables the pgvector extension, creates the document_chunks table with a 1024-dimension vector column, and builds an HNSW index for fast approximate nearest-neighbor search.

Add the text processing functions below init_schema in the same file:


def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = ' '.join(words[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks


def embed_text(text: str) -> List[float]:
    start_time = time.time()
    response = bedrock.invoke_model(
        modelId=EMBEDDING_MODEL_ID,
        body=json.dumps({'inputText': text, 'dimensions': 1024, 'normalize': True})
    )
    embedding = json.loads(response['body'].read())['embedding']
    elapsed = time.time() - start_time
    logger.info(json.dumps({'event': 'embedding_complete', 'elapsed_ms': round(elapsed * 1000)}))
    return embedding

What does this code do?

The chunk_text function splits a document into 800-word chunks with a 100-word overlap. The overlap ensures context is not lost at chunk boundaries.

The embed_text function calls Titan Embeddings v2 via Bedrock to convert each text chunk into a 1024-dimension vector. It also logs the embedding latency for observability.

Add the persistence layer and Lambda handler below embed_text to complete the file:


def upsert_chunks(conn, doc_id: str, chunks: List[str], embeddings: List[List[float]], metadata: dict):
    with conn.cursor() as cur:
        cur.execute('DELETE FROM document_chunks WHERE doc_id = %s', (doc_id,))
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            cur.execute(
                'INSERT INTO document_chunks (doc_id, chunk_index, content, embedding, metadata) VALUES (%s, %s, %s, %s::vector, %s)',
                (doc_id, i, chunk, embedding, json.dumps(metadata))
            )
        conn.commit()


def handler(event, context):
    s3 = boto3.client('s3')
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        pipeline_start = time.time()
        logger.info(json.dumps({'event': 'ingestion_start', 'bucket': bucket, 'key': key}))
        obj = s3.get_object(Bucket=bucket, Key=key)
        text = obj['Body'].read().decode('utf-8')
        chunks = chunk_text(text)
        logger.info(json.dumps({'event': 'chunks_created', 'count': len(chunks), 'key': key}))
        embeddings = [embed_text(chunk) for chunk in chunks]
        conn = get_db_connection()
        try:
            init_schema(conn)
            upsert_chunks(conn, key, chunks, embeddings, {'source': key, 'bucket': bucket})
            total_elapsed = time.time() - pipeline_start
            logger.info(json.dumps({
                'event': 'ingestion_complete',
                'key': key,
                'chunks': len(chunks),
                'total_elapsed_ms': round(total_elapsed * 1000)
            }))
        finally:
            conn.close()
    return {'statusCode': 200}

What does this code do?

The upsert_chunks function deletes any existing chunks for the same document (so re-uploads replace stale data) and inserts each chunk with its embedding into pgvector.

The handler function is the Lambda entry point. It receives S3 event records, downloads each uploaded file, runs it through the chunking and embedding pipeline, and persists everything to Aurora. Structured JSON logs at each stage give you full observability.

Save lambda/ingestion.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before continuing.

ⓧ I'd like to double check the full code

import json
import os
import boto3
import psycopg2
import logging
import time
from typing import List

logger = logging.getLogger()
logger.setLevel(logging.INFO)

bedrock = boto3.client('bedrock-runtime', region_name=os.environ['BEDROCK_REGION'])

AURORA_HOST = os.environ['AURORA_HOST']
AURORA_DB = os.environ['AURORA_DB']
AURORA_USER = os.environ['AURORA_USER']
AURORA_PASSWORD = os.environ['AURORA_PASSWORD']
EMBEDDING_MODEL_ID = 'amazon.titan-embed-text-v2:0'
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100


def get_db_connection():
    return psycopg2.connect(
        host=AURORA_HOST,
        database=AURORA_DB,
        user=AURORA_USER,
        password=AURORA_PASSWORD
    )


def init_schema(conn):
    with conn.cursor() as cur:
        cur.execute('CREATE EXTENSION IF NOT EXISTS vector;')
        cur.execute('''
            CREATE TABLE IF NOT EXISTS document_chunks (
                id SERIAL PRIMARY KEY,
                doc_id TEXT NOT NULL,
                chunk_index INTEGER NOT NULL,
                content TEXT NOT NULL,
                embedding vector(1024),
                metadata JSONB,
                created_at TIMESTAMP DEFAULT NOW()
            );
        ''')
        cur.execute('''
            CREATE INDEX IF NOT EXISTS document_chunks_embedding_idx
            ON document_chunks USING hnsw (embedding vector_cosine_ops)
            WITH (m = 16, ef_construction = 64);
        ''')
        conn.commit()


def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = ' '.join(words[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks


def embed_text(text: str) -> List[float]:
    start_time = time.time()
    response = bedrock.invoke_model(
        modelId=EMBEDDING_MODEL_ID,
        body=json.dumps({'inputText': text, 'dimensions': 1024, 'normalize': True})
    )
    embedding = json.loads(response['body'].read())['embedding']
    elapsed = time.time() - start_time
    logger.info(json.dumps({'event': 'embedding_complete', 'elapsed_ms': round(elapsed * 1000)}))
    return embedding


def upsert_chunks(conn, doc_id: str, chunks: List[str], embeddings: List[List[float]], metadata: dict):
    with conn.cursor() as cur:
        cur.execute('DELETE FROM document_chunks WHERE doc_id = %s', (doc_id,))
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            cur.execute(
                'INSERT INTO document_chunks (doc_id, chunk_index, content, embedding, metadata) VALUES (%s, %s, %s, %s::vector, %s)',
                (doc_id, i, chunk, embedding, json.dumps(metadata))
            )
        conn.commit()


def handler(event, context):
    s3 = boto3.client('s3')
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        pipeline_start = time.time()
        logger.info(json.dumps({'event': 'ingestion_start', 'bucket': bucket, 'key': key}))
        obj = s3.get_object(Bucket=bucket, Key=key)
        text = obj['Body'].read().decode('utf-8')
        chunks = chunk_text(text)
        logger.info(json.dumps({'event': 'chunks_created', 'count': len(chunks), 'key': key}))
        embeddings = [embed_text(chunk) for chunk in chunks]
        conn = get_db_connection()
        try:
            init_schema(conn)
            upsert_chunks(conn, key, chunks, embeddings, {'source': key, 'bucket': bucket})
            total_elapsed = time.time() - pipeline_start
            logger.info(json.dumps({
                'event': 'ingestion_complete',
                'key': key,
                'chunks': len(chunks),
                'total_elapsed_ms': round(total_elapsed * 1000)
            }))
        finally:
            conn.close()
    return {'statusCode': 200}

Why two separate paths?

Production RAG architectures strictly separate the ingestion path (offline, batch) from the query path (online, latency-critical). The ingestion path you just built can run in the background on every new document upload without affecting query response times.

Deploy the Lambda and upload sample documents

With the Lambda code written, you need to add the infrastructure that deploys it and wires up the S3 trigger. You will also create a sample document to test the full pipeline.

In terraform/main.tf, find the aws_s3_bucket "documents" resource block.
Add the S3 bucket notification, Lambda function, and Lambda permission resources below it by pasting this code after the aws_s3_bucket block:

resource "aws_s3_bucket_notification" "documents" {
  bucket = aws_s3_bucket.documents.id
  lambda_function {
    lambda_function_arn = aws_lambda_function.ingestion.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "documents/"
  }
  depends_on = [aws_lambda_permission.s3_invoke]
}

What does this do?

This notification resource tells S3 to invoke your Lambda whenever a new object is created under the documents/ prefix. The depends_on ensures the Lambda permission exists before S3 tries to call it.

Next, find the aws_iam_role_policy "lambda_policy" resource block.
Add the Lambda function and permission resources after it:

resource "aws_lambda_function" "ingestion" {
  function_name = "rag-document-ingestion"
  role          = aws_iam_role.lambda_exec.arn
  handler       = "ingestion.handler"
  runtime       = "python3.12"
  timeout       = 300
  memory_size   = 1024
  filename      = "../lambda/ingestion.zip"
  vpc_config {
    subnet_ids         = aws_subnet.private[*].id
    security_group_ids = [aws_security_group.lambda.id]
  }
  environment {
    variables = {
      AURORA_HOST     = aws_rds_cluster.aurora.endpoint
      AURORA_DB       = "ragdb"
      AURORA_USER     = "ragadmin"
      AURORA_PASSWORD = var.db_password
      BEDROCK_REGION  = var.aws_region
    }
  }
}

resource "aws_lambda_permission" "s3_invoke" {
  statement_id  = "AllowS3Invoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.ingestion.function_name
  principal     = "s3.amazonaws.com"
  source_arn    = aws_s3_bucket.documents.arn
}

What does this configuration do?

The Lambda function is deployed into the private subnets of your VPC so it can reach Aurora directly over port 5432. The 300-second timeout and 1024 MB of memory give it headroom for embedding large documents.

The permission resource grants S3 the right to invoke this specific Lambda. Without it, the S3 notification would be rejected.

Save terraform/main.tf.

Terraform plan showing errors?

Make sure the aws_s3_bucket_notification block is placed after aws_s3_bucket "documents" and the Lambda resources are after the IAM policy block. Terraform does not require strict ordering, but circular references will cause a plan failure.

help me fix Terraform dependency errors in my RAG project

Create the sample_docs/ directory and sample FAQ file by running:

mkdir sample_docs

Create a file called support_faq.txt inside sample_docs/ and paste the following content:

# Customer Support FAQ

## Returns and Refunds
Our return policy allows returns within 30 days of purchase. Items must be in original condition. Refunds are processed within 5-7 business days to the original payment method.

## Shipping
Standard shipping takes 5-7 business days. Express shipping (2-3 days) is available for an additional fee. Free shipping on orders over $50.

## Account Issues
If you cannot log in, use the Forgot Password link on the login page. For account lockouts, contact support@example.com with your account email.

## Product Warranty
All products carry a 1-year manufacturer warranty. Extended warranties are available for purchase within 30 days of buying the product.

Save sample_docs/support_faq.txt.
Package the Lambda code into a zip file by running these commands:

cd lambda
pip install psycopg2-binary pgvector -t . --platform manylinux2014_x86_64 --only-binary=:all:
zip -r ingestion.zip .
cd ..

Why the platform flag?

Lambda runs on Amazon Linux (x86_64). The --platform manylinux2014_x86_64 flag ensures pip downloads Linux-compatible compiled binaries for psycopg2-binary, even if you are building on macOS or Windows.

Deploy the updated infrastructure by running:

cd terraform
terraform apply -auto-approve
cd ..

Wait for Terraform to finish. You should see the Lambda function and S3 notification in the output.
Upload the sample document to S3 under the documents/ prefix by running:

aws s3 cp sample_docs/support_faq.txt s3://[[S3_BUCKET="your-s3-bucket-name-from-terraform-output"]]/documents/support_faq.txt

The upload triggers the ingestion Lambda automatically. Give it about 30 seconds to process.

Upload succeeded but nothing happens?

Check that your file path starts with documents/ since the S3 notification only fires for objects under that prefix. Also confirm the Lambda function deployed successfully by checking the AWS Lambda console.

help me debug why my S3 upload is not triggering the ingestion Lambda

Verify embeddings are stored in Aurora

The ingestion Lambda emits structured JSON logs at every stage. You can verify the pipeline worked by checking CloudWatch Logs for the completion event.

Check the Lambda logs for the ingestion completion event by running:

aws logs filter-log-events --log-group-name /aws/lambda/rag-document-ingestion --filter-pattern "ingestion_complete"

You should see a JSON log entry with "event": "ingestion_complete", a "chunks" count (likely 1 for the small FAQ file), and a "total_elapsed_ms" value.

What should I see?

A successful ingestion shows log events like ingestion_start, chunks_created, embedding_complete, and ingestion_complete. The ingestion_complete event confirms chunks were embedded and stored in Aurora pgvector. If you see all four events, your ingestion pipeline is working end to end.

Not seeing any log events?

Lambda logs can take 15-30 seconds to appear in CloudWatch. Wait a moment and retry the command. If the log group does not exist at all, the Lambda was never invoked. Verify the S3 notification is configured correctly by checking the S3 bucket properties in the AWS Console under the Event notifications section.

help me debug why my ingestion Lambda logs are not appearing in CloudWatch

Your ingestion pipeline is live. Documents uploaded to S3 are now automatically chunked, embedded, and stored in Aurora pgvector. Next up, you will build the LangGraph agent that queries these embeddings to answer user questions.

Build the LangGraph Agent with Tool Calling & Memory

Your document ingestion pipeline is live and your vector database is loaded with embedded support content. Now it is time to build the brain of the system.

In this step, you will create a LangGraph agent that orchestrates the entire query path. The agent decides when to retrieve documents from Aurora, when to create a support ticket, and how to synthesize a grounded answer using Claude Haiku 4.5 on Bedrock.

In this step, get ready to:

Create a LangGraph StateGraph with a RAG retrieval tool and a ticket creation tool.
Wire up DynamoDBSaver for persistent conversation memory across sessions.
Test the agent locally with multi-turn conversations.

Create the agent file with imports and state definition

The agent lives in its own file separate from the ingestion Lambda. It needs access to the same Aurora database for retrieval, plus DynamoDB for both tickets and conversation memory.

Create a new file called agent.py inside your lambda folder by running:

touch lambda/agent.py

Open lambda/agent.py and add the imports, environment configuration, and agent state definition:

import json
import os
import time
import logging
import boto3
import psycopg2
from typing import Annotated, List
from langchain_aws import ChatBedrock
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.dynamodb import DynamoDBSaver
from typing_extensions import TypedDict

logger = logging.getLogger()
logger.setLevel(logging.INFO)

AURORA_HOST = os.environ['AURORA_HOST']
AURORA_DB = os.environ['AURORA_DB']
AURORA_USER = os.environ['AURORA_USER']
AURORA_PASSWORD = os.environ['AURORA_PASSWORD']
BEDROCK_REGION = os.environ.get('BEDROCK_REGION', 'us-east-1')
DYNAMODB_TABLE = os.environ['DYNAMODB_TABLE']
EMBEDDING_MODEL_ID = 'amazon.titan-embed-text-v2:0'
LLM_MODEL_ID = 'anthropic.claude-haiku-4-5'

bedrock_runtime = boto3.client('bedrock-runtime', region_name=BEDROCK_REGION)
dynamodb = boto3.resource('dynamodb', region_name=BEDROCK_REGION)


# Define the agent's state shape
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    session_id: str


# Reuse the same connection pattern from ingestion
def get_db_connection():
    return psycopg2.connect(
        host=AURORA_HOST, database=AURORA_DB,
        user=AURORA_USER, password=AURORA_PASSWORD
    )

What does this code do?

This sets up the foundation for your agent. The key pieces are:

AgentState defines what data flows through the graph. messages holds the conversation history and session_id identifies the user session.
add_messages is a LangGraph reducer that appends new messages to existing ones instead of overwriting.
LLM_MODEL_ID points to Claude Haiku 4.5, which gives the best cost-to-quality ratio for a customer support bot.

Save lambda/agent.py.

Add the RAG retrieval tool

LangGraph agents use tools to interact with external systems. Your first tool queries the vector database you populated in the previous step and returns the most relevant document chunks.

In lambda/agent.py, add the embedding helper and the RAG retrieval tool below the get_db_connection() function:

def embed_query(text: str) -> List[float]:
    response = bedrock_runtime.invoke_model(
        modelId=EMBEDDING_MODEL_ID,
        body=json.dumps({'inputText': text, 'dimensions': 1024, 'normalize': True})
    )
    return json.loads(response['body'].read())['embedding']


@tool
def rag_retrieval_tool(query: str) -> str:
    """Retrieve relevant customer support documentation chunks for a given query."""
    start = time.time()
    query_embedding = embed_query(query)
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                '''
                SELECT content, metadata, 1 - (embedding <=> %s::vector) AS similarity
                FROM document_chunks
                ORDER BY embedding <=> %s::vector
                LIMIT 5
                ''',
                (query_embedding, query_embedding)
            )
            rows = cur.fetchall()
    finally:
        conn.close()
    elapsed = time.time() - start
    logger.info(json.dumps({'event': 'retrieval_complete', 'elapsed_ms': round(elapsed * 1000), 'results': len(rows)}))
    if not rows:
        return 'No relevant documentation found.'
    results = []
    for content, metadata, similarity in rows:
        source = metadata.get('source', 'unknown') if metadata else 'unknown'
        results.append(f'[Source: {source}, Similarity: {similarity:.3f}]\n{content}')
    return '\n\n---\n\n'.join(results)

What does this code do?

embed_query() converts the user's question into a 1024-dimensional vector using Titan Embeddings v2. This is the same embedding model you used during ingestion, which ensures the vectors are comparable.

rag_retrieval_tool() uses the @tool decorator so LangGraph can invoke it automatically. It queries Aurora pgvector using cosine similarity (the <=> operator), returns the top 5 chunks, and includes source citations and similarity scores.

Why cosine similarity?

The <=> operator computes cosine distance in pgvector. Subtracting from 1 gives you cosine similarity, where higher values mean more relevant results. This pairs with the HNSW index you created during ingestion for fast approximate nearest-neighbor lookups.

Save lambda/agent.py.

Seeing a syntax error in the SQL?

Make sure the triple-quoted SQL string uses consistent indentation. The <=> operator is pgvector-specific and requires the vector extension to be installed (which your ingestion Lambda already handled).

help me debug a pgvector cosine similarity query syntax error

Add the ticket creation tool

Not every customer question can be answered from documentation. When the agent cannot resolve an issue, it needs a way to escalate by creating a support ticket.

In lambda/agent.py, add the ticket creation tool below rag_retrieval_tool:

@tool
def create_ticket_tool(subject: str, description: str, priority: str = 'medium') -> str:
    """Create a support ticket in DynamoDB for issues that cannot be resolved via documentation."""
    import uuid
    ticket_id = str(uuid.uuid4())
    table = dynamodb.Table(DYNAMODB_TABLE)
    table.put_item(Item={
        'session_id': 'tickets',
        'id': ticket_id,
        'subject': subject,
        'description': description,
        'priority': priority,
        'status': 'open',
        'created_at': str(time.time())
    })
    logger.info(json.dumps({'event': 'ticket_created', 'ticket_id': ticket_id}))
    return f'Support ticket created with ID: {ticket_id}. Priority: {priority}.'

What does this code do?

This tool writes a support ticket directly to the same DynamoDB table your infrastructure already provisioned. It uses session_id: 'tickets' as the partition key to separate ticket records from conversation memory records.

The @tool decorator exposes the function's docstring to the LLM, so Claude knows when and how to call it. The LLM reads the parameter names and docstring to decide when escalation is appropriate.

Save lambda/agent.py.

Build the LLM node and graph structure

With both tools defined, you can now wire them into a LangGraph StateGraph. The graph has two nodes: an agent node (calls the LLM) and a tools node (executes tool calls). A conditional edge decides whether to loop back through the tools or finish.

In lambda/agent.py, add the LLM setup, agent node, routing logic, and graph construction below create_ticket_tool:

tools = [rag_retrieval_tool, create_ticket_tool]
llm = ChatBedrock(model_id=LLM_MODEL_ID, region_name=BEDROCK_REGION)
llm_with_tools = llm.bind_tools(tools)

SYSTEM_PROMPT = """You are a helpful customer support agent. Use the rag_retrieval_tool to find relevant information before answering questions. If you cannot resolve an issue, use create_ticket_tool to escalate. Always cite your sources."""


def agent_node(state: AgentState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state['messages']
    start = time.time()
    response = llm_with_tools.invoke(messages)
    elapsed = time.time() - start
    logger.info(json.dumps({'event': 'llm_response', 'elapsed_ms': round(elapsed * 1000)}))
    return {'messages': [response]}


def should_continue(state: AgentState):
    last_message = state['messages'][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return 'tools'
    return END


tool_node = ToolNode(tools)

graph_builder = StateGraph(AgentState)
graph_builder.add_node('agent', agent_node)
graph_builder.add_node('tools', tool_node)
graph_builder.set_entry_point('agent')
graph_builder.add_conditional_edges('agent', should_continue)
graph_builder.add_edge('tools', 'agent')

What does this code do?

This is the core agent loop. Here is how the pieces fit together:

bind_tools() tells Claude about your available tools so it can generate structured tool calls in its responses.
agent_node() prepends the system prompt, invokes the LLM, and logs response latency.
should_continue() checks if the LLM wants to call a tool. If yes, route to the tools node. If no, end the graph.
The tools -> agent edge means after a tool executes, results flow back to the LLM for another decision. This creates a ReAct-style loop.

What is a ReAct agent?

ReAct stands for Reasoning + Acting. The agent reasons about what to do, acts by calling a tool, observes the result, then reasons again. This loop repeats until the agent has enough information to produce a final answer.

Save lambda/agent.py.

Import errors when saving?

If your editor shows unresolved imports for langchain_aws or langgraph, make sure you installed the dependencies from requirements.txt in your active Python environment. Run pip install -r requirements.txt if needed.

help me resolve langgraph or langchain import errors in my agent.py

Wire up DynamoDBSaver for persistent conversation memory

Without persistence, your agent forgets everything between requests. DynamoDBSaver is the production checkpointer that stores conversation state in DynamoDB, so users can pick up where they left off across sessions.

At the bottom of lambda/agent.py, add the checkpointer and compile the graph:

checkpointer = DynamoDBSaver(table_name=DYNAMODB_TABLE)
app = graph_builder.compile(checkpointer=checkpointer)

What does this code do?

DynamoDBSaver saves the full graph state (messages, tool results, pending actions) to DynamoDB after every node execution. When a user sends a follow-up message, the checkpointer reloads the prior state so the agent has full conversation context.

compile() turns your graph definition into a runnable application. Passing the checkpointer here means every invocation automatically reads and writes conversation state.

Why DynamoDBSaver over MemorySaver?

MemorySaver only works in a single process and loses everything on restart. DynamoDBSaver persists across Lambda cold starts, scales horizontally, and supports TTL-based auto-cleanup for stale sessions. This is the production pattern recommended by AWS for serverless agents.

Save lambda/agent.py.

Test the agent locally with multi-turn conversations

Before deploying, you should verify the agent can retrieve documents, call tools, and remember context across turns. You will invoke the compiled graph directly from your terminal.

Set the required environment variables for local testing by running:

export AURORA_HOST=$(cd terraform && terraform output -raw aurora_endpoint)
export AURORA_DB=ragdb
export AURORA_USER=ragadmin
export AURORA_PASSWORD=your-db-password-here
export BEDROCK_REGION=us-east-1
export DYNAMODB_TABLE=rag-conversation-memory

What do these variables do?

These are the same environment variables your Lambda uses in production. Setting them locally lets you run the agent code directly without deploying. Replace your-db-password-here with the Aurora password you set in your Terraform variables.

Test the agent with a multi-turn conversation by running this Python script:

python3 -c "
from lambda.agent import app
from langchain_core.messages import HumanMessage

config = {'configurable': {'thread_id': 'test-session-1'}}

# First turn: ask about returns
result = app.invoke({'messages': [HumanMessage(content='What is your return policy?')], 'session_id': 'test-session-1'}, config)
print('Turn 1:', result['messages'][-1].content)

# Second turn: follow-up (tests memory)
result = app.invoke({'messages': [HumanMessage(content='How long do refunds take?')], 'session_id': 'test-session-1'}, config)
print('Turn 2:', result['messages'][-1].content)

# Third turn: trigger ticket creation
result = app.invoke({'messages': [HumanMessage(content='I want to report a defective product, order #12345')], 'session_id': 'test-session-1'}, config)
print('Turn 3:', result['messages'][-1].content)
"

You should see three responses printed in your terminal. Turn 1 should reference the 30-day return policy from your FAQ document. Turn 2 should mention 5-7 business days without you needing to repeat context. Turn 3 should confirm a support ticket was created.

Agent not returning grounded answers?

If the agent gives generic answers without citing your documents, check that your ingestion Lambda ran successfully and the document_chunks table has data. You can verify by connecting to Aurora and running SELECT COUNT(*) FROM document_chunks;.

If you see a ConnectionRefused error, make sure your local machine can reach the Aurora cluster (you may need a bastion host or VPN since Aurora is in a private subnet).

help me debug why my LangGraph agent is not retrieving documents from Aurora pgvector

✔️ Awesome, I've got everything!

Great. Double check you have saved lambda/agent.py with all the code from this step.

ⓧ I'd like to double check the full code

Here is the complete lambda/agent.py file:

import json
import os
import time
import logging
import boto3
import psycopg2
from typing import Annotated, List
from langchain_aws import ChatBedrock
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.dynamodb import DynamoDBSaver
from typing_extensions import TypedDict

logger = logging.getLogger()
logger.setLevel(logging.INFO)

AURORA_HOST = os.environ['AURORA_HOST']
AURORA_DB = os.environ['AURORA_DB']
AURORA_USER = os.environ['AURORA_USER']
AURORA_PASSWORD = os.environ['AURORA_PASSWORD']
BEDROCK_REGION = os.environ.get('BEDROCK_REGION', 'us-east-1')
DYNAMODB_TABLE = os.environ['DYNAMODB_TABLE']
EMBEDDING_MODEL_ID = 'amazon.titan-embed-text-v2:0'
LLM_MODEL_ID = 'anthropic.claude-haiku-4-5'

bedrock_runtime = boto3.client('bedrock-runtime', region_name=BEDROCK_REGION)
dynamodb = boto3.resource('dynamodb', region_name=BEDROCK_REGION)


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    session_id: str


def get_db_connection():
    return psycopg2.connect(
        host=AURORA_HOST, database=AURORA_DB,
        user=AURORA_USER, password=AURORA_PASSWORD
    )


def embed_query(text: str) -> List[float]:
    response = bedrock_runtime.invoke_model(
        modelId=EMBEDDING_MODEL_ID,
        body=json.dumps({'inputText': text, 'dimensions': 1024, 'normalize': True})
    )
    return json.loads(response['body'].read())['embedding']


@tool
def rag_retrieval_tool(query: str) -> str:
    """Retrieve relevant customer support documentation chunks for a given query."""
    start = time.time()
    query_embedding = embed_query(query)
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                '''
                SELECT content, metadata, 1 - (embedding <=> %s::vector) AS similarity
                FROM document_chunks
                ORDER BY embedding <=> %s::vector
                LIMIT 5
                ''',
                (query_embedding, query_embedding)
            )
            rows = cur.fetchall()
    finally:
        conn.close()
    elapsed = time.time() - start
    logger.info(json.dumps({'event': 'retrieval_complete', 'elapsed_ms': round(elapsed * 1000), 'results': len(rows)}))
    if not rows:
        return 'No relevant documentation found.'
    results = []
    for content, metadata, similarity in rows:
        source = metadata.get('source', 'unknown') if metadata else 'unknown'
        results.append(f'[Source: {source}, Similarity: {similarity:.3f}]\n{content}')
    return '\n\n---\n\n'.join(results)


@tool
def create_ticket_tool(subject: str, description: str, priority: str = 'medium') -> str:
    """Create a support ticket in DynamoDB for issues that cannot be resolved via documentation."""
    import uuid
    ticket_id = str(uuid.uuid4())
    table = dynamodb.Table(DYNAMODB_TABLE)
    table.put_item(Item={
        'session_id': 'tickets',
        'id': ticket_id,
        'subject': subject,
        'description': description,
        'priority': priority,
        'status': 'open',
        'created_at': str(time.time())
    })
    logger.info(json.dumps({'event': 'ticket_created', 'ticket_id': ticket_id}))
    return f'Support ticket created with ID: {ticket_id}. Priority: {priority}.'


tools = [rag_retrieval_tool, create_ticket_tool]
llm = ChatBedrock(model_id=LLM_MODEL_ID, region_name=BEDROCK_REGION)
llm_with_tools = llm.bind_tools(tools)

SYSTEM_PROMPT = """You are a helpful customer support agent. Use the rag_retrieval_tool to find relevant information before answering questions. If you cannot resolve an issue, use create_ticket_tool to escalate. Always cite your sources."""


def agent_node(state: AgentState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state['messages']
    start = time.time()
    response = llm_with_tools.invoke(messages)
    elapsed = time.time() - start
    logger.info(json.dumps({'event': 'llm_response', 'elapsed_ms': round(elapsed * 1000)}))
    return {'messages': [response]}


def should_continue(state: AgentState):
    last_message = state['messages'][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return 'tools'
    return END


tool_node = ToolNode(tools)

graph_builder = StateGraph(AgentState)
graph_builder.add_node('agent', agent_node)
graph_builder.add_node('tools', tool_node)
graph_builder.set_entry_point('agent')
graph_builder.add_conditional_edges('agent', should_continue)
graph_builder.add_edge('tools', 'agent')

checkpointer = DynamoDBSaver(table_name=DYNAMODB_TABLE)
app = graph_builder.compile(checkpointer=checkpointer)

Your agent is wired up and tested. Next up, you will add Bedrock Guardrails for security and deploy the whole thing behind an API Gateway so real users can interact with it.

Add Bedrock Guardrails & Deploy the API

Your agent can retrieve documents and create tickets, but it has no security boundary. A user could paste a credit card number into the chat, ask off-topic questions, or the model itself might leak sensitive data in its responses.

In this step, you will add a Bedrock Guardrail that filters PII and blocks off-topic requests. You will then wrap the agent in a query handler Lambda, connect it to API Gateway, and test the full end-to-end flow.

In this step, get ready to:

Create a Bedrock Guardrail for PII filtering and topic boundaries.
Deploy the agent as a Lambda behind API Gateway.
Test the full end-to-end flow from API request to grounded answer.

Create a Bedrock Guardrail for PII filtering and topic boundaries

Bedrock Guardrails let you define security policies that run independently of the foundation model. You can filter PII (emails, credit card numbers, social security numbers) and deny off-topic requests before the LLM ever sees them.

In terraform/main.tf, scroll to the bottom of the file (below the aws_api_gateway_rest_api resource).
Add the Bedrock Guardrail resource by pasting the following code:

resource "aws_bedrock_guardrail" "main" {
  name                      = "rag-support-guardrail"
  description               = "PII filtering and topic boundaries for customer support agent"
  blocked_input_messaging   = "I cannot process that request."
  blocked_outputs_messaging = "I cannot provide that information."
  sensitive_information_policy_config {
    pii_entities_config {
      type   = "EMAIL"
      action = "ANONYMIZE"
    }
    pii_entities_config {
      type   = "CREDIT_DEBIT_CARD_NUMBER"
      action = "BLOCK"
    }
    pii_entities_config {
      type   = "US_SOCIAL_SECURITY_NUMBER"
      action = "BLOCK"
    }
  }
  topic_policy_config {
    topics_config {
      name       = "off-topic"
      definition = "Requests unrelated to customer support, product questions, or account management."
      examples   = ["Tell me a joke", "What is the weather today?"]
      type       = "DENY"
    }
  }
}

What does this code do?

This Terraform resource creates a Bedrock Guardrail with two policy layers.

sensitive_information_policy_config defines how PII is handled. Emails get anonymized (replaced with a placeholder). Credit card numbers and social security numbers are blocked entirely.
topic_policy_config defines which topics the agent should refuse. Any request unrelated to customer support, products, or account management gets denied before reaching the LLM.

Save terraform/main.tf.

You will not see a change yet. The next chunk adds the query Lambda and API Gateway resources, then you will deploy everything together.

Deploy the agent as a Lambda behind API Gateway

The query handler Lambda wraps your LangGraph agent with guardrail checks on both input and output. It receives chat messages from API Gateway, validates them against the guardrail, invokes the agent, and validates the response before returning it to the user.

Create a new file called query.py inside your lambda/ folder.
Paste the following code into lambda/query.py:

import json
import os
import time
import logging
import boto3
from langchain_core.messages import HumanMessage
from agent import app

logger = logging.getLogger()
logger.setLevel(logging.INFO)

bedrock_client = boto3.client('bedrock-runtime', region_name=os.environ.get('BEDROCK_REGION', 'us-east-1'))
GUARDRAIL_ID = os.environ['GUARDRAIL_ID']
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION', 'DRAFT')


def apply_guardrail(text: str, source: str = 'INPUT') -> dict:
    # Call the Bedrock ApplyGuardrail API to check text against PII and topic policies
    response = bedrock_client.apply_guardrail(
        guardrailIdentifier=GUARDRAIL_ID,
        guardrailVersion=GUARDRAIL_VERSION,
        source=source,
        content=[{'text': {'text': text}}]
    )
    return response


def handler(event, context):
    start_time = time.time()
    try:
        body = json.loads(event.get('body', '{}'))
        message = body.get('message', '')
        session_id = body.get('session_id', 'default')

        # Check user input against guardrail before invoking agent
        guardrail_result = apply_guardrail(message, source='INPUT')
        if guardrail_result.get('action') == 'GUARDRAIL_INTERVENED':
            logger.info(json.dumps({'event': 'guardrail_blocked', 'session_id': session_id}))
            return {
                'statusCode': 200,
                'headers': {'Content-Type': 'application/json'},
                'body': json.dumps({'response': 'I cannot process that request.', 'blocked': True})
            }

        # Invoke the LangGraph agent with conversation memory
        config = {'configurable': {'thread_id': session_id}}
        result = app.invoke(
            {'messages': [HumanMessage(content=message)], 'session_id': session_id},
            config=config
        )

        final_message = result['messages'][-1].content

        # Check agent output against guardrail before returning to user
        output_guardrail = apply_guardrail(final_message, source='OUTPUT')
        if output_guardrail.get('action') == 'GUARDRAIL_INTERVENED':
            final_message = 'I cannot provide that information.'

        elapsed = time.time() - start_time
        logger.info(json.dumps({
            'event': 'query_complete',
            'session_id': session_id,
            'elapsed_ms': round(elapsed * 1000)
        }))

        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'response': final_message, 'session_id': session_id})
        }
    except Exception as e:
        logger.error(json.dumps({'event': 'query_error', 'error': str(e)}))
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

What does this code do?

This Lambda handler sits between the user and your LangGraph agent, enforcing guardrails on both sides of the conversation.

apply_guardrail() calls the Bedrock ApplyGuardrail API independently of the model. It evaluates text against your guardrail's PII and topic policies.
The handler first checks the user's input. If the guardrail intervenes (GUARDRAIL_INTERVENED), the request is rejected before invoking the agent.
After the agent responds, the output is also checked. If the agent accidentally leaks PII, the guardrail replaces the response with a safe fallback message.
from agent import app imports the compiled LangGraph agent from agent.py, so both files must be packaged together in the Lambda deployment.

Save lambda/query.py.

You will not see a change yet. The next chunk adds the Terraform resources to deploy this Lambda and wire up the API.

Switch back to terraform/main.tf.
Below the aws_bedrock_guardrail resource you just added, paste the query Lambda and API Gateway resources:

resource "aws_lambda_function" "query" {
  function_name = "rag-query-handler"
  role          = aws_iam_role.lambda_exec.arn
  handler       = "query.handler"
  runtime       = "python3.12"
  timeout       = 60
  memory_size   = 1024
  filename      = "../lambda/query.zip"
  vpc_config {
    subnet_ids         = aws_subnet.private[*].id
    security_group_ids = [aws_security_group.lambda.id]
  }
  environment {
    variables = {
      AURORA_HOST       = aws_rds_cluster.aurora.endpoint
      AURORA_DB         = "ragdb"
      AURORA_USER       = "ragadmin"
      AURORA_PASSWORD   = var.db_password
      BEDROCK_REGION    = var.aws_region
      DYNAMODB_TABLE    = aws_dynamodb_table.conversation_memory.name
      GUARDRAIL_ID      = aws_bedrock_guardrail.main.guardrail_id
      GUARDRAIL_VERSION = "DRAFT"
    }
  }
}

resource "aws_api_gateway_resource" "chat" {
  rest_api_id = aws_api_gateway_rest_api.main.id
  parent_id   = aws_api_gateway_rest_api.main.root_resource_id
  path_part   = "chat"
}

resource "aws_api_gateway_method" "chat_post" {
  rest_api_id   = aws_api_gateway_rest_api.main.id
  resource_id   = aws_api_gateway_resource.chat.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "chat" {
  rest_api_id             = aws_api_gateway_rest_api.main.id
  resource_id             = aws_api_gateway_resource.chat.id
  http_method             = aws_api_gateway_method.chat_post.http_method
  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.query.invoke_arn
}

resource "aws_api_gateway_deployment" "main" {
  rest_api_id = aws_api_gateway_rest_api.main.id
  stage_name  = "prod"
  depends_on  = [aws_api_gateway_integration.chat]
}

resource "aws_lambda_permission" "apigw_invoke" {
  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.main.execution_arn}/*/*"
}

What does this code do?

This block provisions the full API deployment layer.

aws_lambda_function.query deploys your query.py handler. It receives the guardrail ID as an environment variable so the Python code can call apply_guardrail() at runtime.
aws_api_gateway_resource.chat creates a /chat path on your REST API.
aws_api_gateway_integration.chat uses AWS_PROXY integration so API Gateway passes the full HTTP request to your Lambda and returns its response directly.
aws_api_gateway_deployment.main deploys everything to a prod stage, giving you a live URL.
aws_lambda_permission.apigw_invoke grants API Gateway permission to invoke the query Lambda.

Now open terraform/outputs.tf and add these two outputs at the bottom:

output "api_endpoint" {
  value = "${aws_api_gateway_deployment.main.invoke_url}/chat"
}

output "guardrail_id" {
  value = aws_bedrock_guardrail.main.guardrail_id
}

Save both terraform/main.tf and terraform/outputs.tf.

✔️ Awesome, I've got everything!

Great. Double check you have saved both Terraform files and lambda/query.py.

ⓧ I'd like to double check the full code

Here is the complete terraform/main.tf at this point:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

data "aws_caller_identity" "current" {}

resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true
  tags = { Name = "rag-support-vpc" }
}

data "aws_availability_zones" "available" {}

resource "aws_subnet" "private" {
  count             = 2
  vpc_id            = aws_vpc.main.id
  cidr_block        = cidrsubnet(aws_vpc.main.cidr_block, 8, count.index)
  availability_zone = data.aws_availability_zones.available.names[count.index]
  tags = { Name = "rag-support-private-${count.index}" }
}

resource "aws_security_group" "aurora" {
  name   = "rag-aurora-sg"
  vpc_id = aws_vpc.main.id
  ingress {
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/16"]
  }
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_security_group" "lambda" {
  name   = "rag-lambda-sg"
  vpc_id = aws_vpc.main.id
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_rds_cluster" "aurora" {
  cluster_identifier     = "rag-support-cluster"
  engine                 = "aurora-postgresql"
  engine_version         = "15.4"
  database_name          = "ragdb"
  master_username        = "ragadmin"
  master_password        = var.db_password
  db_subnet_group_name   = aws_db_subnet_group.aurora.name
  vpc_security_group_ids = [aws_security_group.aurora.id]
  serverlessv2_scaling_configuration {
    min_capacity = 0.5
    max_capacity = 4.0
  }
  skip_final_snapshot = true
}

resource "aws_rds_cluster_instance" "aurora" {
  cluster_identifier = aws_rds_cluster.aurora.id
  instance_class     = "db.serverless"
  engine             = aws_rds_cluster.aurora.engine
  engine_version     = aws_rds_cluster.aurora.engine_version
}

resource "aws_db_subnet_group" "aurora" {
  name       = "rag-aurora-subnet-group"
  subnet_ids = aws_subnet.private[*].id
}

resource "aws_dynamodb_table" "conversation_memory" {
  name         = "rag-conversation-memory"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "session_id"
  range_key    = "id"
  attribute {
    name = "session_id"
    type = "S"
  }
  attribute {
    name = "id"
    type = "S"
  }
  ttl {
    attribute_name = "expires_at"
    enabled        = true
  }
}

resource "aws_s3_bucket" "documents" {
  bucket = "rag-support-documents-${data.aws_caller_identity.current.account_id}"
}

resource "aws_s3_bucket_notification" "documents" {
  bucket = aws_s3_bucket.documents.id
  lambda_function {
    lambda_function_arn = aws_lambda_function.ingestion.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "documents/"
  }
  depends_on = [aws_lambda_permission.s3_invoke]
}

resource "aws_iam_role" "lambda_exec" {
  name = "rag-lambda-exec-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_policy" {
  name = "rag-lambda-policy"
  role = aws_iam_role.lambda_exec.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream", "bedrock:ApplyGuardrail"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Query", "dynamodb:Scan"]
        Resource = aws_dynamodb_table.conversation_memory.arn
      },
      {
        Effect = "Allow"
        Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
        Resource = [aws_s3_bucket.documents.arn, "${aws_s3_bucket.documents.arn}/*"]
      },
      {
        Effect = "Allow"
        Action = ["rds-data:ExecuteStatement", "rds-data:BatchExecuteStatement"]
        Resource = aws_rds_cluster.aurora.arn
      },
      {
        Effect = "Allow"
        Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
        Resource = "arn:aws:logs:*:*:*"
      },
      {
        Effect = "Allow"
        Action = ["ec2:CreateNetworkInterface", "ec2:DescribeNetworkInterfaces", "ec2:DeleteNetworkInterface"]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = ["cloudwatch:PutMetricData"]
        Resource = "*"
      }
    ]
  })
}

resource "aws_lambda_function" "ingestion" {
  function_name = "rag-document-ingestion"
  role          = aws_iam_role.lambda_exec.arn
  handler       = "ingestion.handler"
  runtime       = "python3.12"
  timeout       = 300
  memory_size   = 1024
  filename      = "../lambda/ingestion.zip"
  vpc_config {
    subnet_ids         = aws_subnet.private[*].id
    security_group_ids = [aws_security_group.lambda.id]
  }
  environment {
    variables = {
      AURORA_HOST     = aws_rds_cluster.aurora.endpoint
      AURORA_DB       = "ragdb"
      AURORA_USER     = "ragadmin"
      AURORA_PASSWORD = var.db_password
      BEDROCK_REGION  = var.aws_region
    }
  }
}

resource "aws_lambda_permission" "s3_invoke" {
  statement_id  = "AllowS3Invoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.ingestion.function_name
  principal     = "s3.amazonaws.com"
  source_arn    = aws_s3_bucket.documents.arn
}

resource "aws_bedrock_guardrail" "main" {
  name                      = "rag-support-guardrail"
  description               = "PII filtering and topic boundaries for customer support agent"
  blocked_input_messaging   = "I cannot process that request."
  blocked_outputs_messaging = "I cannot provide that information."
  sensitive_information_policy_config {
    pii_entities_config {
      type   = "EMAIL"
      action = "ANONYMIZE"
    }
    pii_entities_config {
      type   = "CREDIT_DEBIT_CARD_NUMBER"
      action = "BLOCK"
    }
    pii_entities_config {
      type   = "US_SOCIAL_SECURITY_NUMBER"
      action = "BLOCK"
    }
  }
  topic_policy_config {
    topics_config {
      name       = "off-topic"
      definition = "Requests unrelated to customer support, product questions, or account management."
      examples   = ["Tell me a joke", "What is the weather today?"]
      type       = "DENY"
    }
  }
}

resource "aws_lambda_function" "query" {
  function_name = "rag-query-handler"
  role          = aws_iam_role.lambda_exec.arn
  handler       = "query.handler"
  runtime       = "python3.12"
  timeout       = 60
  memory_size   = 1024
  filename      = "../lambda/query.zip"
  vpc_config {
    subnet_ids         = aws_subnet.private[*].id
    security_group_ids = [aws_security_group.lambda.id]
  }
  environment {
    variables = {
      AURORA_HOST       = aws_rds_cluster.aurora.endpoint
      AURORA_DB         = "ragdb"
      AURORA_USER       = "ragadmin"
      AURORA_PASSWORD   = var.db_password
      BEDROCK_REGION    = var.aws_region
      DYNAMODB_TABLE    = aws_dynamodb_table.conversation_memory.name
      GUARDRAIL_ID      = aws_bedrock_guardrail.main.guardrail_id
      GUARDRAIL_VERSION = "DRAFT"
    }
  }
}

resource "aws_api_gateway_rest_api" "main" {
  name = "rag-support-api"
}

resource "aws_api_gateway_resource" "chat" {
  rest_api_id = aws_api_gateway_rest_api.main.id
  parent_id   = aws_api_gateway_rest_api.main.root_resource_id
  path_part   = "chat"
}

resource "aws_api_gateway_method" "chat_post" {
  rest_api_id   = aws_api_gateway_rest_api.main.id
  resource_id   = aws_api_gateway_resource.chat.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "chat" {
  rest_api_id             = aws_api_gateway_rest_api.main.id
  resource_id             = aws_api_gateway_resource.chat.id
  http_method             = aws_api_gateway_method.chat_post.http_method
  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.query.invoke_arn
}

resource "aws_api_gateway_deployment" "main" {
  rest_api_id = aws_api_gateway_rest_api.main.id
  stage_name  = "prod"
  depends_on  = [aws_api_gateway_integration.chat]
}

resource "aws_lambda_permission" "apigw_invoke" {
  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.main.execution_arn}/*/*"
}

Here is the complete terraform/outputs.tf:

output "aurora_endpoint" {
  value = aws_rds_cluster.aurora.endpoint
}

output "s3_bucket_name" {
  value = aws_s3_bucket.documents.bucket
}

output "dynamodb_table_name" {
  value = aws_dynamodb_table.conversation_memory.name
}

output "api_gateway_id" {
  value = aws_api_gateway_rest_api.main.id
}

output "api_endpoint" {
  value = "${aws_api_gateway_deployment.main.invoke_url}/chat"
}

output "guardrail_id" {
  value = aws_bedrock_guardrail.main.guardrail_id
}

Here is the complete lambda/query.py:

import json
import os
import time
import logging
import boto3
from langchain_core.messages import HumanMessage
from agent import app

logger = logging.getLogger()
logger.setLevel(logging.INFO)

bedrock_client = boto3.client('bedrock-runtime', region_name=os.environ.get('BEDROCK_REGION', 'us-east-1'))
GUARDRAIL_ID = os.environ['GUARDRAIL_ID']
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION', 'DRAFT')


def apply_guardrail(text: str, source: str = 'INPUT') -> dict:
    response = bedrock_client.apply_guardrail(
        guardrailIdentifier=GUARDRAIL_ID,
        guardrailVersion=GUARDRAIL_VERSION,
        source=source,
        content=[{'text': {'text': text}}]
    )
    return response


def handler(event, context):
    start_time = time.time()
    try:
        body = json.loads(event.get('body', '{}'))
        message = body.get('message', '')
        session_id = body.get('session_id', 'default')

        guardrail_result = apply_guardrail(message, source='INPUT')
        if guardrail_result.get('action') == 'GUARDRAIL_INTERVENED':
            logger.info(json.dumps({'event': 'guardrail_blocked', 'session_id': session_id}))
            return {
                'statusCode': 200,
                'headers': {'Content-Type': 'application/json'},
                'body': json.dumps({'response': 'I cannot process that request.', 'blocked': True})
            }

        config = {'configurable': {'thread_id': session_id}}
        result = app.invoke(
            {'messages': [HumanMessage(content=message)], 'session_id': session_id},
            config=config
        )

        final_message = result['messages'][-1].content

        output_guardrail = apply_guardrail(final_message, source='OUTPUT')
        if output_guardrail.get('action') == 'GUARDRAIL_INTERVENED':
            final_message = 'I cannot provide that information.'

        elapsed = time.time() - start_time
        logger.info(json.dumps({
            'event': 'query_complete',
            'session_id': session_id,
            'elapsed_ms': round(elapsed * 1000)
        }))

        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'response': final_message, 'session_id': session_id})
        }
    except Exception as e:
        logger.error(json.dumps({'event': 'query_error', 'error': str(e)}))
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

Package the query Lambda (including agent.py since query.py imports it) and deploy everything by running:

cd lambda
zip -r query.zip query.py agent.py
cd ../terraform
terraform apply -auto-approve

What does this deploy?

Terraform creates the Bedrock Guardrail, the query Lambda, the API Gateway /chat resource with its POST method, deploys to a prod stage, and wires up the Lambda permission. The api_endpoint output gives you the live URL.

Wait for Terraform to complete. You should see output values including api_endpoint and guardrail_id printed in your terminal.

Terraform apply failed?

If you see an error about query.zip not found, make sure you ran the zip command from inside the lambda/ directory first.
If the guardrail creation fails, check that your AWS account has Bedrock access enabled in us-east-1. The guardrail feature requires Bedrock to be active in your region.
If you see IAM permission errors, verify that bedrock:ApplyGuardrail is in your Lambda policy (it should already be there from the prior step).

Still stuck? Help me debug my terraform apply failure for the guardrail and API Gateway deployment.

Test the full end-to-end flow

Your API is live. A POST request to the /chat endpoint flows through API Gateway, invokes the query Lambda, passes the guardrail check, runs the LangGraph agent (which retrieves context from Aurora pgvector), and returns a grounded answer.

Copy the api_endpoint value from your Terraform output.
Test the API by running the following curl command (replace the URL with your endpoint):

curl -X POST [[API_ENDPOINT="https://your-api-id.execute-api.us-east-1.amazonaws.com/prod/chat"]] \
  -H "Content-Type: application/json" \
  -d '{"message": "What is your return policy?", "session_id": "test-session-1"}'

What should I see?

You should receive a JSON response containing a response field with an answer about the 30-day return policy. The answer is grounded in the support_faq.txt document you uploaded earlier, and the agent cites its source.

The session_id in the response confirms that conversation memory is active. Subsequent requests with the same session ID will remember context from this exchange.

Now test that the guardrail blocks off-topic requests:

curl -X POST [[API_ENDPOINT="https://your-api-id.execute-api.us-east-1.amazonaws.com/prod/chat"]] \
  -H "Content-Type: application/json" \
  -d '{"message": "Tell me a joke", "session_id": "test-session-2"}'

You should see "blocked": true and "response": "I cannot process that request." in the response. The guardrail intercepted the off-topic request before it reached the LLM.

Getting a 500 error or timeout?

Check CloudWatch Logs for the rag-query-handler Lambda. Look for the specific error message in the logs.
If you see a timeout, the Lambda may not be able to reach Aurora. Verify that the Lambda's VPC security group allows outbound traffic on port 5432.
If you see an import error for agent, make sure your query.zip contains both query.py and agent.py.

Still stuck? Help me debug my RAG query handler Lambda that is returning errors when I call the API Gateway endpoint.

Your agent is now live behind a production API with security guardrails on both input and output. Next up, you will add monitoring and observability to track how the system performs under load.

Add Monitoring & Observability

Your RAG agent is deployed and answering questions. But without observability, you have no way to know if it's performing well or failing silently.

In production, RAG pipelines degrade gradually. Retrieval gets slower, generation times creep up, and error rates spike without anyone noticing. In this step, you'll build a CloudWatch dashboard that surfaces these problems the moment they happen.

In this step, get ready to:

Create a CloudWatch dashboard tracking latency and error rates.
Add log metric filters that extract custom timing metrics from structured logs.
Test the monitoring with a burst of requests and verify the dashboard populates.

Create the CloudWatch dashboard

Your Lambda functions already emit structured JSON logs at each pipeline stage. The next piece is a dashboard that visualizes four key metrics: end-to-end latency, error rates, retrieval time, and generation time.

Add the CloudWatch dashboard resource to terraform/main.tf by pasting the following at the end of the file:

resource "aws_cloudwatch_dashboard" "main" {
  dashboard_name = "rag-support-dashboard"
  dashboard_body = jsonencode({
    widgets = [
      {
        type = "metric"
        properties = {
          title   = "End-to-End Latency (p50/p95)"
          metrics = [
            ["AWS/Lambda", "Duration", "FunctionName", "rag-query-handler", { stat = "p50" }],
            ["AWS/Lambda", "Duration", "FunctionName", "rag-query-handler", { stat = "p95" }]
          ]
          period = 60
          view   = "timeSeries"
        }
      },
      {
        type = "metric"
        properties = {
          title   = "Lambda Error Rate"
          metrics = [
            ["AWS/Lambda", "Errors", "FunctionName", "rag-query-handler"],
            ["AWS/Lambda", "Errors", "FunctionName", "rag-document-ingestion"]
          ]
          period = 60
          view   = "timeSeries"
        }
      },
      {
        type = "metric"
        properties = {
          title   = "Retrieval Time (ms)"
          metrics = [["RAGSupport", "RetrievalTimeMs", { stat = "Average" }]]
          period  = 60
          view    = "timeSeries"
        }
      },
      {
        type = "metric"
        properties = {
          title   = "Generation Time (ms)"
          metrics = [["RAGSupport", "GenerationTimeMs", { stat = "Average" }]]
          period  = 60
          view    = "timeSeries"
        }
      }
    ]
  })
}

What does this dashboard track?

End-to-End Latency (p50/p95) uses the built-in Lambda Duration metric to show median and 95th percentile response times for rag-query-handler.
Lambda Error Rate tracks errors across both Lambda functions so you spot failures instantly.
Retrieval Time (ms) and Generation Time (ms) are custom metrics from the RAGSupport namespace. These come from log metric filters you'll create next.

Save terraform/main.tf.

You won't see the dashboard in the console yet. You'll deploy everything together after adding the metric filters.

Add log metric filters for custom metrics

Your agent code already emits structured JSON logs with retrieval_complete and llm_response events. Log metric filters scan those log lines and extract the elapsed_ms value into CloudWatch custom metrics.

Add the two log metric filter resources to terraform/main.tf below the dashboard resource:

resource "aws_cloudwatch_log_metric_filter" "retrieval_time" {
  name           = "rag-retrieval-time"
  pattern        = "{ $.event = \"retrieval_complete\" }"
  log_group_name = "/aws/lambda/rag-query-handler"
  metric_transformation {
    name      = "RetrievalTimeMs"
    namespace = "RAGSupport"
    value     = "$.elapsed_ms"
  }
}

resource "aws_cloudwatch_log_metric_filter" "generation_time" {
  name           = "rag-generation-time"
  pattern        = "{ $.event = \"llm_response\" }"
  log_group_name = "/aws/lambda/rag-query-handler"
  metric_transformation {
    name      = "GenerationTimeMs"
    namespace = "RAGSupport"
    value     = "$.elapsed_ms"
  }
}

How do log metric filters work?

Each filter watches the /aws/lambda/rag-query-handler log group for JSON log lines matching a specific event field. When a match is found, the filter extracts $.elapsed_ms from the log line and publishes it as a CloudWatch metric in the RAGSupport namespace.

This connects your structured logging directly to the dashboard widgets you just defined.

Now add the dashboard URL to terraform/outputs.tf so Terraform prints it after deployment:

output "dashboard_url" {
  value = "https://console.aws.amazon.com/cloudwatch/home?region=${var.aws_region}#dashboards:name=rag-support-dashboard"
}

Save both terraform/main.tf and terraform/outputs.tf.

Pattern syntax errors?

The CloudWatch filter pattern uses JSON selector syntax. Make sure the curly braces and dollar sign are present: { $.event = "retrieval_complete" }. Missing the $ or the outer braces will cause the filter to never match.

help me debug my CloudWatch log metric filter pattern

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

Here is the complete terraform/outputs.tf file at this point:

output "aurora_endpoint" {
  value = aws_rds_cluster.aurora.endpoint
}

output "s3_bucket_name" {
  value = aws_s3_bucket.documents.bucket
}

output "dynamodb_table_name" {
  value = aws_dynamodb_table.conversation_memory.name
}

output "api_gateway_id" {
  value = aws_api_gateway_rest_api.main.id
}

output "api_endpoint" {
  value = "${aws_api_gateway_deployment.main.invoke_url}/chat"
}

output "guardrail_id" {
  value = aws_bedrock_guardrail.main.guardrail_id
}

output "dashboard_url" {
  value = "https://console.aws.amazon.com/cloudwatch/home?region=${var.aws_region}#dashboards:name=rag-support-dashboard"
}

Deploy and test the monitoring dashboard

Time to deploy the monitoring resources and generate traffic to populate the dashboard.

Deploy the new resources by running this from your terraform directory:

terraform apply -auto-approve

After Terraform completes, you should see the dashboard_url output printed in your terminal.
Send a burst of 10 or more requests to your API to generate metrics. Run this script from your project root:

for i in $(seq 1 12); do
  curl -s -X POST "$(terraform -chdir=terraform output -raw api_endpoint)" \
    -H "Content-Type: application/json" \
    -d "{\"message\": \"What is your return policy? Request $i\", \"session_id\": \"burst-test-$i\"}" &
done
wait
echo "Burst test complete"

What does this script do?

The loop sends 12 parallel requests to your API endpoint, each with a unique session ID. The & runs each curl in the background, and wait pauses until all finish. This generates enough log events for the metric filters to populate the dashboard.

Wait 2-3 minutes for CloudWatch to process the log events and publish metrics.
Open the dashboard URL from the Terraform output in your browser.

You should see all four widgets populated: latency p50/p95 lines, error rates (ideally at zero), retrieval time averaging a few hundred milliseconds, and generation time showing LLM response durations.

Dashboard widgets showing 'No data'?

CloudWatch custom metrics can take 3-5 minutes to appear after the first log events are emitted. If the built-in Lambda widgets (latency, errors) show data but the custom widgets (Retrieval Time, Generation Time) do not, your metric filters may not be matching.

Check that the Lambda log group /aws/lambda/rag-query-handler exists in CloudWatch Logs. If the Lambda was recently deployed, it may not have a log group until the first invocation.

help me troubleshoot why my CloudWatch custom metrics are not appearing

Your production RAG system now has full observability. You can spot latency spikes, error bursts, and slow retrieval before your users notice. Next up: the Secret Mission where you'll add a human-in-the-loop approval workflow.

Secret mission

Add a Human-in-the-Loop Approval Workflow

Your agent can issue refunds and escalate tickets without any human oversight. In production, sensitive actions like refunds need a human to approve them before they execute. Add a pause-approve-resume workflow using LangGraph's interrupt_before pattern so refund requests wait for explicit human approval.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. Aurora Serverless v2 incurs charges as long as the cluster exists, even at its minimum capacity.

Cost warning

Your Aurora Serverless v2 cluster costs approximately $0.12/ACU-hour at its minimum 0.5 ACU. That adds up to roughly $2-3 per day if left running.

If you are done with this project, run terraform destroy to remove everything. This is the recommended option after completing the project.

Resources you used:

Aurora Serverless v2 cluster (pgvector vector database).
DynamoDB table (rag-conversation-memory for agent checkpoints and conversation memory).
S3 bucket (document storage for ingestion pipeline).
Lambda functions (rag-query-handler and rag-approve-handler).
API Gateway (rag-support-api with /chat and /approve endpoints).
CloudWatch dashboard (rag-support-dashboard).
Bedrock Guardrail (rag-support-guardrail).
IAM roles and VPC with private subnets.
Local project directory with Terraform config, Lambda code, and sample documents.

Keep everything running

No action needed. Choose this if you want to demo the system or continue experimenting with the agent.

Expect approximately $2-3 per day in Aurora costs while the cluster is running.
Bedrock costs are negligible (pay-per-token on Claude Haiku 4.5).
DynamoDB, Lambda, and API Gateway usage stays within free tier for light usage.
Your local project directory remains on your machine with Terraform state intact.

Pause - I'll come back to this later

Aurora Serverless v2 cannot be fully paused to zero cost while the cluster exists. You can reduce charges by ensuring no queries hit the cluster, which keeps it at minimum capacity (0.5 ACU).

Stop sending requests to the API. The Aurora cluster will still cost ~$1.44/day at minimum capacity.
If you want to eliminate all costs while preserving your Terraform configuration for later, run terraform destroy now and terraform apply when you return. Your infrastructure will be recreated from scratch.
Keep your local project directory intact so you can redeploy anytime.

Delete - I don't want to use this again

Remove all project resources and start fresh. Terraform handles the full teardown for you.

Navigate to your Terraform project directory in the terminal.
Destroy all AWS resources by running:

terraform destroy

Terraform will show a plan of all resources being destroyed. Type yes when prompted to confirm.

What does this remove?

This single command removes the Aurora cluster, DynamoDB table, S3 bucket (and its contents), both Lambda functions, the API Gateway, CloudWatch dashboard, Bedrock Guardrail, IAM roles, and the VPC. Terraform determines the correct destruction order based on resource dependencies.

After Terraform finishes, delete your local project directory by running:

cd ~/Desktop
rm -rf rag-support-agent

On Windows, run this instead:

cd ~\Desktop
rmdir /s /q rag-support-agent

Confirm the folder is gone by running:

ls

You should no longer see rag-support-agent in the output.

What about Bedrock model access?

Enabling model access in the Bedrock console is an account-level setting. It does not cost anything to leave enabled and does not need to be revoked.

Nice Work!

You've just built and deployed a production-grade AI customer support system on AWS from scratch!

This is the exact architecture pattern companies use to ship conversational AI at scale. From document ingestion to inference to monitoring, you now understand how the entire stack fits together.

You've learned how to:

Design and deploy a two-path production RAG architecture on AWS with offline document ingestion (S3 to Aurora pgvector) and online query handling (Claude Haiku 4.5 on Bedrock).
Build a LangGraph agent with tool calling, persistent conversation memory (DynamoDB), and Bedrock Guardrails for PII filtering and topic boundaries.
Provision and manage the entire stack as infrastructure as code with Terraform, including CloudWatch dashboards for monitoring latency, token usage, and error rates.
Implement a human-in-the-loop approval workflow using LangGraph's interrupt_before pattern for sensitive actions like refunds and escalations.

Ready to quiz yourself?