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

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

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/6f2179b4-3b6e-c9d6-6a67-3619f55bd341.png)

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

https://github.com/hayao-k/notify-new-model-available-on-bedrock

### 元ネタ
https://twitter.com/minorun365/status/1764967086693183693

## 仕組み
### 全体像
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/04b93ea5-42ca-3c52-b582-5dd42a498959.png)

* [ListFoundationModels API](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_ListFoundationModels.html) を実行し、モデル 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 で取得できます。

https://docs.aws.amazon.com/bedrock/latest/APIReference/API_ListFoundationModels.html

```py
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 することで通知内容をカスタマイズできる機能があります。

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

https://docs.aws.amazon.com/chatbot/latest/adminguide/custom-notifs.html

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

```py
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 が設定済みであること

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

```shell-session
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
https://twitter.com/moritalous/status/1768870148201738477

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