LoginSignup
7
7
@hayao_kin株式会社セゾンテクノロジー

Amazon Bedrock に新しいモデルが追加されたら AWS Chatbot で通知する

Last updated at Posted at 2024-03-18

はじめに

Amazon Bedrock 上で利用可能な基盤モデルは増え続けています。2024 年 3 月だけで Mistral AI、Claude 3 Sonnet、Claude 3 Haiku が利用可能になりました。

東京リージョンで Claude 3 シリーズが使えるようになるのも待ち遠しいですね！

いち早く検証してナレッジを世に発信するために、Bedrock にモデルが追加されたら Slack に通知する仕組みを作ってみます。

通知例

こんな感じ
画像中のモデル名は動作確認のためのテストデータ (ネタ) です。

image.png

ソースコードは以下の GitHub リポジトリに置いてあります。

元ネタ

仕組み

全体像

image.png

  • ListFoundationModels API を実行し、モデル ID のリストを取得
  • 前回実行時のリストを DynamoDB テーブルから取得し、差分を比較
  • 新しいモデルが追加されていれば AWS Chatbot 経由で Slack 通知し、DynamoDB テーブルに最新のリストを保存
  • 上記処理を行う Lambda 関数を EventBridge Scheduler で定期的に実行する

冒頭の GitHub リポジトリには Lambda 関数、EventBridge schedule、DynamoDB テーブルおよび関連 IAM リソースをデプロイするための AWS SAM テンプレートが含まれています。

通知用の SNS Topic および AWS Chatbot は作成、設定済みであることを前提とし、トピック名を CloudFormation スタックのパラメーターとして指定します。

モデル ID の取得

Amazon Bedrock で利用可能な基盤モデルの詳細は ListFoundationModels API で取得できます。

import boto3

bedrock = boto3.client('bedrock')
response = bedrock.list_foundation_models()
model_ids = [item['modelId'] for item in response["modelSummaries"]]

AWS Chatbot のカスタム通知機能

AWS Chatbot には事前定義されているイベントスキーマに沿ってメッセージを Publish することで通知内容をカスタマイズできる機能があります。

イベントスキーマーやサンプルは以下のドキュメントに記載されています。

今回のケースでは Lambda 関数内でメッセージを作成し、AWS Chatbot 用の SNS Topic に送信しています。

import boto3

sns = boto3.client('sns')
message = {
    "version": "1.0",
    "source": "custom",
    "content": {
        "textType": "client-markdown",
        "title": f":tada: Amazon Bedrock に新しいモデルが追加されました!! Region: {region}",
        "description": '\n'.join(new_models),
        "nextSteps": [
            "宇宙最速を目指そう:rocket: ",
            f'''Go to <https://{region}.console.aws.amazon.com/bedrock/home?region={
                    region}#/modelaccess|*Model Access*>'''
        ]
    }
}
sns.publish(
    TopicArn=f"arn:aws:sns:{region}:{account_id}:{topic_name}"
    Message=json.dumps(message)
)

アプリケーションのデプロイ

以下の前提条件を確認してください。

  • AWS SAM CLI がインストール済みであること
  • AWS Chatbot 用の SNS Topic が作成、設定済みであること
  • AWS Chatbot が設定済みであること

以下のコマンドを実行します。

git clone https://github.com/hayao-k/notify-new-model-available-on-bedrock.git
cd notify-new-model-available-on-bedrock
sam deploy --guided

次のパラメーターを使用できます。

  • Parameter: pTopicName
    デプロイ先リージョンに存在する AWS Chatbot 通知用の SNS Topic 名を指定します。同一アカウント内である必要があります。 デフォルトは AWSChatbot-Topic です。
  • Parameter: pRateMinute
    EventBridge Scheduler の Rate 式に設定する実行間隔 (分) を指定します。デフォルトは 5 です。

ソースコード

Lambda 関数

lambda_function.py
import json
from logging import getLogger
import boto3
from botocore.exceptions import ClientError
from typing import List, Dict, Any
import os

account_id = boto3.client("sts").get_caller_identity()["Account"]
bedrock = boto3.client('bedrock')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE_NAME'])
sns = boto3.client('sns')
topic_name = os.environ['TOPIC_NAME']
region = os.environ["AWS_REGION"]

logger = getLogger()

def fetch_model_ids() -> List[str]:
    try:
        response = bedrock.list_foundation_models()
        return [item['modelId'] for item in response['modelSummaries']]
    except ClientError as e:
        log_and_raise_error(e)
        
def fetch_previous_model_ids() -> List[str]:
    try:
        response = table.get_item(Key={'id': 'models_list'})
        if 'Item' in response:
            return response['Item']['models']
        else:
            return [] 
    except ClientError as e:
        log_and_raise_error(e)

def update_model_ids_in_dynamodb(model_ids: List[str]) -> None:
    try:
        table.put_item(Item={'id': 'models_list', 'models': model_ids})
    except ClientError as e:
        log_and_raise_error(e)

def send_notification(new_models: List[str]) -> None:
    message = {
        "version": "1.0",
        "source": "custom",
        "content": {
            "textType": "client-markdown",
            "title": f":tada: Amazon Bedrock に新しいモデルが追加されました!! Region: {region}",
            "description": '\n'.join(new_models),
            "nextSteps": [
                "宇宙最速を目指そう:rocket: ",
                f'''Go to <https://{region}.console.aws.amazon.com/bedrock/home?region={
                    region}#/modelaccess|*Model Access*>'''
            ]
        }
    }
    try:
        sns.publish(
            TopicArn=f"arn:aws:sns:{region}:{account_id}:{topic_name}"
            Message=json.dumps(message)
        )
    except ClientError as e:
        log_and_raise_error(e)

def log_and_raise_error(e):
    error_message = e.response['Error']['Message']
    logger.error(error_message)
    raise

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    logger.debug(event)
    model_ids = fetch_model_ids()
    previous_model_ids = fetch_previous_model_ids()
    new_models = list(set(model_ids) - set(previous_model_ids))
    
    if new_models:
        logger.info("New models found: %s", new_models)
        update_model_ids_in_dynamodb(model_ids)
        send_notification(new_models)

        return {
            'statusCode': 200,
            'body': json.dumps('New models found!')
        }

    return {
        'statusCode': 200,
        'body': json.dumps('No additional models.')
    }

AWS SAM テンプレート

template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  AWS SAM template for notify-new-model-available-on-bedrock

Parameters:
  pTopicName:
    Type: String
    Description: The Name of the SNS topic for notifications.
    Default: "AWSChatbot-Topic"
  pRateMinute:
    Type: String
    Description: Rate minutes to be set in EventBridge scheduler.
    Default: "5"

Resources:
  ModelsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      BillingMode: PROVISIONED
      ProvisionedThroughput:
        ReadCapacityUnits: 1
        WriteCapacityUnits: 1

  NotifyNewModelsFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.12
      Architectures:
        - arm64
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
      MemorySize: 128
      Timeout: 10
      Environment:
        Variables:
          DYNAMODB_TABLE_NAME: !Ref ModelsTable
          TOPIC_NAME: !Ref pTopicName
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref ModelsTable
        - SNSPublishMessagePolicy:
            TopicName: !Ref pTopicName
        - Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - bedrock:ListFoundationModels
              Resource: '*'
      Events:
        ScheduleEvent:
          Type: ScheduleV2
          Properties:
            ScheduleExpression: !Sub 'rate(${pRateMinute} minute)'
            State: ENABLED

  LogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${NotifyNewModelsFunction}'
      RetentionInDays: 3

Outputs:
  ModelsTableArn:
    Description: "ARN of the DynamoDB table"
    Value: !GetAtt ModelsTable.Arn

  NotifyNewModelsFunctionArn:
    Description: "ARN of the Lambda function"
    Value: !GetAtt NotifyNewModelsFunction.Arn

Next Step

以上です。
参考になれば幸いです。

7
7
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
7
7