DEV Community

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

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

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

講演の様子

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

Image description

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

Image description

ディレクトリ構成

├── 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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

パネルディスカッションの様子

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

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

Heroku

Simplify your DevOps and maximize your time.

Since 2007, Heroku has been the go-to platform for developers as it monitors uptime, performance, and infrastructure concerns, allowing you to focus on writing code.

Learn More

Top comments (0)

Amazon S3 to AWS StepFunctions pattern cover image

Amazon S3 to AWS StepFunctions pattern

Explore methods to integrate Amazon S3 and AWS StepFunctions for highly scalable workflows: straightforward via AWS Lambda, applying the storage-first pattern with Amazon SQS and event-driven by linking AWS CloudTrail and Amazon EventBridge.

Read full post