Add DynamoDB database to log all Rekongition and A2I events for offline analytics use

Author: leschan

README.md

@@ -50,14 +50,16 @@ We use Parameter Store in two different ways. Firstly, we provide a set of seven
 
 We use two EventBridge rules to initiate Step Functions state machine runs. The first rule is based on a Systems Manager event pattern. The Systems Manager rule is triggered by changes to the Parameter Store and initiates the state machine to invoke a Lambda function to apply changes to the impacted resources. The second rule is a schedule rule. The schedule rule is triggered periodically to initiate the state machine to invoke a Lambda function to check for new model training.
 
+We use an [Amazon DynamoDB](https://aws.amazon.com/dynamodb/) NoSQL database to log all Rekognition and A2I events and results for performance analysis and model drift detection. Although we did NOT include an analytics feature in this example,  you can quickly deploy [Amazon QuickSight](https://aws.amazon.com/quicksight/) with the DynamoDB table as your source to visualize the performance of your model over time.
+
 We use a Step Functions state machine to orchestrate the ML workflow. The state machine initiates different processes based on events received from EventBridge and responses from Lambda. In addition, the state machine uses an internal process such as Wait to wait for model training and deployment to complete and Choice to evaluate for next tasks.
 
 We use an S3 bucket and a set of predefined folders to store training and inference images and model artifacts. Each folder has a dedicated purpose. The model operator uploads new images to the folder images_labeled_by_folder for training, and the model consumer uploads inference images to the folder images_for_detection for custom label detection.
 
 We use three different sets of Lambda functions:
 
 - The first set consists of two Lambda functions that build the Rekognition Custom Labels project and Amazon A2I human flow definition. These two Lambda functions are only used initially as part of the CloudFormation stack deployment process.
-- The second set of Lambda functions are invoked by the state machine to run Amazon Rekognition and Amazon A2I APIs, create a manifest file for training, collect labeled images for training, and manage system resources.
+- The second set of Lambda functions are invoked by the state machine to run Amazon Rekognition, A2I & DynamoDB APIs, create a manifest file for training, collect labeled images for training, and manage system resources.
 - The last set is a single Lambda function to redirect S3 PutObject events to the state machine.
 
 We use [Amazon Simple Notification Service](https://aws.amazon.com/sns/) (Amazon SNS) as a communication mechanism to alert the model operator and model consumer of relevant model training and detection events. All SNS messages are published by the corresponding Lambda functions.

assets/architecture.drawio

@@ -4,25 +4,49 @@
 s3 = boto3.client('s3')
 s3_resource = boto3.resource('s3')
 ssm = boto3.client('ssm')
+dynamodb = boto3.client('dynamodb')
 def get_parameter(parameter_name):
     response = ssm.get_parameter(Name=parameter_name)
     return json.loads(response['Parameter']['Value'])
+def append_human_results(dynamodb_table, detectlabel_request_id, label, new_key, submission_time):
+    response = dynamodb.update_item(
+        TableName=dynamodb_table,
+        Key={
+            'DetectLabelRequestId': {'S':detectlabel_request_id}
+        },
+        ExpressionAttributeNames={
+            '#HL':'HumanLabel',
+            '#S3':'CopiedS3ObjectKey',
+            '#AT': 'HumanLableCompletedDate'
+        },
+        ExpressionAttributeValues={
+            ':h': {'S':label},
+            ':s': {'S':new_key},
+            ':a': {'S':submission_time}
+        },
+        UpdateExpression='SET #HL=:h, #S3=:s, #AT=:a'
+    )
 def handler(event, context):
     parameter_name = os.environ['parameter_store_path'] + 'For-System-Use-Only'
     sys_vars = get_parameter(parameter_name)
+    dynamodb_table = sys_vars['dynamodb_table']
     bucket = event['s3event']['s3']['bucket']['name']
     key = event['s3event']['s3']['object']['key']
     response = s3.get_object(Bucket = bucket, Key = key)
     response = json.loads(response['Body'].read())
+    new_key=''
     if len(response['humanAnswers']) != 0:
         label = response['humanAnswers'][0]['answerContent']['crowd-image-classifier']['label']
+        submission_time = response['humanAnswers'][0]['submissionTime']
         if label != 'None of the Above':
             taskObject = response['inputContent']['taskObject']
             bucket, key = taskObject.replace("s3://", "").split("/", 1)
             copy_source = {'Bucket': bucket, 'Key': key}
             image_name = 'humanLoopName-' + response['humanLoopName'] + '-' + taskObject.split('/')[-1]
             new_key = 'images_labeled_by_folder/' + label + '/' + image_name
             s3_resource.meta.client.copy(copy_source, sys_vars['s3_bucket'], new_key)
+    detectlabel_request_id=response['inputContent']['detectLabelRequestId']
+    append_human_results(dynamodb_table, detectlabel_request_id, label, new_key, submission_time)
     response['s3event'] = event['s3event']
     return {
         'message': response

assets/architecture.png

@@ -1,35 +1,72 @@
 import json
 import os
 import boto3
-import uuid
-sns = boto3.client('sns')
-a2i = boto3.client('sagemaker-a2i-runtime')
 ssm = boto3.client('ssm')
-def get_parameter(parameter_name):
-    response = ssm.get_parameter(Name=parameter_name)
-    parameter_value = json.loads(response['Parameter']['Value'])
-    return parameter_value
+sns = boto3.client('sns')
+rekognition = boto3.client('rekognition')
+dynamodb = boto3.client('dynamodb')
+def get_parameters():
+    response = ssm.get_parameters_by_path(
+        Path=os.environ['parameter_store_path'],
+        Recursive=True
+    )
+    parameter_store = {}
+    for parameter in response['Parameters']:
+        parameter_name = (parameter['Name'].split('/'))[-1]
+        parameter_store[parameter_name] = parameter['Value']
+    return parameter_store
+def store_detection_results(dynamodb_table, detectlabel_request_id, detectlabel_date, s3_bucket_name, s3_object_key, s3_object_eTag, project_version, detected_label, confidence_level, minimum_confidence_level, A2I):
+    response = dynamodb.put_item(
+        TableName=dynamodb_table,
+        Item={
+            'DetectLabelRequestId': {'S':detectlabel_request_id},
+            'DetectLabelDate': {'S':detectlabel_date},
+            'S3Bucket': {'S':s3_bucket_name},
+            'S3ObjectKey': {'S':s3_object_key},
+            'S3ObjectEtag': {'S':s3_object_eTag},
+            'ProjectVersionArn': {'S':project_version},
+            'DectectedLabel': {'S':detected_label},
+            'DetectedConfidenceLevel': {'N':str(confidence_level)},
+            'MinimumConfidenceLevel': {'N':str(minimum_confidence_level)},
+            'A2IEnabled': {'BOOL':A2I}
+        }
+    )
 def publish_message(sns_subject, sns_message, topic_arn):
     response = sns.publish(
         TopicArn=topic_arn,
         Message=sns_message,
         Subject=sns_subject
     )
 def handler(event, context):
-    parameter_name = os.environ['parameter_store_path'] + 'For-System-Use-Only'
-    sys_vars = get_parameter(parameter_name)
-    response = a2i.start_human_loop(
-        HumanLoopName = str(uuid.uuid4()),
-        FlowDefinitionArn = sys_vars['flow_definition_arn'],
-        HumanLoopInput = {
-            'InputContent': json.dumps({
-                'initialValue': event['message']['CustomLabels'][0]['Confidence'],
-                'taskObject': 's3://'+event['message']['s3event']['s3']['bucket']['name']+'/'+event['message']['s3event']['s3']['object']['key']
-            })
-        }
+    parameter_store = get_parameters()
+    a2i = True if parameter_store['Enable-A2I-Workflow'].lower().capitalize() == 'True' else False
+    sys_vars = json.loads(parameter_store['For-System-Use-Only'])
+    project_version = sys_vars['rekognition_project_version_arn']
+    dynamodb_table = sys_vars['dynamodb_table']
+    minimum_confidence_level=parameter_store['Minimum-Label-Detection-Confidence']
+    s3_bucket_name=event['s3event']['s3']['bucket']['name']
+    s3_object_key=event['s3event']['s3']['object']['key']
+    s3_object_eTag=event['s3event']['s3']['object']['eTag']
+    s3_event_time=event['s3event']['eventTime']
+    response = rekognition.detect_custom_labels(
+        ProjectVersionArn=project_version,
+        Image={
+            'S3Object': {
+                'Bucket': s3_bucket_name,
+                'Name': s3_object_key
+            }
+        },
+        MaxResults=1,
+        MinConfidence=0
     )
-    response['s3event'] = event['message']['s3event']
-    publish_message('A2I Human Loop Initiated', json.dumps(response), sys_vars['sns-topic'])
+    confidence_level = response['CustomLabels'][0]['Confidence']
+    detected_label = response['CustomLabels'][0]['Name']
+    detectlabel_request_id = response['ResponseMetadata']['RequestId']
+    if confidence_level < float(parameter_store['Minimum-Label-Detection-Confidence']):
+        response['A2I'] = True if a2i else False
+    store_detection_results(dynamodb_table, detectlabel_request_id, s3_event_time, s3_bucket_name, s3_object_key, s3_object_eTag, project_version, detected_label, confidence_level, minimum_confidence_level, a2i)
+    publish_message('Rekgnition Custom Labels Detection Invoked', json.dumps(response), sys_vars['sns-topic'])
+    response['s3event'] = event['s3event']
     return {
         'message': response
     }

lambda/a2i-human-loop.py

@@ -1,10 +1,10 @@
 import json
 import os
 import boto3
-import uuid
 ssm = boto3.client('ssm')
 sns = boto3.client('sns')
 rekognition = boto3.client('rekognition')
+dynamodb = boto3.client('dynamodb')
 def get_parameters():
     response = ssm.get_parameters_by_path(
         Path=os.environ['parameter_store_path'],
@@ -15,6 +15,22 @@ def get_parameters():
         parameter_name = (parameter['Name'].split('/'))[-1]
         parameter_store[parameter_name] = parameter['Value']
     return parameter_store
+def store_detection_results(dynamodb_table, detectlabel_request_id, detectlabel_date, s3_bucket_name, s3_object_key, s3_object_eTag, project_version, detected_label, confidence_level, minimum_confidence_level, A2I):
+    response = dynamodb.put_item(
+        TableName=dynamodb_table,
+        Item={
+            'DetectLabelRequestId': {'S':detectlabel_request_id},
+            'DetectLabelDate': {'S':detectlabel_date},
+            'S3Bucket': {'S':s3_bucket_name},
+            'S3ObjectKey': {'S':s3_object_key},
+            'S3ObjectEtag': {'S':s3_object_eTag},
+            'ProjectVersionArn': {'S':project_version},
+            'DectectedLabel': {'S':detected_label},
+            'DetectedConfidenceLevel': {'N':str(confidence_level)},
+            'MinimumConfidenceLevel': {'N':str(minimum_confidence_level)},
+            'A2IEnabled': {'BOOL':A2I}
+        }
+    )
 def publish_message(sns_subject, sns_message, topic_arn):
     response = sns.publish(
         TopicArn=topic_arn,
@@ -23,25 +39,33 @@ def publish_message(sns_subject, sns_message, topic_arn):
     )
 def handler(event, context):
     parameter_store = get_parameters()
+    a2i = True if parameter_store['Enable-A2I-Workflow'].lower().capitalize() == 'True' else False
     sys_vars = json.loads(parameter_store['For-System-Use-Only'])
+    project_version = sys_vars['rekognition_project_version_arn']
+    dynamodb_table = sys_vars['dynamodb_table']
+    minimum_confidence_level=parameter_store['Minimum-Label-Detection-Confidence']
+    s3_bucket_name=event['s3event']['s3']['bucket']['name']
+    s3_object_key=event['s3event']['s3']['object']['key']
+    s3_object_eTag=event['s3event']['s3']['object']['eTag']
+    s3_event_time=event['s3event']['eventTime']
     response = rekognition.detect_custom_labels(
-        ProjectVersionArn=sys_vars['rekognition_project_version_arn'],
+        ProjectVersionArn=project_version,
         Image={
             'S3Object': {
-                'Bucket': event['s3event']['s3']['bucket']['name'],
-                'Name': event['s3event']['s3']['object']['key']
+                'Bucket': s3_bucket_name,
+                'Name': s3_object_key
             }
         },
         MaxResults=1,
         MinConfidence=0
     )
     confidence_level = response['CustomLabels'][0]['Confidence']
-    publish_message('Rekgnition Custom Labels Detection Invoked', json.dumps(response), sys_vars['sns-topic'])
+    detected_label = response['CustomLabels'][0]['Name']
+    detectlabel_request_id = response['ResponseMetadata']['RequestId']
     if confidence_level < float(parameter_store['Minimum-Label-Detection-Confidence']):
-        if parameter_store['Enable-A2I-Workflow'] in ['True', 'true', 'TRUE']:
-            response['A2I'] = True
-        else:
-            response['A2I'] = False
+        response['A2I'] = True if a2i else False
+    store_detection_results(dynamodb_table, detectlabel_request_id, s3_event_time, s3_bucket_name, s3_object_key, s3_object_eTag, project_version, detected_label, confidence_level, minimum_confidence_level, a2i)
+    publish_message('Rekgnition Custom Labels Detection Invoked', json.dumps(response), sys_vars['sns-topic'])
     response['s3event'] = event['s3event']
     return {
         'message': response