Introduction AWS offers a complete ecosystem of AI services that allows you to implement intelligent solutions without the need for deep machine learning expertise. This guide explores how to use these services to automate processes and build smarter applications.
AWS AI Services 1. Amazon Comprehend - Text Analysis Sentiment Analysis import boto3 import json def analyze_sentiment(text): """Analyze text sentiment using Comprehend""" comprehend = boto3.client('comprehend') response = comprehend.detect_sentiment( Text=text, LanguageCode='pt' ) return { 'sentiment': response['Sentiment'], 'confidence': response['SentimentScore'] } # Usage example text = "I am very satisfied with the company's service!" result = analyze_sentiment(text) print(f"Sentiment: {result['sentiment']}") print(f"Confidence: {result['confidence']}") Entity Extraction def extract_entities(text): """Extract named entities from text""" comprehend = boto3.client('comprehend') response = comprehend.detect_entities( Text=text, LanguageCode='pt' ) entities = [] for entity in response['Entities']: entities.append({ 'text': entity['Text'], 'type': entity['Type'], 'confidence': entity['Score'] }) return entities # Example text = "João Silva works at Amazon in São Paulo since 2020" entities = extract_entities(text) for entity in entities: print(f"{entity['text']} - {entity['type']} ({entity['confidence']:.2f})") 2. Amazon Rekognition - Image Analysis Object Detection def detect_objects_in_image(bucket_name, image_key): """Detect objects in an S3 image""" rekognition = boto3.client('rekognition') response = rekognition.detect_labels( Image={ 'S3Object': { 'Bucket': bucket_name, 'Name': image_key } }, MaxLabels=10, MinConfidence=80 ) objects = [] for label in response['Labels']: objects.append({ 'name': label['Name'], 'confidence': label['Confidence'], 'instances': len(label.get('Instances', [])) }) return objects Facial Recognition def detect_faces(bucket_name, image_key): """Detect faces in an image""" rekognition = boto3.client('rekognition') response = rekognition.detect_faces( Image={ 'S3Object': { 'Bucket': bucket_name, 'Name': image_key } }, Attributes=['ALL'] ) faces = [] for face in response['FaceDetails']: faces.append({ 'age_range': face['AgeRange'], 'gender': face['Gender']['Value'], 'emotions': [ { 'type': emotion['Type'], 'confidence': emotion['Confidence'] } for emotion in face['Emotions'] if emotion['Confidence'] > 50 ] }) return faces 3. Amazon Polly - Text-to-Speech def text_to_speech(text, output_bucket, output_key): """Convert text to audio using Polly""" polly = boto3.client('polly') s3 = boto3.client('s3') # Synthesize speech response = polly.synthesize_speech( Text=text, OutputFormat='mp3', VoiceId='Camila', # Brazilian Portuguese voice LanguageCode='pt-BR' ) # Save to S3 s3.put_object( Bucket=output_bucket, Key=output_key, Body=response['AudioStream'].read(), ContentType='audio/mpeg' ) return f"s3://{output_bucket}/{output_key}" # Example audio_url = text_to_speech( "Hello! This is an example of speech synthesis using Amazon Polly.", "my-audio-bucket", "speech/example.mp3" ) Practical Use Cases 1. Automatic Customer Feedback Analysis import boto3 from datetime import datetime import json class FeedbackAnalyzer: def __init__(self): self.comprehend = boto3.client('comprehend') self.dynamodb = boto3.resource('dynamodb') self.sns = boto3.client('sns') self.table = self.dynamodb.Table('customer-feedback') def process_feedback(self, feedback_text, customer_id): """Process customer feedback""" # Sentiment analysis sentiment_response = self.comprehend.detect_sentiment( Text=feedback_text, LanguageCode='pt' ) # Key topic extraction key_phrases_response = self.comprehend.detect_key_phrases( Text=feedback_text, LanguageCode='pt' ) # Prepare data for storage feedback_data = { 'feedback_id': f"{customer_id}_{int(datetime.now().timestamp())}", 'customer_id': customer_id, 'text': feedback_text, 'sentiment': sentiment_response['Sentiment'], 'sentiment_scores': sentiment_response['SentimentScore'], 'key_phrases': [ phrase['Text'] for phrase in key_phrases_response['KeyPhrases'] if phrase['Score'] > 0.8 ], 'timestamp': datetime.now().isoformat(), 'processed': True } # Save to DynamoDB self.table.put_item(Item=feedback_data) # Alert if negative feedback if sentiment_response['Sentiment'] == 'NEGATIVE': self.send_alert(feedback_data) return feedback_data def send_alert(self, feedback_data): """Send alert for negative feedback""" message = { 'alert_type': 'negative_feedback', 'customer_id': feedback_data['customer_id'], 'sentiment_score': feedback_data['sentiment_scores']['Negative'], 'key_issues': feedback_data['key_phrases'][:3], 'timestamp': feedback_data['timestamp'] } self.sns.publish( TopicArn='arn:aws:sns:region:account:customer-alerts', Message=json.dumps(message), Subject='Negative Feedback Detected' ) # Class usage analyzer = FeedbackAnalyzer() result = analyzer.process_feedback( "The product arrived defective and the customer service was terrible!", "customer_123" ) 2. Automatic Content Moderation class ContentModerator: def __init__(self): self.rekognition = boto3.client('rekognition') self.comprehend = boto3.client('comprehend') self.s3 = boto3.client('s3') def moderate_image(self, bucket_name, image_key): """Moderate image content""" # Detect inappropriate content moderation_response = self.rekognition.detect_moderation_labels( Image={ 'S3Object': { 'Bucket': bucket_name, 'Name': image_key } }, MinConfidence=60 ) inappropriate_content = [] for label in moderation_response['ModerationLabels']: inappropriate_content.append({ 'category': label['Name'], 'confidence': label['Confidence'], 'parent_category': label.get('ParentName', '') }) # Detect text in the image text_response = self.rekognition.detect_text( Image={ 'S3Object': { 'Bucket': bucket_name, 'Name': image_key } } ) detected_text = ' '.join([ text['DetectedText'] for text in text_response['TextDetections'] if text['Type'] == 'LINE' ]) # Analyze sentiment of detected text text_sentiment = None if detected_text: sentiment_response = self.comprehend.detect_sentiment( Text=detected_text, LanguageCode='pt' ) text_sentiment = sentiment_response['Sentiment'] return { 'image_key': image_key, 'inappropriate_content': inappropriate_content, 'detected_text': detected_text, 'text_sentiment': text_sentiment, 'approved': len(inappropriate_content) == 0, 'confidence_score': min([label['confidence'] for label in inappropriate_content]) if inappropriate_content else 100 } def moderate_text(self, text_content): """Moderate text content""" # Detect toxic language using Comprehend sentiment_response = self.comprehend.detect_sentiment( Text=text_content, LanguageCode='pt' ) # Prohibited words list (simplified example) prohibited_words = ['spam', 'scam', 'fraud'] contains_prohibited = any( word.lower() in text_content.lower() for word in prohibited_words ) return { 'text': text_content, 'sentiment': sentiment_response['Sentiment'], 'sentiment_scores': sentiment_response['SentimentScore'], 'contains_prohibited_words': contains_prohibited, 'approved': not contains_prohibited and sentiment_response['Sentiment'] != 'NEGATIVE' } # Usage example moderator = ContentModerator() # Moderate image image_result = moderator.moderate_image('content-bucket', 'user-uploads/image.jpg') print(f"Image approved: {image_result['approved']}") # Moderate text text_result = moderator.moderate_text("This is a normal comment about the product.") print(f"Text approved: {text_result['approved']}") 3. Intelligent Chatbot with Lex class IntelligentChatbot: def __init__(self): self.lex = boto3.client('lexv2-runtime') self.comprehend = boto3.client('comprehend') self.dynamodb = boto3.resource('dynamodb') self.conversation_table = self.dynamodb.Table('chatbot-conversations') def process_message(self, user_id, message, session_id=None): """Process user message""" if not session_id: session_id = f"{user_id}_{int(datetime.now().timestamp())}" # Analyze intent with Lex lex_response = self.lex.recognize_text( botId='your-bot-id', botAliasId='your-bot-alias-id', localeId='pt_BR', sessionId=session_id, text=message ) # Analyze message sentiment sentiment_response = self.comprehend.detect_sentiment( Text=message, LanguageCode='pt' ) # Prepare response based on intent intent_name = lex_response.get('sessionState', {}).get('intent', {}).get('name', 'Unknown') bot_response = lex_response.get('messages', [{}])[0].get('content', 'Sorry, I did not understand.') # Customize response based on sentiment if sentiment_response['Sentiment'] == 'NEGATIVE': bot_response = f"I can see you're frustrated. {bot_response} Would you like me to transfer you to a human agent?" # Save conversation conversation_data = { 'conversation_id': f"{session_id}_{int(datetime.now().timestamp())}", 'user_id': user_id, 'session_id': session_id, 'user_message': message, 'bot_response': bot_response, 'intent': intent_name, 'sentiment': sentiment_response['Sentiment'], 'confidence': lex_response.get('sessionState', {}).get('intent', {}).get('confirmationState', 'None'), 'timestamp': datetime.now().isoformat() } self.conversation_table.put_item(Item=conversation_data) return { 'response': bot_response, 'intent': intent_name, 'sentiment': sentiment_response['Sentiment'], 'session_id': session_id } def get_conversation_analytics(self, user_id): """Get conversation analytics""" response = self.conversation_table.query( IndexName='user-id-index', KeyConditionExpression='user_id = :user_id', ExpressionAttributeValues={':user_id': user_id} ) conversations = response['Items'] # Calculate metrics total_messages = len(conversations) sentiments = [conv['sentiment'] for conv in conversations] intents = [conv['intent'] for conv in conversations] return { 'total_messages': total_messages, 'sentiment_distribution': { 'positive': sentiments.count('POSITIVE'), 'negative': sentiments.count('NEGATIVE'), 'neutral': sentiments.count('NEUTRAL') }, 'top_intents': list(set(intents)), 'last_interaction': max([conv['timestamp'] for conv in conversations]) if conversations else None } # Usage example chatbot = IntelligentChatbot() # Process message response = chatbot.process_message( user_id="user_123", message="I need to cancel my order", session_id="session_456" ) print(f"Bot response: {response['response']}") print(f"Detected intent: {response['intent']}") Automation with Step Functions Document Processing Workflow { "Comment": "Automatic document processing workflow", "StartAt": "ExtractText", "States": { "ExtractText": { "Type": "Task", "Resource": "arn:aws:states:::aws-sdk:textract:startDocumentTextDetection", "Parameters": { "DocumentLocation": { "S3Object": { "Bucket.$": "$.bucket", "Name.$": "$.key" } } }, "Next": "WaitForExtraction" }, "WaitForExtraction": { "Type": "Wait", "Seconds": 10, "Next": "GetExtractionResults" }, "GetExtractionResults": { "Type": "Task", "Resource": "arn:aws:states:::aws-sdk:textract:getDocumentTextDetection", "Parameters": { "JobId.$": "$.JobId" }, "Next": "AnalyzeText" }, "AnalyzeText": { "Type": "Task", "Resource": "arn:aws:states:::lambda:invoke", "Parameters": { "FunctionName": "analyze-extracted-text", "Payload.$": "$" }, "Next": "ClassifyDocument" }, "ClassifyDocument": { "Type": "Task", "Resource": "arn:aws:states:::aws-sdk:comprehend:detectSentiment", "Parameters": { "Text.$": "$.extractedText", "LanguageCode": "pt" }, "Next": "StoreResults" }, "StoreResults": { "Type": "Task", "Resource": "arn:aws:states:::dynamodb:putItem", "Parameters": { "TableName": "processed-documents", "Item": { "documentId": {"S.$": "$.documentId"}, "extractedText": {"S.$": "$.extractedText"}, "sentiment": {"S.$": "$.Sentiment"}, "processedAt": {"S.$": "$$.State.EnteredTime"} } }, "End": true } } } Monitoring and Optimization CloudWatch Metrics for AI Services def monitor_ai_services(): """Monitor usage and performance of AI services""" cloudwatch = boto3.client('cloudwatch') # Custom metrics metrics = [ { 'MetricName': 'ComprehendRequests', 'Value': 1, 'Unit': 'Count', 'Dimensions': [ { 'Name': 'Service', 'Value': 'Comprehend' } ] }, { 'MetricName': 'RekognitionRequests', 'Value': 1, 'Unit': 'Count', 'Dimensions': [ { 'Name': 'Service', 'Value': 'Rekognition' } ] } ] cloudwatch.put_metric_data( Namespace='AI/Services', MetricData=metrics ) # Dashboard for monitoring dashboard_config = { "widgets": [ { "type": "metric", "properties": { "metrics": [ ["AI/Services", "ComprehendRequests"], ["AI/Services", "RekognitionRequests"] ], "period": 300, "stat": "Sum", "region": "us-east-1", "title": "AI Services Usage" } } ] } Conclusion AWS AI services democratize access to artificial intelligence, enabling developers to implement sophisticated solutions without deep ML expertise. The main advantages include:
...