deploy.cfn.yaml

AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  BucketNamePrefix:
    Type: String
    Default: my-bucket-prefix
    Description: Prefix for the random S3 bucket name
  CognitoUserPoolName:
    Type: String
    Default: 'myuserpool'
    Description: Name for the Cognito user pool
  UserPoolDomainName:
    Type: String
    Default: 'myuserdomain'
    Description: Name for the Cognito domain name

Resources:
  # init dynamodb table
  ProductReviewsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: ProductReviews
      AttributeDefinitions:
        - AttributeName: ProductID
          AttributeType: S
        - AttributeName: Timestamp
          AttributeType: N
      KeySchema:
        - AttributeName: ProductID
          KeyType: HASH
        - AttributeName: Timestamp
          KeyType: RANGE
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5

  ProductReviewsCNTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: ProductReviewsCN
      AttributeDefinitions:
        - AttributeName: ProductID
          AttributeType: S
        - AttributeName: Timestamp
          AttributeType: N
      KeySchema:
        - AttributeName: ProductID
          KeyType: HASH
        - AttributeName: Timestamp
          KeyType: RANGE
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5

  # init S3 bucket
  RandomBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub "${BucketNamePrefix}-${AWS::AccountId}-${AWS::Region}-${AWS::StackName}"

  # 创建 Cognito 用户池
  CognitoUserPool:
    Type: AWS::Cognito::UserPool
    Properties:
      UserPoolName: !Ref CognitoUserPoolName
      Policies:
        PasswordPolicy:
          MinimumLength: 8
          RequireLowercase: false
          RequireUppercase: false
          RequireNumbers: false
          RequireSymbols: false

  MyCognitoUserPoolDomain:
    Type: AWS::Cognito::UserPoolDomain
    Properties:
      Domain: !Ref UserPoolDomainName
      UserPoolId: !Ref CognitoUserPool

  CognitoUserPoolClientId:
    Type: AWS::Cognito::UserPoolClient
    Properties:
      UserPoolId: !Ref CognitoUserPool
      ClientName: MyUserPoolClient
      GenerateSecret: false
      ExplicitAuthFlows:
        - ADMIN_NO_SRP_AUTH

  # 创建 Cognito 身份池
  CognitoIdentityPool:
    Type: AWS::Cognito::IdentityPool
    Properties:
      IdentityPoolName: MyIdentityPool
      AllowUnauthenticatedIdentities: false
      CognitoIdentityProviders:
        - ProviderName: !Sub cognito-idp.${AWS::Region}.amazonaws.com/${CognitoUserPool}
          ClientId: !Ref CognitoUserPoolClientId

  # 创建 IAM 角色，授权 Cognito 用户池访问 OpenSearch
  MyAuthenticatedRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Federated: "cognito-identity.amazonaws.com"
            Action: "sts:AssumeRoleWithWebIdentity"
            Condition:
              StringEquals:
                "cognito-identity.amazonaws.com:aud": !Ref CognitoIdentityPool
      Policies:
        - PolicyName: "CognitoAccessPolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action: "*"
                Resource: "*"

  MyIdentityPoolRoleAttachment:
    Type: AWS::Cognito::IdentityPoolRoleAttachment
    Properties:
      IdentityPoolId: !Ref CognitoIdentityPool
      Roles:
        authenticated: !GetAtt MyAuthenticatedRole.Arn

  # 创建 OpenSearch (Elasticsearch) 域
  MyOpenSearchRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service: "es.amazonaws.com"
            Action: "sts:AssumeRole"
      ManagedPolicyArns:
        - "arn:aws:iam::aws:policy/AmazonESFullAccess" 
        - "arn:aws:iam::aws:policy/AmazonCognitoPowerUser"

  OpenSearchDomain:
    Type: AWS::OpenSearchService::Domain
    Properties:
      EngineVersion: OpenSearch_2.11
      ClusterConfig:
        InstanceCount: 1
        InstanceType: "m6g.large.search"
      NodeToNodeEncryptionOptions:
        Enabled: true
      EncryptionAtRestOptions:
        Enabled: true
      DomainEndpointOptions:
        EnforceHTTPS: true
      AdvancedSecurityOptions:
        Enabled: true
        InternalUserDatabaseEnabled: false
        MasterUserOptions:
          MasterUserARN: !GetAtt MyAuthenticatedRole.Arn
      CognitoOptions:
        Enabled: true
        IdentityPoolId: !Ref CognitoIdentityPool
        RoleArn: !GetAtt MyOpenSearchRole.Arn
        UserPoolId: !Ref CognitoUserPool
      EBSOptions:
        EBSEnabled: true
        VolumeType: gp2
        VolumeSize: 20

  LambdaFunctionProduct:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Runtime: python3.11
      Code:
        ZipFile: |
          import boto3
          import json
          from decimal import Decimal
          from datetime import datetime

          class DecimalEncoder(json.JSONEncoder):
              def default(self, obj):
                  if isinstance(obj, Decimal):
                      return str(obj)
                  return super().default(obj)

          # 获取全部商品列表
          def get_products(language):
              dynamodb = boto3.resource('dynamodb')
              if language == 'en':
                  table = dynamodb.Table('ProductDetails')
              else: 
                  table = dynamodb.Table('ProductDetailsCN')
              response = table.scan()
              return response['Items']

          # 获取指定商品的全部评价按时间排序
          def get_product_reviews(product_id,language):
              dynamodb = boto3.resource('dynamodb')
              if language == 'en':
                  table = dynamodb.Table('ProductReviews')
              else:
                  table = dynamodb.Table('ProductReviewsCN')
              response = table.query(
                  KeyConditionExpression='ProductID = :pk',
                  ExpressionAttributeValues={
                      ':pk': product_id,
                  },
                  ScanIndexForward=False  # 设置为 False 表示按排序键降序排序
              )
              return response['Items']

          # 对指定商品添加评价
          def add_product_review(product_id,product_name,rate,comment,user_id,language):
              dynamodb = boto3.resource('dynamodb')
              if language == 'en':
                  table = dynamodb.Table('ProductReviews')
              else:
                  table = dynamodb.Table('ProductReviewsCN')
              timestamp = int(datetime.now().timestamp())
              item = {
                  'ProductID': product_id,
                  'Timestamp': timestamp,
                  'Comment': comment,
                  'ProductName': product_name,
                  'Rating': rate,
                  'UserID': user_id
              }

              # 写入数据
              table.put_item(Item=item)
              return ['success']
              
          def lambda_handler(event, context):
              # 根据URL 参数判断请求
              query_params = event.get('queryStringParameters', {})
              if event.get('body') is not None:
                  post_params = json.loads(event.get('body'))
              else:
                  post_params = []
              # 接口请求类型
              type = query_params.get('type')
              # 语言版本 cn 中文 en 英文
              language = query_params.get('language')
              
              result = []
              if  type == 'get_products':
                  result = get_products(language)
              elif type == 'get_product_reviews':
                  result = get_product_reviews(query_params.get('product_id'),language)
              elif type == 'add_product_review':
                  product_id = post_params.get('product_id')
                  product_name = post_params.get('product_name')
                  rate = post_params.get('rate')
                  comment = post_params.get('comment')
                  user_id = post_params.get('user_id')
                  result = add_product_review(product_id,product_name,rate,comment,user_id,language)
              else:
                  result = ['request type error']
              # 返回api gateway 请求
              return {
                  'statusCode': 200,
                  'headers': {
                      "Access-Control-Allow-Origin": '*'
                  },
                  "isBase64Encoded": False,
                  'body': json.dumps(result, cls=DecimalEncoder)
              }

  LambdaFunctionBedrock:
    Type: AWS::Lambda::Function
    Properties:
      Handler: lambda_handler_bedrock.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Runtime: python3.12
      Code:
        ZipFile: |
          from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
          import boto3
          import json

          opensearch_host = ""

          def product_recommend(input_text,language):
              # 构建bedrock与 es客户端
              credentials = boto3.Session().get_credentials()
              auth = AWSV4SignerAuth(credentials, 'us-east-1', 'es')
              esClient = OpenSearch(
                  hosts = [{'host': opensearch_host, 'port': 443}],
                  http_auth = auth,
                  use_ssl = True,
                  verify_certs = True,
                  connection_class = RequestsHttpConnection,
                  pool_maxsize = 20
              )
              brt = boto3.client(service_name='bedrock-runtime')
              if language == 'en':
                  index = 'product-details-index-en'
              else: 
                  index = 'product-details-index-cn'
              query = {  
                "size": 10,
                "sort": [
                  {
                    "_score": {
                      "order": "desc"
                    }
                  }
                ],
                "_source": {
                  "includes": ["ProductName", "Category", "Description", "ProductID","Image"]
                },
                "query": {
                  "neural": {
                    "product_embedding": {
                      "query_text": input_text,
                      "model_id": "m6jIgowBXLzE-9O0CcNs",
                      "k": 13
                    }
                  }
                }
              }
              # 拼接 prompt
              es_response = esClient.search(
                  body = query,
                  index = index
              )
              try:
                  es_res = es_response['hits']['hits'][0]
              except json.JSONDecodeError as e:
                  return {
                      'statusCode': 400,
                      'body': json.dumps({'error': 'no result in system'})
                  }
              # llm 构建答案
              if language == 'en':
                  llm_prompt = 'Human: You are currently a professional clothing store assistant. The customer has asked you the following question: ' + input_text + ',You must respond to the customer inquiry using the provided information. Feel free to ask the customer for additional details if needed. Has a wide variety of clothing options, especially for women\'s fashion' + str(es_res) + ' Assistant:'
              else: 
                  llm_prompt = 'Human: 你现在是一个导购客服，需要帮助客户推荐商品，根据商品的描述信息，给客户推荐具体的商品名称和编号. 客户的问题如下: ' + input_text + ',你必须基于以下商品信息进行推荐.适当的时候如果客户问题不清晰，可以反问一些关键信息.有各种各样的服装选择，尤其是女性时尚' + str(es_res) + ' Assistant:'
              
              llm_request_body = json.dumps({
                  "prompt": llm_prompt,
                  "max_tokens_to_sample": 4000,
                  "temperature": 0.1,
                  "top_p": 0.9,
              })
              
              modelId = 'anthropic.claude-v2:1'
              accept = 'application/json'
              contentType = 'application/json'
              
              response = brt.invoke_model(body=llm_request_body, modelId=modelId, accept=accept, contentType=contentType)
              
              response_body = json.loads(response.get('body').read())
              
              llm_result = response_body.get('completion')
              return llm_result,es_response
              
          def reviews_analytis(input_text,language):
              # 构建bedrock与 es客户端
              credentials = boto3.Session().get_credentials()
              auth = AWSV4SignerAuth(credentials, 'us-east-1', 'es')
              esClient = OpenSearch(
                  hosts = [{'host': opensearch_host, 'port': 443}],
                  http_auth = auth,
                  use_ssl = True,
                  verify_certs = True,
                  connection_class = RequestsHttpConnection,
                  pool_maxsize = 20
              )
              brt = boto3.client(service_name='bedrock-runtime')
              if language == 'en':
                  index = 'product-reviews-index-en'
              else: 
                  index = 'product-reviews-index-cn'
              query = {  
                  "size" :50,
                  "_source": {
                    "includes": "combined_field"
                  },
                  "query": {
                      "neural": {
                        "product_reviews_embedding": {
                          "query_text": input_text,
                          "model_id": "m6jIgowBXLzE-9O0CcNs",
                          "k": 11
                        }
                      }
                    }
              }
              es_response = esClient.search(
                  body = query,
                  index = index
              )
              try:
                  es_res = es_response['hits']['hits'][0]
              except json.JSONDecodeError as e:
                  return {
                      'statusCode': 400,
                      'body': json.dumps({'error': 'no result in system'})
                  }
              # llm 构建答案
              if language == 'en':
                  llm_prompt = 'Human: You are now a customer service representative assisting customers in analyzing product reviews. Based on the historical comments about the product, you need to provide customers with a summary of the reviews, focusing primarily on the product rating and the emotional expressions conveyed in the comments. The customer\'s inquiries are as follows: ' + input_text + ',You must respond to the customer inquiry using the provided information. Feel free to ask the customer for additional details if needed.' + str(es_res) + ' Assistant:'
              else: 
                  llm_prompt = 'Human: 你现在是一个导购客服，需要帮助客户分析商品的评价，根据商品过去的评论信息，给客户做评论总结，主要关注商品的评分，评论内容的情绪表达. 客户的问题如下: ' + input_text + ',你必须基于以下商品评价信息进行总结.适当的时候如果客户问题不清晰，可以反问一些关键信息.' + str(es_res) + ' Assistant:'
              
              llm_request_body = json.dumps({
                  "prompt": llm_prompt,
                  "max_tokens_to_sample": 4000,
                  "temperature": 0.1,
                  "top_p": 0.9,
              })
              
              modelId = 'anthropic.claude-v2:1'
              accept = 'application/json'
              contentType = 'application/json'
              
              response = brt.invoke_model(body=llm_request_body, modelId=modelId, accept=accept, contentType=contentType)
              
              response_body = json.loads(response.get('body').read())
              
              llm_result = response_body.get('completion')
              return llm_result,es_response

          def lambda_handler(event, context):
              
              #接收参数 判断类型 
              query_params = event.get('queryStringParameters')
              # 接口请求类型
              type = query_params.get('type')
              # 语言版本 cn 中文 en 英文
              language = query_params.get('language')
              if event.get('body') is not None:
                  body = json.loads(event.get('body'))
              else:
                  body = []
              input_text = body.get('input_text')
              if  type == 'product_recommend':
                  llm_result,es_response = product_recommend(input_text,language)
              else:
                  llm_result,es_response = reviews_analytis(input_text,language)

              return {
                  'statusCode': 200,
                  'headers': {
                      "Access-Control-Allow-Origin": '*'
                  },
                  "isBase64Encoded": False,
                  'body': json.dumps({
                      'llm_result': llm_result,
                      'es_response': es_response
                  })
              }


  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service: "lambda.amazonaws.com"
            Action: "sts:AssumeRole"
      ManagedPolicyArns:
        - "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
        - "arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess"
        - "arn:aws:iam::aws:policy/AmazonOpenSearchServiceFullAccess"
        - "arn:aws:iam::aws:policy/AmazonBedrockFullAccess"

  ApiGatewayRestApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: shopworkshop
      Description: My REST API
      FailOnWarnings: true
      EndpointConfiguration:
        Types:
          - REGIONAL

  ApiGatewayResourceRecomm:
    Type: AWS::ApiGateway::Resource
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
      ParentId: !GetAtt ApiGatewayRestApi.RootResourceId
      PathPart: "recomm"

  ApiGatewayMethodRecomm:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
      ResourceId: !Ref ApiGatewayResourceRecomm
      HttpMethod: POST
      AuthorizationType: NONE
      Integration:
        IntegrationHttpMethod: POST
        Type: AWS_PROXY
        Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaFunctionBedrock.Arn}/invocations
  
  ApiGatewayResourceShop:
    Type: AWS::ApiGateway::Resource
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
      ParentId: !GetAtt ApiGatewayRestApi.RootResourceId
      PathPart: "shop"


  ApiGatewayMethodProduct:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
      ResourceId: !Ref ApiGatewayResourceShop
      HttpMethod: GET
      AuthorizationType: NONE
      Integration:
        IntegrationHttpMethod: GET
        Type: AWS_PROXY
        Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaFunctionProduct.Arn}/invocations

  ApiGatewayMethodReviews:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
      ResourceId: !Ref ApiGatewayResourceShop
      HttpMethod: POST
      AuthorizationType: NONE
      Integration:
        IntegrationHttpMethod: POST
        Type: AWS_PROXY
        Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaFunctionProduct.Arn}/invocations

  CloudWatchLogsRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service: "apigateway.amazonaws.com"
            Action: "sts:AssumeRole"
      Policies:
        - PolicyName: "CloudWatchLogsPolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action:
                  - "logs:CreateLogGroup"
                  - "logs:CreateLogStream"
                  - "logs:DescribeLogGroups"
                  - "logs:DescribeLogStreams"
                  - "logs:PutLogEvents"
                  - "logs:GetLogEvents"
                  - "logs:FilterLogEvents"
                Resource: "*"

  ApiGatewayCloudwatch:
    Type: AWS::ApiGateway::Account
    Properties:
      CloudWatchRoleArn: !GetAtt CloudWatchLogsRole.Arn

  ApiGatewayDeployment:
    Type: AWS::ApiGateway::Deployment
    DependsOn:
      - ApiGatewayMethodProduct
      - ApiGatewayMethodReviews
      - ApiGatewayMethodRecomm
    Properties:
      RestApiId: !Ref ApiGatewayRestApi
  ApiGatewayStage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: Prod
      RestApiId: !Ref ApiGatewayRestApi
      DeploymentId: !Ref ApiGatewayDeployment
      MethodSettings:
        - ResourcePath: "/*"
          HttpMethod: "*"
          DataTraceEnabled: true
          LoggingLevel: INFO
          MetricsEnabled: true
          ThrottlingBurstLimit: 5000
          ThrottlingRateLimit: 10000
          CachingEnabled: false
          CacheTtlInSeconds: 300
          CacheDataEncrypted: false
      AccessLogSetting:
        DestinationArn: !GetAtt AccessLogGroup.Arn
        Format: '$context.identity.sourceIp - - [$context.requestTime] "$context.httpMethod $context.resourcePath $context.protocol" $context.status $context.responseLength $context.requestId'

  AccessLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/apigateway/${ApiGatewayRestApi}/accesslogs

  DynamoDBETLRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: dynamodb-etl
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - es.amazonaws.com
                - osis-pipelines.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonBedrockFullAccess
        - arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess
        - arn:aws:iam::aws:policy/AmazonOpenSearchIngestionFullAccess
        - arn:aws:iam::aws:policy/AmazonOpenSearchIngestionReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonOpenSearchServiceFullAccess
        - arn:aws:iam::aws:policy/AmazonOpenSearchServiceReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonS3FullAccess

Outputs:
  S3BucketWebsiteURL:
    Description: "The URL of the S3 bucket website endpoint"
    Value: !Sub "http://${RandomBucket}.s3-website.${AWS::Region}.amazonaws.com"
  OpensearchIAMRoleARN:
    Description: "ARN of the Opensearch IAM Role"
    Value: !GetAtt MyAuthenticatedRole.Arn
  LambdaIAMRoleARN:
    Description: "ARN of the Lambda IAM Role"
    Value: !GetAtt LambdaExecutionRole.Arn
  DynamoDBETLRoleARN:
    Description: "ARN of the DynamoDB ETL Role"
    Value: !GetAtt DynamoDBETLRole.Arn
  OpenSearchDashboardsURL:
    Description: OpenSearch Dashboard URL
    Value: !Join 
      - ''
      - - 'https://'
        - !GetAtt OpenSearchDomain.DomainEndpoint
        - '/_dashboards'
  ApiUrl:
    Description: URL of the API Gateway endpoint
    Value: !Sub "https://${ApiGatewayRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/"