feat(construct): updated summarization architecture and added support… (#351)

* feat(construct): updated summarization architecture and added support for images with claude 3

* chore: self mutation

Signed-off-by: github-actions <[email protected]>

* feat(construct): removed unwanted dead code

---------

Signed-off-by: github-actions <[email protected]>
Signed-off-by: Dinesh Sajwan <[email protected]>
Co-authored-by: Dinesh Sajwan <[email protected]>
Co-authored-by: github-actions <[email protected]>
Co-authored-by: Alain Krok <[email protected]>

Author: 4 people

apidocs/classes/SummarizationAppsyncStepfn.md

@@ -27,7 +27,6 @@
 - [lambdaTracing](SummarizationAppsyncStepfn.md#lambdatracing)
 - [node](SummarizationAppsyncStepfn.md#node)
 - [processedAssetBucket](SummarizationAppsyncStepfn.md#processedassetbucket)
-- [redisCluster](SummarizationAppsyncStepfn.md#rediscluster)
 - [retention](SummarizationAppsyncStepfn.md#retention)
 - [securityGroup](SummarizationAppsyncStepfn.md#securitygroup)
 - [stage](SummarizationAppsyncStepfn.md#stage)
@@ -200,14 +199,6 @@ Returns the instance of s3.IBucket used by the construct
 
 ___
 
-### redisCluster
-
-• `Readonly` **redisCluster**: `CfnCacheCluster`
-
-Returns an instance of redis cluster created by the construct
-
-___
-
 ### retention
 
 • **retention**: `RetentionDays` = `logs.RetentionDays.TEN_YEARS`

apidocs/interfaces/SummarizationAppsyncStepfnProps.md

@@ -8,7 +8,6 @@
 
 - [bucketInputsAssetsProps](SummarizationAppsyncStepfnProps.md#bucketinputsassetsprops)
 - [bucketProcessedAssetsProps](SummarizationAppsyncStepfnProps.md#bucketprocessedassetsprops)
-- [cfnCacheClusterProps](SummarizationAppsyncStepfnProps.md#cfncacheclusterprops)
 - [cognitoUserPool](SummarizationAppsyncStepfnProps.md#cognitouserpool)
 - [customDocumentReaderDockerLambdaProps](SummarizationAppsyncStepfnProps.md#customdocumentreaderdockerlambdaprops)
 - [customInputValidationDockerLambdaProps](SummarizationAppsyncStepfnProps.md#custominputvalidationdockerlambdaprops)
@@ -19,7 +18,6 @@
 - [existingInputAssetsBucketObj](SummarizationAppsyncStepfnProps.md#existinginputassetsbucketobj)
 - [existingMergedApi](SummarizationAppsyncStepfnProps.md#existingmergedapi)
 - [existingProcessedAssetsBucketObj](SummarizationAppsyncStepfnProps.md#existingprocessedassetsbucketobj)
-- [existingRedisCulster](SummarizationAppsyncStepfnProps.md#existingredisculster)
 - [existingSecurityGroup](SummarizationAppsyncStepfnProps.md#existingsecuritygroup)
 - [existingVpc](SummarizationAppsyncStepfnProps.md#existingvpc)
 - [isFileTransformationRequired](SummarizationAppsyncStepfnProps.md#isfiletransformationrequired)
@@ -61,27 +59,6 @@ Providing both this and `existingProcessedAssetsBucketObj` will cause an error.
 
 ___
 
-### cfnCacheClusterProps
-
-• `Optional` `Readonly` **cfnCacheClusterProps**: `CfnCacheClusterProps`
-
-Optional. Custom cfnCacheClusterProps for Redis.
-Providing existingRedisCulster and cfnCacheClusterProps together will result in error.
-
-**`Default`**
-
-```ts
-cacheNodeType -  'cache.r6g.xlarge'
-```
-
-**`Default`**
-
-```ts
-numCacheNodes- 1
-```
-
-___
-
 ### cognitoUserPool
 
 • `Readonly` **cognitoUserPool**: `IUserPool`
@@ -217,21 +194,6 @@ If None is provided then this contruct will create one.
 
 ___
 
-### existingRedisCulster
-
-• `Optional` `Readonly` **existingRedisCulster**: `CfnCacheCluster`
-
-Optional. Existing Redis cluster to cache the generated summary
-for subsequent request of same document.
-
-**`Default`**
-
-```ts
-- none
-```
-
-___
-
 ### existingSecurityGroup
 
 • `Optional` `Readonly` **existingSecurityGroup**: `ISecurityGroup`

docs/generative_ai_cdk_constructs.drawio

@@ -21,6 +21,7 @@
 tracer = Tracer(service="SUMMARY_DOCUMENT_READER")
 
 s3 = boto3.resource('s3')
+rekognition_client=boto3.client('rekognition')
 
 @tracer.capture_method
 def read_file_from_s3(bucket, key):
@@ -59,17 +60,44 @@ def get_file_transformation(transformed_asset_bucket,transformed_file_name,
     }
     if (check_file_exists(transformed_asset_bucket, transformed_file_name) is False):
             logger.info("Starting file transformation")
-            loader = S3FileLoaderInMemory(input_asset_bucket, original_file_name)
-            document_content = loader.load()
-            if not document_content:
-                response['status'] = 'Error'
-                response['summary'] = 'Not able to transform the file.'
-                return response 
-            encoded_string = document_content.encode("utf-8")
-            s3.Bucket(transformed_asset_bucket).put_object(Key=transformed_file_name, Body=encoded_string)
-            response['status'] = 'File transformed'
-            response['name'] = transformed_file_name
-            response['summary']=''
+            if(original_file_name.endswith('.pdf')):
+                loader = S3FileLoaderInMemory(input_asset_bucket, original_file_name)
+                document_content = loader.load()
+                if not document_content:
+                    response['status'] = 'Error'
+                    response['summary'] = 'Not able to transform the file.'
+                    return response 
+                encoded_string = document_content.encode("utf-8")
+                s3.Bucket(transformed_asset_bucket).put_object(Key=transformed_file_name, Body=encoded_string)
+                response['status'] = 'File transformed'
+                response['name'] = transformed_file_name
+                response['summary']=''
+            else:
+                with open(original_file_name, "rb") as img_file:
+                    image_bytes = {"Bytes": img_file.read()}
+                if(moderate_image(image_bytes) is False):
+                     logger.info("Upload image to processed assets bucket")
+                     s3.Bucket(transformed_asset_bucket).put_object(Key=original_file_name, Body=image_bytes)
+                     response['status'] = 'File transformed'
+                     response['name'] = original_file_name
+                     response['summary']=''
+                 
     else:   
             logger.info("File already exists,skip transformation.")           
-    return response
+    return response
+
+def moderate_image(image_bytes)-> str:
+        isToxicImage = False
+        try: 
+            rekognition_response = rekognition_client.detect_moderation_labels(
+                            Image=image_bytes)
+            print(rekognition_response)
+            for label in rekognition_response['ModerationLabels']:
+                    if(label['Confidence'] > 0.60):
+                            isToxicImage=True
+                            print(f'Image failed moderation check, exit image uploading')
+                            break  
+        except Exception as exp:
+            print(f"Couldn't analyze image: {exp}")  
+        
+        return isToxicImage

lambda/aws-summarization-appsync-stepfn/document_reader/helper.py

@@ -12,7 +12,6 @@
 #
 import os
 from helper import check_file_exists,get_file_transformation
-import redis
 
 from update_summary_status import updateSummaryJobStatus
 from aws_lambda_powertools import Logger, Tracer, Metrics
@@ -34,40 +33,30 @@
 def handler(event, context: LambdaContext):
 
     logger.info(f"{event=}")
-    ignore_existing = event.get("ignore_existing", False)
 
     original_file_name = event["name"]
-    job_id = event["jobid"]  
+    job_id = event["jobid"] 
+    summary_model = event["summary_model"] 
+    language = event["language"]  
     response = {
-        "is_summary_available": False,
         "summary_job_id": job_id,
         "file_name": original_file_name,
-        "status": "Pending",
+        "status": "Working on generating the summary",
         "summary": "",
         "transformed_file_name":'',
+        "summary_model": summary_model,
+        "language": language,
     }
 
     logger.set_correlation_id(job_id)
     metrics.add_metadata(key='correlationId', value=job_id)
     tracer.put_annotation(key="correlationId", value=job_id)
 
-    filesummary = get_summary_from_cache(original_file_name) if not ignore_existing else None
 
-    if filesummary is not None:
-        metrics.add_metric(name="summary_cache_hit",unit=MetricUnit.Count, value=1)
-        response.update(
-            {
-                "file_name": original_file_name, 
-                "status": "Completed", 
-                "summary": filesummary,
-                "is_summary_available": True,
-            }
-        )
-    else:
-        metrics.add_metric(name="summary_llm_hit", unit=MetricUnit.Count, value=1)
-        transformed_file_name = original_file_name.replace(".pdf", ".txt")
+    metrics.add_metric(name="summary_llm_hit", unit=MetricUnit.Count, value=1)
+    transformed_file_name = original_file_name.replace(".pdf", ".txt")
 
-        if(is_file_tranformation_required):
+    if(is_file_tranformation_required):
              logger.info("File transformation required")
              transformed_file  = get_file_transformation(transformed_bucket_name, 
                                                          transformed_file_name,
@@ -79,13 +68,12 @@ def handler(event, context: LambdaContext):
                   "status": transformed_file['status'], 
                   "summary": transformed_file['summary'],
                   "transformed_file_name":transformed_file_name,
-                  "is_summary_available": False  
                 }
             )
-        else:
-             pdf_transformed_file = check_file_exists(transformed_bucket_name,
+    else:
+        pdf_transformed_file = check_file_exists(transformed_bucket_name,
                                                       transformed_file_name)
-             if pdf_transformed_file is False:
+        if pdf_transformed_file is False:
                 response.update(
                     {
                      "file_name": original_file_name, 
@@ -97,37 +85,8 @@ def handler(event, context: LambdaContext):
 
 
     logger.info({"document reader response:::": response})
-    updateSummaryJobStatus({'jobid': job_id, 'files':
-                            [{ 'status':response["status"],
-                              'name':response['file_name'] , 
-                              'summary':response["summary"] }]})
+    updateSummaryJobStatus({'summary_job_id': job_id,
+                            'status':response["status"],
+                            'name':response['file_name'] , 
+                            'summary':response["summary"] })
     return response
-
-@tracer.capture_method
-def get_summary_from_cache(file_name):
-
-    logger.info({"Searching Redis for cached summary file: "+file_name})
-    redis_host = os.environ.get("REDIS_HOST")
-    redis_port = os.environ.get("REDIS_PORT")
-    
-    logger.info(f"Redis host: {redis_host}")
-    logger.info(f"Redis port: {redis_port}")
-    
-    if redis_host is None or redis_port is None:
-        logger.exception({"Redis host or port is not set"})
-    else:
-        try:
-            logger.info({"Connecting Redis......"})
-            redis_client = redis.Redis(host=redis_host, port=redis_port)
-            fileSummary = redis_client.get(file_name)
-        except (ValueError, redis.ConnectionError) as e:
-            logger.exception({"An error occured while connecting to Redis" : e})
-            return
-
-        if fileSummary:
-            logger.info({"File summary found in cache: ": fileSummary})
-            return fileSummary.decode()
-
-
-    logger.info("File summary not found in cache, generating it from llm")
-

lambda/aws-summarization-appsync-stepfn/document_reader/lambda.py

@@ -45,29 +45,24 @@ def get_credentials(secret_id: str, region_name: str) -> str:
 def updateSummaryJobStatus(variables):
 
     logger.info(f"send  status variables :: {variables}")
+    summary = variables['summary']
+    
     query = """
         mutation updateSummaryJobStatus {
-            updateSummaryJobStatus(files: $files, summary_job_id: \"$jobid\") {
-                files {
+            updateSummaryJobStatus(summary_job_id: \"$summary_job_id\",
+            name: \"$name\",status: \"$status\",summary: \""""+summary+"""\",) {       
+                    summary_job_id
                     name
                     status
                     summary
-                }
-                summary_job_id
+                      
             }
         }
     """
 
-    query = query.replace("$jobid", variables['jobid'])
-    query = query.replace("$files", str(variables['files']).replace("\'", "\""))
-    query = query.replace("\"name\"", "name")
-    query = query.replace("\"status\"", "status")
-    query = query.replace("\"summary\"", "summary")
-    
-
-    # query = query.replace("\"file_name\"", "file_name")
-    # query = query.replace("\"status\"", "status")
-    query = query.replace("\n", "")
+    query = query.replace("$summary_job_id", variables['summary_job_id'])
+    query = query.replace("$name", variables['name'])
+    query = query.replace("$status", variables['status'])
 
     request = {'query':query}