はじめまして！

SIer企業でインフラエンジニアをしている naka と申します。
このたび、ありがたいことに 2025 Japan AWS Jr. Champions に選出いただき、同世代の若手エンジニアの熱い取り組みに刺激を受ける日々を送っています。
ということで、これをきっかけにこれからは、「Qiita見る専」から「投稿する側」へ挑戦していこうと思います！

普段の業務について

AWSやAzureを活用し、

IaC（Infrastructure as Code）
運用自動化

といった分野を中心に日々取り組んでいます。

これからは、気になるトピックや学びをアウトプットとして記事にまとめていく予定です。
少しでも参考になる記事があれば、ぜひ読んでいただけると嬉しいです！

では、記念すべき1本目の記事、始めます！

AWS × 生成AIで進化するAIOpsの世界

早速ですが皆さん、AIOps ってご存知ですか？

AIOpsとは？

AIOps（Artificial Intelligence for IT Operations）は、
AIや機械学習の技術を活用して、IT運用を自動化・効率化するアプローチのことです。
具体的には以下のような領域で活用されます。

大量のログやメトリクスを解析して、異常検知や根本原因分析（RCA）を行う
アラートの相関分析・優先度付けにより、インシデント対応を最適化する
過去のトラブル事例を元に、将来の障害予測や自動修復（Self-Healing）を実現する

これまで人が手作業で行ってきた調査・分析・一次切り分けを、AIに任せることで
MTTR（平均復旧時間）の短縮や運用コスト削減が可能になります。

AWSにおけるAIOpsの可能性

AWSでは、CloudWatchやX-Ray、DevOps Guruなどの運用系サービスに加え、
Amazon Bedrock や Amazon Q といった生成AIサービスが登場したことで、
「ログ解析 × 自然言語要約 × 自動提案」 といった次世代のAIOpsが現実的になっています。

今回やること

今回は、その第一歩として
「CloudWatch Logs × Amazon Bedrock（Claude）でインシデント要約Botを作る」
というPoC（Proof of Concept）に挑戦します。

具体的には以下の機能を持つ簡易Botをサーバーレス構成で構築します。

CloudWatch Logsからエラーイベントを取得
Claudeがログを解析・要約
Teamsに自動で通知

この記事を通して、生成AIをAWS運用に取り込むイメージをつかんでもらえたら嬉しいです。

アーキテクチャ図（イメージ）

今回構築する インシデント要約Bot は、以下のようなアーキテクチャを組んで動作させていきます。

早速作っていく

環境構築はCloudShellからやっていこうと思います。

まずは以下のコマンドでログの出力先となるロググループとログストリームを作成していきます。

qiita.rb

# 1. ロググループ作成（存在しなければ）
aws logs create-log-group --log-group-name "Naka_LOG_GROUP" 2>/dev/null

# 2. ログストリーム作成
aws logs create-log-stream --log-group-name "Naka_LOG_GROUP" --log-stream-name "Naka_LOG_STREAM" 2>/dev/null

ログストリームまで作成できました。

続いて、エラーイベントが発生した際に動いてもらうLambdaを作ります。
ここで作成するLambdaには、
①Cloudwatch Logsへログ出力するための権限
②Amazon Bedrockでモデルを呼び出す権限
を付与する必要があるので、まずは以下のコマンドで、①②を許可したロールを作成します。

qiita.rb

# 3. Lambda実行用ロールの作成
aws iam create-role --role-name NakaLambdaExecutionRole \
  --assume-role-policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Principal": { "Service": "lambda.amazonaws.com" },
      "Action": "sts:AssumeRole"
    }]
  }'

# 4. Lambda実行用ロールに「①」のポリシーを付与
aws iam attach-role-policy \
  --role-name NakaLambdaExecutionRole \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

# 5. Lambda実行用ロールに「②」のポリシーを付与
aws iam attach-role-policy \
  --role-name NakaLambdaExecutionRole \
  --policy-arn arn:aws:iam::aws:policy/AmazonBedrockFullAccess

作成できましたね！

続いて以下のコマンドで、今回メインとなるLambdaコードを作成します。
利用するモデルは　Claude3.5 Sonnet　にしています。
※事前に対象のモデルが有効化されていることを確認してください！

Lambdaコードの作成（長いのでたたんでいます）

# 6. Lambdaコードの作成
cat << 'EOF' > lambda_function.py
import json
import boto3
import base64
import gzip

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
model_id = "anthropic.claude-3.5-sonnet-20240620"

def summarize_with_claude(logs):
    messages = "\n".join(f"- {line}" for line in logs)
    prompt = f"""以下はシステムログの抜粋です。内容を簡潔に要約し、原因や傾向があれば記述してください。

{messages}

要約:"""
    body = {
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "max_tokens": 300,
        "temperature": 0.5,
        "anthropic_version": "bedrock-2023-05-31"
    }
    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json"
    )
    result = json.loads(response["body"].read())
    return result["content"][0]["text"]

def lambda_handler(event, context):
    cw_data = event['awslogs']['data']
    compressed_payload = base64.b64decode(cw_data)
    uncompressed_payload = gzip.decompress(compressed_payload)
    log_data = json.loads(uncompressed_payload)

    messages = [entry['message'] for entry in log_data['logEvents']]
    recent_messages = messages[-10:]

    try:
        summary = summarize_with_claude(recent_messages)
        print("=== 要約結果 ===")
        print(summary)
    except Exception as e:
        print("要約中にエラー:", e)

    return {
        'statusCode': 200,
        'body': json.dumps('Processed logs.')
    }
EOF

Lambdaコードとして設定するために、このスクリプトはZIP圧縮する必要があります。

qiita.rb

# 7. デプロイZIPの作成
zip NakaLogSummaryBot.zip lambda_function.py

これでLambda関数を作成する準備が整ったので、以下のコマンドを実行してLambda関数を作成していきます！

qiita.rb

# 8. Lambda関数の作成
aws lambda create-function \
  --function-name NakaLogSummaryBot \
  --runtime python3.12 \
  --role arn:aws:iam::$(aws sts get-caller-identity --query Account --output text):role/NakaLambdaExecutionRole \
  --handler lambda_function.lambda_handler \
  --zip-file fileb://NakaLogSummaryBot.zip \
  --region us-east-1

無事作成できました！

続いて、エラーイベントが発生した際に、今回作成したLambda関数が動くようにサブスクリプションフィルターを作成していきます。
対象のロググループに「error」というログが出力されたことをトリガーに、今回作成したLambda関数が実行されるようにします。

qiita.rb

# 9. Lambda に invoke権限 を付与
aws lambda add-permission \
  --function-name NakaLogSummaryBot \
  --statement-id AllowCWLogsInvoke \
  --action "lambda:InvokeFunction" \
  --principal logs.amazonaws.com \
  --source-arn "arn:aws:logs:us-east-1:<アカウントID>:log-group:Naka_LOG_GROUP:*" \
  --region us-east-1

  # 10. サブスクリプションフィルターを作成
  aws logs put-subscription-filter \
  --log-group-name "Naka_LOG_GROUP" \
  --filter-name "NakaErrorFilter" \
  --filter-pattern "error" \
  --destination-arn "arn:aws:lambda:us-east-1:<アカウントID>:function:NakaLogSummaryBot" \
  --region us-east-1

これでサブスクリプションフィルターも作成できました！

ここまでできたら、一旦想定通りに動くかテストしていきます。
最初に作成したロググループに、テスト用のログを出力させて、それをトリガーに今回作成したLambdaを動かしてみます。

qiita.rb

aws logs put-log-events \
  --log-group-name "Naka_LOG_GROUP" \
  --log-stream-name "Naka_LOG_STREAM" \
  --log-events timestamp=$(date +%s%3N),message="error: failed to connect to database" \
  --sequence-token $(aws logs describe-log-streams \
    --log-group-name "Naka_LOG_GROUP" \
    --log-stream-name-prefix "Naka_LOG_STREAM" \
    --query "logStreams[0].uploadSequenceToken" \
    --output text) \
  --region us-east-1

これで、ロググループに以下のようにログが出力されました。

このログに含まれる「error」がトリガーとなり、Lambdaが動いてくれているはずです。
では、Lamndaの実行ログをみてみましょう！
※このときタイムアウトエラーが出たので、Lambdaの処理時間を5分に延ばしてから再度実行しました。

しっかり動いて、いい感じに要約もしてくれてそうです！

あとはこの内容をTeamsに飛ばしたいので、Lambdaのコードを一部修正します。
※Teamsへの通知は別途作成しておいたWebhook URLが必要になります。このURLの発行については以下の記事を参考にさせていただきました。
ありがとうございます！
https://qiita.com/n0bisuke/items/3bb150967fb84a85eebe

Teams通知用Lambdaコード（長いのでたたんでいます）

import json
import boto3
import base64
import gzip
import urllib3
import os

# Claude 3.5 Sonnetのモデル設定
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
model_id = "anthropic.claude-3.5-sonnet-20240620"

# TeamsのWebhook URL（Lambdaの環境変数に設定）
TEAMS_WEBHOOK_URL = os.environ.get("TEAMS_WEBHOOK_URL")

def summarize_with_claude(logs):
    messages = "\n".join(f"- {line}" for line in logs)
    prompt = f"""以下はCloudWatch Logsに記録されたエラーログです。
この内容から、何が起こった可能性があるかをわかりやすく要約してください。

ログ:
{messages}

要約:"""

    body = {
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "max_tokens": 300,
        "temperature": 0.5,
        "anthropic_version": "bedrock-2023-05-31"
    }

    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json"
    )

    result = json.loads(response["body"].read())
    return result["content"][0]["text"]

def send_teams_adaptive_card(summary):
    if not TEAMS_WEBHOOK_URL:
        print("❌ TEAMS_WEBHOOK_URL が設定されていません")
        return

    http = urllib3.PoolManager()

    adaptive_card_payload = {
        "attachments": [
            {
                "contentType": "application/vnd.microsoft.card.adaptive",
                "content": {
                    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                    "type": "AdaptiveCard",
                    "version": "1.4",
                    "body": [
                        {
                            "type": "TextBlock",
                            "size": "Medium",
                            "weight": "Bolder",
                            "text": "🚨 Claude 要約結果"
                        },
                        {
                            "type": "TextBlock",
                            "text": summary,
                            "wrap": True
                        }
                    ]
                }
            }
        ]
    }

    response = http.request(
        "POST",
        TEAMS_WEBHOOK_URL,
        body=json.dumps(adaptive_card_payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )

    print(f"Teams通知ステータスコード: {response.status}")

def lambda_handler(event, context):
    try:
        # CloudWatch Logs イベント展開
        cw_data = event['awslogs']['data']
        compressed_payload = base64.b64decode(cw_data)
        uncompressed_payload = gzip.decompress(compressed_payload)
        log_data = json.loads(uncompressed_payload)

        messages = [entry['message'] for entry in log_data['logEvents']]
        recent_logs = messages[-10:]

        # Claudeで要約
        summary = summarize_with_claude(recent_logs)
        print("=== Claude要約 ===")
        print(summary)

        # Teamsに通知
        send_teams_adaptive_card(summary)

    except Exception as e:
        print("❌ Lambda実行エラー:", str(e))

    return {
        'statusCode': 200,
        'body': json.dumps('Log processed.')
    }

コードの修正が完了したので、再度ロググループにテスト用のログを出力させてLambdaを動かしてみます！
すると、以下のようにしっかりTeamsにも通知が飛んできていることが確認できました！

まとめ：Claudeの自然言語要約、運用に使える！

いかがでしたでしょうか？
実運用で活用するには、まだいくつかの改善ポイントはありますが、
今回のPoCを通じて、AWS運用にAIOpsを取り込むイメージを少しでも掴んでいただけたのではないかと思います。

私自身、これまで生成AI系のサービスに触れる機会はあまり多くありませんでしたが、
今回の検証を通じて、AWSの生成AIサービスが実運用レベルでも十分に活用できる手応えを感じることができました。

今後も「AWS運用 × 生成AI」の領域をさらに深掘りし、
業務改善や自動化の可能性を広げていければと考えています。

最後までお読みいただき、本当にありがとうございました！
また次回の記事でお会いしましょう！🌟