AWSで始めるRAGチャットボット構築⑤〜CDKによるインフラ定義〜

はじめに

前回の記事では、ユーザーの質問に対して適切な情報を検索し、LLMで自然な回答を生成するRAG処理を実装しました。今回は、AWS CDKを使ってすべてのAWSリソースをコードで定義していきます。

CDKを使うことで、複雑なAWSリソースをコードで管理し、再現可能な環境を構築できます。次回の最終回では、実際にデプロイして動作確認を行います。

CDKによるインフラ構築の概要

今回CDKで構築するAWSリソースは以下の通りです。

IAMロール – Lambda実行権限
S3バケット – PDFドキュメント保存
DynamoDBテーブル – ユーザー情報管理
OpenSearchドメイン – ベクトル検索エンジン
Lambda Layer × 2 – requests、PyPDF2
Lambda関数 × 3 – rag_handler、indexing_handler、create_index
Custom Resource – インデックス自動初期化
API Gateway – RESTエンドポイント

CDKスタックの処理フローは以下の通りです。

[CDK Deploy実行]
    ↓
[IAMロール作成]
    ↓
[S3、DynamoDB、OpenSearch作成]
    ↓
[Lambda Layer準備]
    ↓
[Lambda関数デプロイ]
    ↓
[Custom Resource実行] ← create_indexが自動実行
    ↓
[API Gateway作成]
    ↓
[デプロイ完了] 即座に利用可能な状態

CDKプロジェクトの構成

CDKプロジェクトは、主に2つのファイルで構成されます。

エントリーポイントファイル（app.py） – CDKアプリケーションのエントリーポイント
スタックファイル（rag_cdk/rag_cdk_stack.py） – 実際のAWSリソースを定義するスタック

エントリーポイントファイル（app.py）の構成

まず、app.pyでCDKアプリケーションを初期化し、スタックをインスタンス化します。

#!/usr/bin/env python3

import aws_cdk as cdk

from rag_cdk.rag_cdk_stack import RagCdkStack

app = cdk.App()
RagCdkStack(app, "RagStack")

app.synth()

app.pyの役割

cdk.App() – CDKアプリケーションのインスタンスを作成
RagCdkStack(app, “RagStack”) – スタックをインスタンス化（第2引数がCloudFormationスタック名）
app.synth() – CloudFormationテンプレートを生成

スタック名「RagStack」は任意に変更可能です。この名前がAWS CloudFormationコンソールに表示されます。

スタックファイル（rag_cdk/rag_cdk_stack.py）の構成

次に、実際のAWSリソースを定義するrag_cdk_stack.pyの構造を確認しておきましょう。

from aws_cdk import CustomResource
from aws_cdk import Duration
from aws_cdk import Stack
from aws_cdk import aws_apigateway as apigateway
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as _lambda
from aws_cdk import aws_opensearchservice as opensearch
from aws_cdk import aws_s3 as s3
from aws_cdk import custom_resources as cr
from aws_cdk.aws_logs import RetentionDays
from constructs import Construct


class RagCdkStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
        
        # 以降、各リソースを定義していきます

import文の説明

CustomResource – カスタムリソース（インデックス自動初期化用）
Duration – タイムアウト時間などの期間設定
Stack – CDKスタックの基底クラス
aws_apigateway – API Gatewayの定義
aws_dynamodb – DynamoDBテーブルの定義
aws_iam – IAMロール・ポリシーの定義
aws_lambda – Lambda関数とLayerの定義
aws_opensearchservice – OpenSearchドメインの定義
aws_s3 – S3バケットの定義
custom_resources – Custom Resourceプロバイダー
RetentionDays – CloudWatch Logsの保持期間設定

RagCdkStackクラス

Stackクラスを継承してRagCdkStackクラスを定義します。initメソッド内で、すべてのAWSリソースを定義していきます。

それでは、各リソースの定義を順番に見ていきましょう。

Step 1: IAMロールの設定 (rag_cdk_stack.py)

これから、AWS CDKを使ってRAGシステムのインフラを定義していきます。まずは、Lambda関数が各種AWSサービスにアクセスするためのIAMロールを作成します。

# Lambda 実行ロール
lambda_role = iam.Role(
    self,
    "RagLambdaRole",
    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
    managed_policies=[
        iam.ManagedPolicy.from_aws_managed_policy_name(
            "service-role/AWSLambdaBasicExecutionRole"
        ),
    ],
)

AWSLambdaBasicExecutionRoleにより、CloudWatch Logsへのログ出力が可能になります。Bedrock、OpenSearch、S3、DynamoDBへのアクセス権限は、後ほど個別に追加します。

Step 2: S3バケットの作成 (rag_cdk_stack.py)

# S3 バケット（データレイク）
data_bucket = s3.Bucket(
    self,
    "RagDataLakeBucket",
    bucket_name="rag-datalake-bucket",
)

S3バケットは、生命保険プランなどのPDFドキュメントを保存するために使用します。シンプルな構成ですが、将来的にバージョニングや暗号化を追加することも可能です。

Step 3: DynamoDBテーブルの作成 (rag_cdk_stack.py)

# DynamoDB（状況コンテキスト格納）
context_table = dynamodb.Table(
    self,
    "RagContextTable",
    partition_key=dynamodb.Attribute(name="user_id", type=dynamodb.AttributeType.STRING),
    table_name="rag-context-table",
)

user_idをパーティションキーとして、ユーザーごとの属性情報を管理します。

Step 4: OpenSearchドメインの設定 (rag_cdk_stack.py)

# アクセスポリシー
access_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"AWS": lambda_role.role_arn},
            "Action": ["es:ESHttpPost", "es:ESHttpGet", "es:ESHttpPut", "es:ESHttpDelete"],
            "Resource": f"arn:aws:es:{self.region}:{self.account}:domain/rag-search/*",
        }
    ],
}

# OpenSearch（ベクトル検索用ドメイン）
search_domain = opensearch.CfnDomain(
    self,
    "RagSearchDomain",
    domain_name="rag-search",
    engine_version="OpenSearch_3.1",
    cluster_config={
        "instanceType": "t3.small.search",
        "instanceCount": 1,
    },
    ebs_options={"ebsEnabled": True, "volumeSize": 10, "volumeType": "gp3"},
    encryption_at_rest_options={"enabled": True},
    node_to_node_encryption_options={"enabled": True},
    domain_endpoint_options={"enforce_https": True},
    access_policies=access_policy,
)

検証環境向けの低コスト構成

今回は検証環境向けの低コスト構成を採用しています。

インスタンスタイプ: t3.small.search – 開発・検証用の低コストインスタンス
ノード数: 1台 – シングルノード構成でコストを最小化
エンジンバージョン: OpenSearch 3.1 – 最新のベクトル検索機能をサポート
保存時・通信時の両方で暗号化を有効化 – セキュリティ要件を満たす
月額コスト: 約$30 – PoCや検証に最適

本番環境への拡張

本番環境で高可用性が必要な場合は、以下のような構成に変更できます。

cluster_config={
    "instanceType": "r5.large.search",
    "instanceCount": 3,
    "zoneAwarenessEnabled": True,
    "zoneAwarenessConfig": {"availabilityZoneCount": 3},
    "dedicatedMasterEnabled": True,
    "dedicatedMasterType": "m5.large.search",
    "dedicatedMasterCount": 3,
}

3ノード構成で高可用性を確保
マルチAZ（3つのアベイラビリティゾーン）に分散配置
専用マスターノード（m5.large.search × 3）でクラスター管理の安定性向上
データノード（r5.large.search × 3）はメモリ最適化インスタンスでベクトル検索に最適
月額コスト: 約$600 – 本番運用向けの高可用性構成

Step 5: Lambda Layerの準備 (rag_cdk_stack.py)

# Lambda Layer(requests・requests-aws4auth)
requests_layer = _lambda.LayerVersion(
    self,
    "RequestsLayer",
    code=_lambda.Code.from_asset("lambda/requests-layer.zip"),
    compatible_runtimes=[_lambda.Runtime.PYTHON_3_11],
    description="Layer with requests and requests-aws4auth",
)
# Lambda Layer(PyPDF2)
pypdf2_layer = _lambda.LayerVersion(
    self,
    "PyPDF2Layer",
    code=_lambda.Code.from_asset("lambda/pypdf2-layer.zip"),
    compatible_runtimes=[_lambda.Runtime.PYTHON_3_11],
    description="Layer with PyPDF2",
)

Lambda Layerは、事前に作成したzipファイルをCDKでデプロイします。Layer作成手順は第三弾の記事で解説しています。

Step 6: Lambda関数の定義 (rag_cdk_stack.py)

RAGハンドラー関数

# Lambda 関数（検索・ベクトル生成・プロンプト組立て）
rag_lambda = _lambda.Function(
    self,
    "RagHandlerFunction",
    runtime=_lambda.Runtime.PYTHON_3_11,
    handler="main.lambda_handler",
    code=_lambda.Code.from_asset("lambda/rag_handler"),
    log_retention=RetentionDays.ONE_WEEK,
    role=lambda_role,
    function_name="rag-handler",
    timeout=Duration.seconds(60),
    environment={
        "BEDROCK_CLAUDE_INFERENCE_PROFILE_ID": (
            "apac.anthropic.claude-3-5-sonnet-20241022-v2:0"
        ),
        "BEDROCK_TITAN_MODEL_ID": "amazon.titan-embed-text-v1",
        "DYNAMO_TABLE": context_table.table_name,
        "OPENSEARCH_ENDPOINT": search_domain.attr_domain_endpoint,
        "OPENSEARCH_INDEX": "rag-documents",
    },
    layers=[requests_layer],
)

environmentで環境変数を設定することで、コード内でos.environから各種設定を取得できます。search_domain.attr_domain_endpointのように、他のリソースの属性を参照できるのがCDKの強みです。

インデックスハンドラー関数

# Lambda 関数（インデックス登録）
indexing_lambda = _lambda.Function(
    self,
    "RagIndexingHandlerFunction",
    runtime=_lambda.Runtime.PYTHON_3_11,
    handler="main.lambda_handler",
    code=_lambda.Code.from_asset("lambda/indexing_handler"),
    log_retention=RetentionDays.ONE_WEEK,
    role=lambda_role,
    function_name="indexing-handler",
    timeout=Duration.seconds(60),
    environment={
        "BEDROCK_TITAN_MODEL_ID": "amazon.titan-embed-text-v1",
        "OPENSEARCH_ENDPOINT": search_domain.attr_domain_endpoint,
        "OPENSEARCH_INDEX": "rag-documents",
        "S3_BUCKET": data_bucket.bucket_name,
    },
    layers=[requests_layer, pypdf2_layer],
)

indexing_handlerには、PyPDF2とrequestsの両方のLayerをアタッチしています。

インデックス作成関数

# Lambda 関数（初回インデックス作成）
create_index_lambda = _lambda.Function(
    self,
    "RagCreateIndexFunction",
    runtime=_lambda.Runtime.PYTHON_3_11,
    handler="main.lambda_handler",
    code=_lambda.Code.from_asset("lambda/create_index"),
    log_retention=RetentionDays.ONE_WEEK,
    role=lambda_role,
    function_name="create-index",
    timeout=Duration.seconds(60),
    environment={
        "OPENSEARCH_ENDPOINT": search_domain.attr_domain_endpoint,
        "OPENSEARCH_INDEX": "rag-documents",
        "VECTOR_DIMENSION": "1536",
    },
    layers=[requests_layer],
)

Step 7: Custom Resourceによる自動初期化 (rag_cdk_stack.py)

# Custom Resource のプロバイダー
provider = cr.Provider(
    self,
    "CreateIndexProvider",
    on_event_handler=create_index_lambda,
)
# Custom Resource 本体
CustomResource(
    self,
    "CreateIndexResource",
    service_token=provider.service_token,
)

Custom Resourceは、CDKデプロイ時に任意の処理を実行できる仕組みです。今回は、OpenSearchドメインが作成された直後に自動的にcreate_index_lambdaを実行し、インデックスを初期化します。

これにより、デプロイ後すぐに利用可能な状態になります。第三弾の記事で説明した「インデックスが存在しないエラー」を防ぐための重要な設定です。

Step 8: アクセス権限の設定 (rag_cdk_stack.py)

Lambda関数が各AWSサービスにアクセスできるよう、IAMポリシーを設定します。

# Bedrock へのアクセスポリシー
lambda_role.add_to_policy(
    iam.PolicyStatement(actions=["bedrock:InvokeModel"], resources=["*"])
)
# OpenSearch へのアクセスポリシー
lambda_role.add_to_policy(
    iam.PolicyStatement(
        actions=["es:ESHttpPost", "es:ESHttpPut", "es:ESHttpGet", "es:ESHttpDelete"],
        resources=[f"{search_domain.attr_arn}/*"],
    )
)

# アクセス権限付与
data_bucket.grant_read(indexing_lambda)
context_table.grant_read_write_data(rag_lambda)

各設定の詳細

1. Bedrockへのアクセスポリシー

lambda_role.add_to_policy(
    iam.PolicyStatement(actions=["bedrock:InvokeModel"], resources=["*"])
)

bedrock:InvokeModel – Bedrockのモデル（Titan Embeddings、Claude 3.5 Sonnet v2）を呼び出す権限
resources=[“*”] – すべてのBedrockモデルへのアクセスを許可（特定のモデルARNに限定することも可能）
この権限により、rag_handlerとindexing_handlerの両方でBedrockが利用可能になる

2. OpenSearchへのアクセスポリシー

lambda_role.add_to_policy(
    iam.PolicyStatement(
        actions=["es:ESHttpPost", "es:ESHttpPut", "es:ESHttpGet", "es:ESHttpDelete"],
        resources=[f"{search_domain.attr_arn}/*"],
    )
)

es:ESHttpPost – ドキュメントの登録、検索クエリの実行に必要
es:ESHttpPut – インデックスの作成、ドキュメントの更新に必要
es:ESHttpGet – インデックスの確認、ドキュメントの取得に必要
es:ESHttpDelete – インデックスやドキュメントの削除に必要
resources=[f”{search_domain.attr_arn}/*”] – 作成したOpenSearchドメイン内のすべてのリソースへのアクセスを許可

3. S3バケットへの読み取り権限

data_bucket.grant_read(indexing_lambda)

grant_read – S3バケットからオブジェクトを読み取る権限を付与するCDKの便利メソッド
indexing_lambda – PDFファイルを読み込むために必須
rag_lambdaには付与していない（最小権限の原則）
内部的にs3:GetObject、s3:ListBucketなどの権限が自動的に設定される

4. DynamoDBテーブルへの読み書き権限

context_table.grant_read_write_data(rag_lambda)

grant_read_write_data – DynamoDBテーブルへの読み書き権限を付与するCDKの便利メソッド
rag_lambda – ユーザー情報の取得のために必須
indexing_lambdaはDynamoDBを使用しないため権限を付与していない（最小権限の原則）
内部的にdynamodb:GetItem、dynamodb:PutItem、dynamodb:UpdateItemなどの権限が自動的に設定される

最小権限の原則

今回の権限設定では、各Lambda関数が必要とする最小限の権限のみを付与しています。

Lambda関数	Bedrock	OpenSearch	S3	DynamoDB
rag_lambda	✓	✓	–	✓
indexing_lambda	✓	✓	✓	–
create_index_lambda	–	✓	–	–

このように、必要な権限のみを付与することで、セキュリティリスクを最小化しています。

Step 9: API Gatewayの作成 (rag_cdk_stack.py)

# API Gateway（Lambda 関数（検索・ベクトル生成・プロンプト組立て））
apigateway.LambdaRestApi(
    self,
    "RagApi",
    handler=rag_lambda,
    deploy_options=apigateway.StageOptions(
        logging_level=apigateway.MethodLoggingLevel.INFO,
        data_trace_enabled=True,
    ),
)

シンプルな設定で、rag_lambdaへのRESTエンドポイントを作成します。デプロイ後、API GatewayのURLからRAGシステムにアクセスできるようになります。

ハマったポイントと対処法

OpenSearchのデプロイが失敗する

問題: サービスリンクロールが存在しないというエラー
原因: OpenSearch用のサービスリンクロールが未作成
対処:

aws iam create-service-linked-role --aws-service-name es.amazonaws.com

Lambda Layerのインポートエラー

問題: デプロイ後にModuleNotFoundError: No module named ‘PyPDF2’ または ‘requests’
原因: Layerのディレクトリ構造が誤っている、またはzipファイルが正しく作成されていない
対処: 必ずpython/ディレクトリ配下にライブラリをインストールし、そのディレクトリをzipで圧縮

# 正しい構造
lambda/
  requests-layer.zip
    └── python/
        └── requests/
        └── requests_aws4auth/
        └── ...（その他の依存関係）

CDKデプロイがタイムアウトする

問題: OpenSearchドメインの作成に時間がかかりすぎてタイムアウト
原因: OpenSearchの大規模構成（3ノード + 専用マスター）は作成に時間がかかる
対処: 気長に待つ（通常15〜20分）。失敗した場合は、cdk deployを再実行すると途中から再開される

IAM権限エラー

問題: CDKデプロイ時に権限不足エラーが発生
原因: 使用しているIAMユーザー/ロールに必要な権限がない
対処: 以下の権限が必要です

CloudFormation操作権限
IAM作成・管理権限
Lambda作成権限
OpenSearch作成権限
S3バケット作成権限
DynamoDB作成権限
API Gateway作成権限

検証環境では、AdministratorAccessポリシーを一時的に付与するのが最も簡単です。

まとめ

今回は、AWS CDKを使ったRAGシステムのインフラ構築を実装しました。
実装した内容は以下の通りです。

IAMロールと権限設定
S3バケットの作成
DynamoDBテーブルの作成
OpenSearchドメインの設定
Lambda Layerの準備
Lambda関数の定義（3つ）
Custom Resourceによる自動初期化
API Gatewayの作成

rag_cdk_stack.pyの全9つのステップを実装し、コードでAWSインフラを管理する完全なCDKスタックが完成しました。
次回の最終回では、実際にデプロイを行い、動作確認とテスト、そして全体のまとめを行います。いよいよ完成です！

AWSで始めるRAGチャットボット構築⑤
〜CDKによるインフラ定義〜

はじめに

CDKによるインフラ構築の概要

CDKプロジェクトの構成

エントリーポイントファイル（app.py）の構成

app.pyの役割

スタックファイル（rag_cdk/rag_cdk_stack.py）の構成

import文の説明

RagCdkStackクラス

Step 1: IAMロールの設定 (rag_cdk_stack.py)

Step 2: S3バケットの作成 (rag_cdk_stack.py)

Step 3: DynamoDBテーブルの作成 (rag_cdk_stack.py)

Step 4: OpenSearchドメインの設定 (rag_cdk_stack.py)

検証環境向けの低コスト構成

本番環境への拡張

Step 5: Lambda Layerの準備 (rag_cdk_stack.py)

Step 6: Lambda関数の定義 (rag_cdk_stack.py)

RAGハンドラー関数

インデックスハンドラー関数

インデックス作成関数

Step 7: Custom Resourceによる自動初期化 (rag_cdk_stack.py)

Step 8: アクセス権限の設定 (rag_cdk_stack.py)

各設定の詳細

最小権限の原則

Step 9: API Gatewayの作成 (rag_cdk_stack.py)

ハマったポイントと対処法

OpenSearchのデプロイが失敗する

Lambda Layerのインポートエラー

CDKデプロイがタイムアウトする

IAM権限エラー

まとめ

【作ってみた】「ペットカードジェネレーター」で、お気に入りのあの子が最強カードに！？

生成 AI でスライド作成！Gemini 3 と NotebookLM のスライド生成を実際に試してみた

契約書の検索時間を約85%削減！NotebookLM で実現したパートナーアライアンス業務効率化

AI駆動開発の第一歩！エクセル設計書を「AIが読める」マークダウンに変換する最強プロンプト術

【AI時代のタスク管理】Obsidianで実現する「AIを意識したタスク整理術」

AWSで始めるRAGチャットボット構築⑤〜CDKによるインフラ定義〜

はじめに

CDKによるインフラ構築の概要

CDKプロジェクトの構成

エントリーポイントファイル（app.py）の構成

app.pyの役割

スタックファイル（rag_cdk/rag_cdk_stack.py）の構成

import文の説明

RagCdkStackクラス

Step 1: IAMロールの設定 (rag_cdk_stack.py)

Step 2: S3バケットの作成 (rag_cdk_stack.py)

Step 3: DynamoDBテーブルの作成 (rag_cdk_stack.py)

Step 4: OpenSearchドメインの設定 (rag_cdk_stack.py)

検証環境向けの低コスト構成

本番環境への拡張

Step 5: Lambda Layerの準備 (rag_cdk_stack.py)

Step 6: Lambda関数の定義 (rag_cdk_stack.py)

RAGハンドラー関数

インデックスハンドラー関数

インデックス作成関数

Step 7: Custom Resourceによる自動初期化 (rag_cdk_stack.py)

Step 8: アクセス権限の設定 (rag_cdk_stack.py)

各設定の詳細

最小権限の原則

Step 9: API Gatewayの作成 (rag_cdk_stack.py)

ハマったポイントと対処法

OpenSearchのデプロイが失敗する

Lambda Layerのインポートエラー

CDKデプロイがタイムアウトする

IAM権限エラー

まとめ

関連記事Related Articles

AWSで始めるRAGチャットボット構築②〜構成設計と各AWSサービスの役割〜

AWSで始めるRAGチャットボット構築③〜インデックス作成とPDF埋め込み処理の実装〜

AWSで始めるRAGチャットボット構築④〜RAG検索処理とLLM応答生成の実装〜

【AWS re:Invent 2022】AWS Application Composer を試してみた所感

AWS CDK Advanced Workshop（S3 クロスリージョンレプリケーション設定）をやって気になったこと

AWSで始めるRAGチャットボット構築⑤
〜CDKによるインフラ定義〜

AWSで始めるRAGチャットボット構築②
〜構成設計と各AWSサービスの役割〜

AWSで始めるRAGチャットボット構築③
〜インデックス作成とPDF埋め込み処理の実装〜

AWSで始めるRAGチャットボット構築④
〜RAG検索処理とLLM応答生成の実装〜