2025.02.28

AWS Healthからの通知をBedrockに解釈させて運用負荷を下げる

keichan

Contents

はじめに
仕組み
コード
実際に通知されるメッセージの例
注意点

はじめに

30歳という数字を目前にして、色々と焦りを感じているkeichanです。
今回は、AWS Healthからの通知をBedrockで解釈し、運用エンジニアの負担を軽減する仕組みを構築したので、その紹介をします。

仕組み

仕組み自体はシンプルで、
AWS Health Event -> EventBridge -> Lambda -> Slackという流れで処理をしています。
Lambda関数の中でBedrockを呼び出し、Healthイベントをプロンプトの内容に応じて解釈させることで、
発生したイベントに対して、エンジニアが対処する必要があるかどうかの判断負荷を低下させています。

もっと作り込めば、各イベントごとに対応手順書をSlackメッセージに含めることも可能です。

コード

import json
import boto3
import os
import requests

bedrock = boto3.client('bedrock-runtime')
SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']

def lambda_handler(event, context):
    detail = event['detail']
    event_arn = detail['eventArn']
    service = detail['service']
    event_type = detail['eventTypeCode']
    status = detail['statusCode']
    description = detail['eventDescription'][0]['latestDescription']

    message_content = f"""
    AWS Health Event が発生しました。以下の情報を基に、人間が理解しやすい形で状況を説明してください。
    また、長い文章になるときは適切に改行を入れてください。出力はSlackに通知するので、Slack上で読みやすい文章にしてください。：

    イベントARN: {event_arn}
    サービス: {service}
    イベントタイプ: {event_type}
    ステータス: {status}
    説明: {description}

    説明は以下の点を含めてください：
    1. 発生している問題点
    2. イベントの重要性
    3. 運用担当者による手作業が必要かどうか
    4. 推奨される対応策
    """

    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-sonnet-20240229-v1:0',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 1000,
            "temperature": 0.7,
            "top_p": 0.999,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": message_content
                        }
                    ]
                }
            ]
        })
    )

    response_body = json.loads(response['body'].read())
    interpreted_message = response_body['content'][0]['text']

    # Slackメッセージを作成（最大3900文字に制限）
    slack_message = f"AWS Health Event Alert\n\nEvent ARN: {event_arn}\n\n{interpreted_message}"
    if len(slack_message) > 3900:
        slack_message = slack_message[:3897] + "..."

    # Slackに送信
    slack_response = requests.post(SLACK_WEBHOOK_URL, json={"text": slack_message})
    print(f"Slack notification sent. Status code: {slack_response.status_code}")

    return {
        'statusCode': 200,
        'body': json.dumps('Event processed and sent to Slack successfully'),
        'interpretedMessage': interpreted_message
    }