Amazon SESとAmazon Bedrockで問い合わせメールの一次対応を自動化する

lambda_function.py

import boto3
import json
import os
import urllib.parse
import email
import email.policy
import datetime
import uuid

sender_email_address = os.environ['SENDER_EMAIL_ADDRESS']
dynamodb_table_name  = os.environ['DYNAMODB_TABLE_NAME']

s3              = boto3.client('s3')
dynamodb        = boto3.client('dynamodb')
ses             = boto3.client('sesv2')
bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1')


######################################################################
# Lambdaハンドラーメソッド
######################################################################
def lambda_handler(event, context):
    # for debug
    print(f'boto3 version: {boto3.__version__}')
    print(json.dumps(event))

    # イベントデータからS3バケット名とオブジェクトキーを取得する
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    object_key_name = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    print(f'bucket_name: {bucket_name}')
    print(f'object_key_name: {object_key_name}')

    # S3に保存されたメールを読み込む
    response = get_email(bucket_name, object_key_name)
    email_date = response['Date']
    email_from = response['From']
    email_body = response['Body']
    print(f'email_date: {email_date}')
    print(f'email_from: {email_from}')
    print(f'email_body: {email_body}')

    # メール文面から本文のみを抽出する
    main_text = extract_main_text(email_body)
    print(f'main_text: {main_text}')

    # 問い合わせ内容から問い合わせカテゴリを分類する
    inquiry_categiry = categorize_inquiry(main_text)
    print(f'inquiry_categiry: {inquiry_categiry}')

    # カテゴリに応じて処理を分岐
    if '質問' in inquiry_categiry:
        # 「質問」の場合のみ、質問に対する回答を生成して、回答から返信文を作成する
        answer_text = answer_inquiry(main_text)
        response_text = \
            'このたびはお問い合わせ頂き、ありがとうございます。\r\nお問い合わせ頂きました件について以下の通り回答いたします。' + \
            '\r\n\r\n' + answer_text + '\r\n\r\n回答は以上でございます。'
    elif '要望' in inquiry_categiry:
        # 「質問」以外の場合、定形の返信文を返す (以下同様)
        answer_text = ''
        response_text = 'このたびは貴重なご意見を頂き、誠にありがとうございます。\r\nお客様から頂いたご意見は、弊社サービスの改善に役立たせて頂きます。'
    elif '苦情' in inquiry_categiry:
        answer_text = ''
        response_text = 'このたびはご不快な思いをさせてしまい大変申し訳ございませんでした。\r\nお客様のご指摘を真摯に受け止め、今後の再発防止に努めて参ります。'
    else:
        answer_text = ''
        response_text = 'このたびはお問い合わせ頂き、ありがとうございました。'

    print(f'answer_text: {answer_text}')
    print(f'response_text: {response_text}')

    # 返信メールを送信する
    send_email(email_from, response_text)

    # 問い合わせ内容をデータベースへ登録する
    inquiry_id = str(uuid.uuid4())  # 主キーとして一意なIDを生成
    write_database(inquiry_id, email_date, email_from, inquiry_categiry, main_text, answer_text)


######################################################################
# メールを読み込む
######################################################################
def get_email(bucket_name: str, object_key_name: str) -> dict:
    # S3オブジェクトを読み込む
    response = s3.get_object(
        Bucket=bucket_name,
        Key=object_key_name,
    )
    print(response)

    # 読み込んだデータからメールメッセージ全体を取り出す
    object_body = response['Body'].read()
    email_message = email.message_from_bytes(object_body, policy=email.policy.default)

    # メールのヘッダを取り出す
    email_header_date = str(email_message.get('date', ''))
    email_header_from = str(email_message.get('from', ''))
    email_header_subject = str(email_message.get('subject', ''))

    # メールの本文を取り出す
    email_body = email_message.get_body(preferencelist=('plain', 'html'))
    email_body_content = email_body.get_content()

    return {
        'Date'   : email_header_date,
        'From'   : email_header_from,
        'Subject': email_header_subject,
        'Body'   : email_body_content,
    }


######################################################################
# メール文面から本文のみを抽出する
######################################################################
def extract_main_text(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下のメール文面から冒頭の挨拶や署名などを除去して、本文のみを抽出してください。' + \
        '結果のみを発言してください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text


######################################################################
# 問い合わせ内容をカテゴリに分類する
######################################################################
def categorize_inquiry(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下の文章を「質問」「要望」「苦情」のカテゴリのいずれかに分類してください。' + \
        '結果のみを発言してください。' + \
        'いずれのカテゴリにも属さない場合は「その他」と答えてください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text


######################################################################
# 問い合わせの質問に回答する
######################################################################
def answer_inquiry(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下の質問に回答してください。' + \
        '回答内容のみを発言してください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text


######################################################################
# Bedrockを使用して回答を生成する (Anthropic Claude 2)
######################################################################
def generate_answer(prompt_text :str, inquiry_text: str) -> str:
    # プロンプトの指定
    prompt = f'\n\nHuman: {prompt_text}\n---\n{inquiry_text}\n\nAssistant:'

    # 各種パラメーターの指定
    model_id = 'anthropic.claude-v2'
    accept = 'application/json'
    contentType = 'application/json'

    # リクエストBODYを組み立てる
    body = json.dumps({
        'prompt': prompt,
        'max_tokens_to_sample': 400,
    })

    # Bedrock APIの呼び出し
    response = bedrock_runtime.invoke_model(
    	modelId=model_id,
    	accept=accept,
    	contentType=contentType,
        body=body,
    )
    print(response)

    # APIレスポンスから応答テキストを取り出す
    answer_text = json.loads(response.get('body').read()).get('completion')

    return answer_text


######################################################################
# メールを送信する
######################################################################
def send_email(dst_addr:str, body:str) -> None:
    src_addr = sender_email_address
    subject  = 'お問い合わせ頂きました件につきまして'
    body     = '当メールはサポート窓口から自動送信しております。\r\n\r\n' + body

    # SESを使用してメール送信を行う
    response = ses.send_email(
        FromEmailAddress=src_addr,
        Destination={
            'ToAddresses': [
                dst_addr,
            ],
        },
        Content={
            'Simple': {
                'Subject': {
                    'Data': subject,
                    'Charset': 'utf-8',
                },
                'Body': {
                    'Text': {
                        'Data': body,
                        'Charset': 'utf-8',
                    },
                },
            },
        },
    )
    print(response)


######################################################################
# 問い合わせ内容と回答をDynamoDBへ書き込む
######################################################################
def write_database(inquiry_id: str, date: str, email_address: str, categiry: str, inquiry: str, answer: str) -> None:
    # DynamoDBへアイテムを新規作成する
    response = dynamodb.put_item(
        TableName=dynamodb_table_name,
        Item={
            'InquiryID': {
                'S': inquiry_id,
            },
            'Date': {
                'S': date,
            },
            'EmailAddress': {
                'S': email_address,
            },
            'Category': {
                'S': categiry,
            },
            'Inquiry': {
                'S': inquiry,
            },
            'Answer': {
                'S': answer,
            },
        },
    )
    print(response)

lambda_function.py

######################################################################
# Lambdaハンドラーメソッド
######################################################################
def lambda_handler(event, context):
    # for debug
    print(f'boto3 version: {boto3.__version__}')
    print(json.dumps(event))

    # イベントデータからS3バケット名とオブジェクトキーを取得する
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    object_key_name = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    print(f'bucket_name: {bucket_name}')
    print(f'object_key_name: {object_key_name}')

    # S3に保存されたメールを読み込む
    response = get_email(bucket_name, object_key_name)
    email_date = response['Date']
    email_from = response['From']
    email_body = response['Body']
    print(f'email_date: {email_date}')
    print(f'email_from: {email_from}')
    print(f'email_body: {email_body}')

    # メール文面から本文のみを抽出する
    main_text = extract_main_text(email_body)
    print(f'main_text: {main_text}')

    # 問い合わせ内容から問い合わせカテゴリを分類する
    inquiry_categiry = categorize_inquiry(main_text)
    print(f'inquiry_categiry: {inquiry_categiry}')

    # カテゴリに応じて処理を分岐
    if '質問' in inquiry_categiry:
        # 「質問」の場合のみ、質問に対する回答を生成して、回答から返信文を作成する
        answer_text = answer_inquiry(main_text)
        response_text = \
            'このたびはお問い合わせ頂き、ありがとうございます。\r\nお問い合わせ頂きました件について以下の通り回答いたします。' + \
            '\r\n\r\n' + answer_text + '\r\n\r\n回答は以上でございます。'
    elif '要望' in inquiry_categiry:
        # 「質問」以外の場合、定形の返信文を返す (以下同様)
        answer_text = ''
        response_text = 'このたびは貴重なご意見を頂き、誠にありがとうございます。\r\nお客様から頂いたご意見は、弊社サービスの改善に役立たせて頂きます。'
    elif '苦情' in inquiry_categiry:
        answer_text = ''
        response_text = 'このたびはご不快な思いをさせてしまい大変申し訳ございませんでした。\r\nお客様のご指摘を真摯に受け止め、今後の再発防止に努めて参ります。'
    else:
        answer_text = ''
        response_text = 'このたびはお問い合わせ頂き、ありがとうございました。'

    print(f'answer_text: {answer_text}')
    print(f'response_text: {response_text}')

    # 返信メールを送信する
    send_email(email_from, response_text)

    # 問い合わせ内容をデータベースへ登録する
    inquiry_id = str(uuid.uuid4())  # 主キーとして一意なIDを生成
    write_database(inquiry_id, email_date, email_from, inquiry_categiry, main_text, answer_text)

lambda_function.py

######################################################################
# メールを読み込む
######################################################################
def get_email(bucket_name: str, object_key_name: str) -> dict:
    # S3オブジェクトを読み込む
    response = s3.get_object(
        Bucket=bucket_name,
        Key=object_key_name,
    )
    print(response)

    # 読み込んだデータからメールメッセージ全体を取り出す
    object_body = response['Body'].read()
    email_message = email.message_from_bytes(object_body, policy=email.policy.default)

    # メールのヘッダを取り出す
    email_header_date = str(email_message.get('date', ''))
    email_header_from = str(email_message.get('from', ''))
    email_header_subject = str(email_message.get('subject', ''))

    # メールの本文を取り出す
    email_body = email_message.get_body(preferencelist=('plain', 'html'))
    email_body_content = email_body.get_content()

    return {
        'Date'   : email_header_date,
        'From'   : email_header_from,
        'Subject': email_header_subject,
        'Body'   : email_body_content,
    }

lambda_function.py

######################################################################
# メール文面から本文のみを抽出する
######################################################################
def extract_main_text(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下のメール文面から冒頭の挨拶や署名などを除去して、本文のみを抽出してください。' + \
        '結果のみを発言してください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text


######################################################################
# 問い合わせ内容をカテゴリに分類する
######################################################################
def categorize_inquiry(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下の文章を「質問」「要望」「苦情」のカテゴリのいずれかに分類してください。' + \
        '結果のみを発言してください。' + \
        'いずれのカテゴリにも属さない場合は「その他」と答えてください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text


######################################################################
# 問い合わせの質問に回答する
######################################################################
def answer_inquiry(input_text: str) -> str:
    # プロンプト
    prompt_text = \
        '以下の質問に回答してください。' + \
        '回答内容のみを発言してください。'

    # Bedrockを使用して回答を生成する
    answer_text = generate_answer(prompt_text, input_text)

    return answer_text

lambda_function.py

######################################################################
# Bedrockを使用して回答を生成する (Anthropic Claude 2)
######################################################################
def generate_answer(prompt_text :str, inquiry_text: str) -> str:
    # プロンプトの指定
    prompt = f'\n\nHuman: {prompt_text}\n---\n{inquiry_text}\n\nAssistant:'

    # 各種パラメーターの指定
    model_id = 'anthropic.claude-v2'
    accept = 'application/json'
    contentType = 'application/json'

    # リクエストBODYを組み立てる
    body = json.dumps({
        'prompt': prompt,
        'max_tokens_to_sample': 400,
    })

    # Bedrock APIの呼び出し
    response = bedrock_runtime.invoke_model(
    	modelId=model_id,
    	accept=accept,
    	contentType=contentType,
        body=body,
    )
    print(response)

    # APIレスポンスから応答テキストを取り出す
    answer_text = json.loads(response.get('body').read()).get('completion')

    return answer_text

lambda_function.py

######################################################################
# メールを送信する
######################################################################
def send_email(dst_addr:str, body:str) -> None:
    src_addr = sender_email_address
    subject  = 'お問い合わせ頂きました件につきまして'
    body     = '当メールはサポート窓口から自動送信しております。\r\n\r\n' + body

    # SESを使用してメール送信を行う
    response = ses.send_email(
        FromEmailAddress=src_addr,
        Destination={
            'ToAddresses': [
                dst_addr,
            ],
        },
        Content={
            'Simple': {
                'Subject': {
                    'Data': subject,
                    'Charset': 'utf-8',
                },
                'Body': {
                    'Text': {
                        'Data': body,
                        'Charset': 'utf-8',
                    },
                },
            },
        },
    )
    print(response)

lambda_function.py

######################################################################
# 問い合わせ内容と回答をDynamoDBへ書き込む
######################################################################
def write_database(inquiry_id: str, date: str, email_address: str, categiry: str, inquiry: str, answer: str) -> None:
    # DynamoDBへアイテムを新規作成する
    response = dynamodb.put_item(
        TableName=dynamodb_table_name,
        Item={
            'InquiryID': {
                'S': inquiry_id,
            },
            'Date': {
                'S': date,
            },
            'EmailAddress': {
                'S': email_address,
            },
            'Category': {
                'S': categiry,
            },
            'Inquiry': {
                'S': inquiry,
            },
            'Answer': {
                'S': answer,
            },
        },
    )
    print(response)