AWSを活用したAIサービス開発

昨年12月6日に開催された石川県中小企業診断士会AIシンポジウムで中小企業診断士４名による講演とパネルディスカッションを開催した。

私は「AWSを活用したAIサービス開発」と題して講演を行った。

50名ほどの方にお話を聞いていただき、その中でご紹介したLINE公式アカウントに組み込んで生成AIの活用事例を実際に簡単に触れていただいた。

生成AI未経験者がBedrockを組み込んだLINEチャットBotをつくってみたを参考にanthropic.claude-instant-v1のモデルで動作させられるようにしたもの。

ディレクトリ構成

├── linebot_reply
│   ├── __init__.py
│   └── app.py
├── python
│   └── linebot
│       ├── __about__.py
│       ├── __init__.py
│       ├── aiohttp_async_http_client.py
│       ├── api.py
│       ├── async_api.py
│       ├── async_http_client.py
│       ├── constants
│       ├── deprecations.py
│       ├── exceptions.py
│       ├── http_client.py
│       ├── models
│       ├── utils.py
│       ├── v3
│       └── webhook.py
├── samconfig.toml
└── template.yaml

app.py

import os
import sys
import json
import boto3
import time
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key, Attr

from linebot import (
    LineBotApi, WebhookHandler
)
from linebot.models import (
    MessageEvent, TextMessage, TextSendMessage,
)
from linebot.exceptions import (
    LineBotApiError, InvalidSignatureError
)
import logging

logger = logging.getLogger()
logger.setLevel(logging.ERROR)

# LINE認証情報をLambdaの環境変数から取得
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    logger.error('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    logger.error('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

# LINE APIクライアントの初期化
line_bot_api = LineBotApi(channel_access_token)
handler = WebhookHandler(channel_secret)

# DynamoDB設定
table_name = os.getenv('MESSAGE_TABLE_NAME', None)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)

# Bedrockランタイムのクライアント作成
bedrock_runtime = boto3.client(service_name='bedrock-runtime')

def lambda_handler(event, context):
    if "x-line-signature" in event["headers"]:
        signature = event["headers"]["x-line-signature"]
    elif "X-Line-Signature" in event["headers"]:
        signature = event["headers"]["X-Line-Signature"]
    body = event["body"]
    ok_json = {
        "isBase64Encoded": False,
        "statusCode": 200,
        "headers": {}, 
        "body": ""
    }
    error_json = {
        "isBase64Encoded": False,
        "statusCode": 500,
        "headers": {},
        "body": "Error"
    }
    try:
        handler.handle(body, signature)
    except LineBotApiError as e:
        logger.error("Got exception from LINE Messaging API: %s\n" % e.message)
        for m in e.error.details:
            logger.error("  %s: %s" % (m.property, m.message))
        return error_json
    except InvalidSignatureError:
        return error_json
    return ok_json

# Webhookから送られてきたイベントの処理
@handler.add(MessageEvent, message=TextMessage)
def message(line_event):
    # LINEからのイベント情報を取得
    user_message = line_event.message.text
    user_message_type = line_event.message.type
    user_id = line_event.source.user_id
    timestamp = int(line_event.timestamp)

    # ユーザーの会話履歴を取得
    pre_user_message, pre_bot_message = get_user_message_history(user_id)
    #pre_message_history = f"User:{pre_user_message}\nBot:{pre_bot_message}"

    # 会話履歴と新しいメッセージを結合してプロンプトを作成
    #prompt = f"{pre_message_history}User:  {user_message}\nBot:"

    pre_message_history = f"Human:{pre_user_message}\n\nAssistant:{pre_bot_message}"
    # 会話履歴と新しいメッセージを結合してプロンプトを作成
    prompt = f"{pre_message_history}\n\nHuman:  {user_message}\n\nAssistant:"

    # プロンプトをclaudeで処理する
    bot_message = handle_bedrock_claude(prompt)
    # プロンプトをtitan-text-expressで処理する
    #bot_message = handle_bedrock_titan(prompt)

    bot_message_type = 'text'

    # DynamoDBにユーザーの会話を記録
    save_conversation_to_dynamodb(user_id, user_message_type, user_message, bot_message_type, bot_message) 

    # LINEユーザーに応答を返す
    line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=bot_message))


# DynamoDBからユーザーの会話履歴を取得
def get_user_message_history(user_id):
    try:
        response = table.query(
            KeyConditionExpression=Key('userId').eq(user_id),
            ScanIndexForward=False,
            Limit=1
        )
        if 'Items' in response and len(response['Items']) > 0:
            user_message = response['Items'][0]['userMessage']
            bot_message = response['Items'][0]['botMessage']
            return user_message, bot_message
        else:
            return "",""
    except ClientError as e:
        logger.error("DynamoDB query failed: {}".format(e.response['Error']['Message']))
        return "",""

# DynamoDBにユーザーの会話を記録
def save_conversation_to_dynamodb(user_id, user_message_type, user_message ,bot_message_type ,bot_message):
    timestamp = int(time.time())
    try:
        response = table.put_item(
            Item={
                'userId': user_id,
                'requestDate': timestamp,
                'userMessage': user_message,
                'userMessageType': user_message_type,
                'botMessage': bot_message,
                'botMessageType': bot_message_type,
                'botActionType': 'reply' 
            }
        )
        logger.info("DynamoDB save successful.")
    except Exception as e:
        logger.error("Error saving to DynamoDB: {}".format(e))

# プロンプトをtitan-text-expressで処理する
def handle_bedrock_titan(prompt):
    # Bedrock Runtimeを使用してAI応答を生成
    response = bedrock_runtime.invoke_model(
        body=json.dumps({
            "inputText": prompt,
            "textGenerationConfig": {
                "temperature": 0.6,  
                "topP": 0.999,
                "maxTokenCount": 300,
                "stopSequences": ["User:"]
            }
        }),
        modelId="amazon.titan-text-express-v1",
        contentType="application/json",
        accept="*/*"
    )

    # 応答をJSON形式で取得し、テキストメッセージを取り出す
    response_body = json.loads(response.get('body').read())
    output_txt = response_body['results'][0]['outputText']
    bot_message = output_txt.strip() #response取得時、空白文字が入るため、削除
    return bot_message

# プロンプトをClaudeで処理する
def handle_bedrock_claude(prompt):
    response = bedrock_runtime.invoke_model(
        body=json.dumps({
            "prompt": prompt,
            "max_tokens_to_sample": 3000,
            "temperature": 0.6,
            "top_k": 250,
            "top_p": 0.999,
            "stop_sequences": ["\n\nHuman:"],
            "anthropic_version": "bedrock-2023-05-31"
        }),
        modelId='anthropic.claude-instant-v1',
        contentType='application/json',
        accept='*/*'
    )
    # 応答をJSON形式で取得し、テキストメッセージを取り出す(claude)
    response_body = json.loads(response['body'].read().decode('utf-8'))
    output_txt = response_body['completion']
    bot_message = output_txt.strip() 
    return bot_message

中小企業診断士でのパネルディスカッションでは生成AIの利活用が広がる中でどのような能力が求められるようになるかについてお話をさせていただいた。

シンポジウムの登壇を通じてAIのユースケースを多くの人に知ってもらい、これならできると思ってもらえることが重要だと感じた。

同じ中小企業診断士の遠田先生のブログにも紹介がありますので、イベントの様子を併せてご覧いただきたい。