update code

fmops/full-stack/README.md

@@ -2,6 +2,9 @@
 
 This repository contains examples of deployment patterns for a full-stack LLM solution.
 
+## Disclaimer
+Sample code, software libraries, command line tools, proofs of concept, templates, or other related technology are provided as AWS Content or Third-Party Content under the AWS Customer Agreement, or the relevant written agreement between you and AWS (whichever applies). You should not use this AWS Content or Third-Party Content in your production accounts, or on production or other critical data. You are responsible for testing, securing, and optimizing the AWS Content or Third-Party Content, such as sample code, as appropriate for production grade use based on your specific quality control practices and standards. Deploying AWS Content or Third-Party Content may incur AWS charges for creating or using AWS chargeable resources, such as running Amazon EC2 instances or using Amazon S3 storage.
+
 ## Pattern 1: Retrieval Augmented Generation (RAG) based on PDF documents
 
 This pattern shows how to build a RAG system based on a library of PDF documents.

fmops/full-stack/pattern1-rag/cdk/lambda/embeddingprocessor/lambda_function.py

@@ -2,7 +2,7 @@
 import json
 import boto3
 from opensearchpy.helpers import bulk
-from opensearchpy import OpenSearch
+from opensearchpy import OpenSearch, AWSV4SignerAuth, RequestsHttpConnection
 import os
 import uuid
 
@@ -53,9 +53,18 @@ def lambda_handler(event, context):
     embedding_model = os.environ['EMBEDDINGS_MODEL_ENDPOINT']
 
     fhclient = boto3.client('firehose')
-
+  
     try:
-        opensearch = OpenSearch(os_url)
+        credentials = boto3.Session().get_credentials()
+        region = boto3.Session().region_name
+        auth = AWSV4SignerAuth(credentials, region, "es")
+        opensearch = OpenSearch( 
+          hosts = [{'host': os_url, 'port': 443}],
+          http_auth = auth,
+          use_ssl = True,
+          verify_certs = True,
+          connection_class = RequestsHttpConnection
+        )
         requests = []
         fh_stream_records = []
         for item in event['Items']:

fmops/full-stack/pattern1-rag/cdk/lambda/ossetup/lambda_function.py

@@ -1,6 +1,9 @@
 import json
 import os
 import requests
+import boto3
+from aws_requests_auth.boto_utils import BotoAWSRequestsAuth
+from opensearchpy import OpenSearch, AWSV4SignerAuth, RequestsHttpConnection
 
 def on_event(event, context):
 
@@ -33,11 +36,32 @@ def on_event(event, context):
             }
         }
         print(f"Checking domain {os_url}/{index_name}")
-        response = requests.head(f"{os_url}/{index_name}")
-        # If the index does not exist (status code 404), create the index
-        if response.status_code == 404:
-            response = requests.put(f"{os_url}/{index_name}", json=mapping)
-            print(f'Index created: {response.text}')
+
+        # region_name = boto3.Session().region_name
+        # auth = BotoAWSRequestsAuth(aws_host=os_url, aws_region=regio_name, aws_service='es')
+        # response = requests.head(f"{os_url}/{index_name}", auth=auth)
+        # # If the index does not exist (status code 404), create the index
+        # if response.status_code == 404:
+        #     response = requests.put(f"{os_url}/{index_name}", json=mapping, auth=auth)
+        #     print(f'Index created: {response.text}')
+        # else:
+        #     print('Index already exists!')
+        # return { 'PhysicalResourceId': index_name}
+
+        credentials = boto3.Session().get_credentials()
+        region = boto3.Session().region_name
+        auth = AWSV4SignerAuth(credentials, region, "es")
+        opensearch = OpenSearch( 
+          hosts = [{'host': os_url, 'port': 443}],
+          http_auth = auth,
+          use_ssl = True,
+          verify_certs = True,
+          connection_class = RequestsHttpConnection
+        )
+
+        if not opensearch.indices.exists(index_name):
+            opensearch.indices.create(index_name, body=mapping)
+            print(f'Index created: {opensearch.indices.get(index_name)}')
         else:
             print('Index already exists!')
         return { 'PhysicalResourceId': index_name}

fmops/full-stack/pattern1-rag/cdk/lambda/ossetup/requirements.txt

@@ -1 +1,3 @@
-requests
+requests
+opensearch-py
+aws-requests-auth

fmops/full-stack/pattern1-rag/cdk/lambda/textractpostprocessor/.gitignore

@@ -1 +1,2 @@
 app/__pycache__/
+env.json

fmops/full-stack/pattern1-rag/cdk/lib/backend-stack.ts

@@ -229,20 +229,25 @@ export class BackendStack extends cdk.Stack {
       },
       accessPolicies: [
         new cdk.aws_iam.PolicyStatement({
-          actions: ['es:*',],
+          actions: ['es:ESHttp*',],
           resources: ['*'],
           effect: cdk.aws_iam.Effect.ALLOW,
-          principals: [new cdk.aws_iam.AnyPrincipal()]
+          principals: [new cdk.aws_iam.AccountRootPrincipal]
         })],
+        enforceHttps: true,
+        nodeToNodeEncryption: true,
+        encryptionAtRest: {
+          enabled: true
+        },
         removalPolicy: cdk.RemovalPolicy.DESTROY
     })
     openSearchDomain.connections.allowFrom(ec2.Peer.ipv4(vpc.vpcCidrBlock), ec2.Port.allTraffic(), 'All traffic from VPC to OpenSearch')
     const createOsIndexLambda = new lambda.Function( this, `osIndexCustomResourceLambda`, {
-        runtime: lambda.Runtime.PYTHON_3_9,
+        runtime: lambda.Runtime.PYTHON_3_11,
         vpc: vpc,
         code: lambda.Code.fromAsset( "lambda/ossetup", {
           bundling: {
-            image: lambda.Runtime.PYTHON_3_9.bundlingImage,
+            image: lambda.Runtime.PYTHON_3_11.bundlingImage,
             command: [
               'bash', '-c',
               'pip install -r requirements.txt -t /asset-output && cp -au . /asset-output'
@@ -254,11 +259,20 @@ export class BackendStack extends cdk.Stack {
         timeout: cdk.Duration.minutes(1),
         memorySize: 1024,
         environment: {
-          DOMAINURL: "https://" + openSearchDomain.domainEndpoint,
+          DOMAINURL: openSearchDomain.domainEndpoint,
           INDEX: 'embeddings'
         }
       }
     );
+    createOsIndexLambda.addToRolePolicy(
+      new iam.PolicyStatement({
+        effect: iam.Effect.ALLOW,
+        actions: [
+          'es:ESHttp*'
+          ],
+        resources: [openSearchDomain.domainArn + "/*"]
+      })
+    )
     const customResourceProvider = new customResources.Provider( this, `osIndexCustomResourceProvider`, {
         onEventHandler: createOsIndexLambda,
       }
@@ -365,11 +379,11 @@ export class BackendStack extends cdk.Stack {
     });
 
     const csvToEmbeddingFn = new lambda.Function(this, 'csvToEmbeddingFn', {
-      runtime: lambda.Runtime.PYTHON_3_9,
+      runtime: lambda.Runtime.PYTHON_3_11,
       vpc: vpc,
       code: lambda.Code.fromAsset('lambda/embeddingprocessor', {
           bundling: {
-            image: lambda.Runtime.PYTHON_3_9.bundlingImage,
+            image: lambda.Runtime.PYTHON_3_11.bundlingImage,
             command: [
               'bash', '-c',
               'pip install -r requirements.txt -t /asset-output && cp -au . /asset-output'
@@ -381,7 +395,7 @@ export class BackendStack extends cdk.Stack {
       timeout: cdk.Duration.minutes(1),
       memorySize: 1024,
       environment: {
-        DOMAINURL: "https://" + openSearchDomain.domainEndpoint,
+        DOMAINURL: openSearchDomain.domainEndpoint,
         INDEX: 'embeddings',
         FIREHOSE: fh_embed.ref,
         EMBEDDINGS_MODEL_ENDPOINT: endpointEmbed
@@ -403,6 +417,17 @@ export class BackendStack extends cdk.Stack {
         resources: [fh_embed.attrArn]
       })
     );
+// Add policy to allow access to OpenSearch
+    csvToEmbeddingFn.addToRolePolicy(
+      new iam.PolicyStatement({
+        effect: iam.Effect.ALLOW,
+        actions: [
+          'es:ESHttp*'
+          ],
+        resources: [openSearchDomain.domainArn + "/*"]
+      })
+    );
+
     const csvToEmbedding = new sfn.CustomState(this, "CsvToEmbeddingMap", {
       stateJson: {
         Type: "Map",
@@ -469,7 +494,9 @@ export class BackendStack extends cdk.Stack {
       new iam.PolicyStatement({
         effect: iam.Effect.ALLOW,
         actions: [
-          'states:*',
+          'states:StartExecution',
+          'states:DescribeExecution',
+          'states:StopExecution'
         ],
         resources: ['arn:aws:states:' + this.region + ':' + this.account + ':stateMachine:StateMachinePdfToText*']
       })
@@ -514,7 +541,8 @@ export class BackendStack extends cdk.Stack {
       partitionKey: { name: 'jobtype', type: dynamodb.AttributeType.STRING },
       sortKey: { name: 'jobdate', type: dynamodb.AttributeType.NUMBER },
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
-      pointInTimeRecovery: true
+      pointInTimeRecovery: true,
+      encryption: dynamodb.TableEncryption.AWS_MANAGED
     });
     // Fields:
     // jobtypedate (jobtype-jobdate) (PK)
@@ -524,19 +552,22 @@ export class BackendStack extends cdk.Stack {
       partitionKey: { name: 'jobtypedate', type: dynamodb.AttributeType.STRING },
       sortKey: { name: 'centroid', type: dynamodb.AttributeType.NUMBER },
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
-      pointInTimeRecovery: true
+      pointInTimeRecovery: true,
+      encryption: dynamodb.TableEncryption.AWS_MANAGED
     });
     const driftTablePrompts = new dynamodb.Table(this, 'DriftTablePrompts', {
       partitionKey: { name: 'jobtype', type: dynamodb.AttributeType.STRING },
       sortKey: { name: 'jobdate', type: dynamodb.AttributeType.NUMBER },
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
-      pointInTimeRecovery: true
+      pointInTimeRecovery: true,
+      encryption: dynamodb.TableEncryption.AWS_MANAGED
     });
     const driftTableCentroidsPrompts = new dynamodb.Table(this, 'DriftTableCentroidsPrompts', {
       partitionKey: { name: 'jobtypedate', type: dynamodb.AttributeType.STRING },
       sortKey: { name: 'centroid', type: dynamodb.AttributeType.NUMBER },
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
-      pointInTimeRecovery: true
+      pointInTimeRecovery: true,
+      encryption: dynamodb.TableEncryption.AWS_MANAGED
     });
     // Fields:
     // jobtype (PK) [DISTANCE]
@@ -548,7 +579,8 @@ export class BackendStack extends cdk.Stack {
       partitionKey: { name: 'jobtype', type: dynamodb.AttributeType.STRING },
       sortKey: { name: 'jobdate', type: dynamodb.AttributeType.NUMBER },
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
-      pointInTimeRecovery: true
+      pointInTimeRecovery: true,
+      encryption: dynamodb.TableEncryption.AWS_MANAGED
     });
 
     // Glue job for embedding drift and prompt distance
@@ -570,10 +602,10 @@ export class BackendStack extends cdk.Stack {
       }
     });
     contentBucket.grantRead(driftJob);
-    driftTable.grantFullAccess(driftJob);
-    driftTableCentroids.grantFullAccess(driftJob);
-    driftTablePrompts.grantFullAccess(driftJob);
-    driftTableCentroidsPrompts.grantFullAccess(driftJob);
+    driftTable.grantReadWriteData(driftJob);
+    driftTableCentroids.grantReadWriteData(driftJob);
+    driftTablePrompts.grantReadWriteData(driftJob);
+    driftTableCentroidsPrompts.grantReadWriteData(driftJob);
     const distanceJob = new glue.Job(this, 'EmbeddingDistanceJob', {
       jobName: 'embedding-distance-analysis',
       executable: glue.JobExecutable.pythonEtl({
@@ -625,8 +657,17 @@ export class BackendStack extends cdk.Stack {
         ],
         resources: ['*']
       })
-    );  
-
+    );
+    nb_role.addToPolicy(
+      new iam.PolicyStatement({
+        effect: iam.Effect.ALLOW,
+        actions: [
+          'es:ESHttp*'
+          ],
+        resources: [openSearchDomain.domainArn + "/*"]
+      })
+    );
+
     const notebook = new sagemaker.CfnNotebookInstance(this, 'NotebookInstance', {
       instanceType: 'ml.t3.medium',
       roleArn: nb_role.roleArn,

fmops/full-stack/pattern1-rag/cdk/lib/frontend-stack.ts

@@ -108,7 +108,8 @@ export class FrontendStack extends cdk.Stack {
     taskDefinition.addToTaskRolePolicy(new iam.PolicyStatement({
         effect: iam.Effect.ALLOW,
         actions: [
-          'dynamodb:*'
+          "dynamodb:PutItem",
+          "dynamodb:GetItem"
           ],
           resources: [conversationMemoryTable.tableArn]
         }))
@@ -139,9 +140,9 @@ export class FrontendStack extends cdk.Stack {
     taskDefinition.addToTaskRolePolicy(new iam.PolicyStatement({
         effect: iam.Effect.ALLOW,
         actions: [
-          'es:*'
+          'es:ESHttp*'
           ],
-        resources: [props.openSearchDomainArn]
+        resources: [props.openSearchDomainArn + "/*"]
       }))
     appContainer.addPortMappings({ containerPort: 8501, protocol: ecs.Protocol.TCP});
 
@@ -204,7 +205,7 @@ export class FrontendStack extends cdk.Stack {
 
     // Setup cognito for user authentication
     const userPool = new UserPool(this, 'UserPool', {
-      selfSignUpEnabled: true,
+      selfSignUpEnabled: false,
       signInAliases: { email: true },
     });
     const userPoolClient = userPool.addClient('UserPoolClient', {
@@ -296,5 +297,8 @@ export class FrontendStack extends cdk.Stack {
     new cdk.CfnOutput(this, 'AppURL', {
       value: `https://${appCustomDomainName}`
     });
+    new cdk.CfnOutput(this, 'CognitoUserPool', {
+      value: userPool.userPoolId
+    });
   }
 }