AmazonBedrockちょろっと試してみました。
構成はこんな感じで、定期実行したLambdaにCloudwatchlogsのログをBedrockに要約させてSNSでメール通知する仕組み
従来の英語だらけのエラーメッセージを自然言語のメールとすることで運用効率を向上できないかな!みたいな
EventBridge、SNSは、CloudwatchLogsは割愛します。
IAMはLambdaに以下のポリシーをアタッチしています。※業務利用では絞ってください。
・AmazonBedrockFullAccess1
・AmazonSNSFullAccess1
・AWSLambdaBasicExecutionRole
・CloudWatchLogsFullAccess
ハイライト行は適宜書き換えてください。
コード自体もAmazon Qに殆ど書いてもらいました
※Bedrockの応答に時間がかかる場合があるので、Lambdaの実行時間を延ばすのもポイントです。
8行目のmodel_id = “amazon.titan-text-express-v1”ですが、利用に際し申請が必要なものとそうでないものがあるので
申請不要のモデルとして上記の記載としています。
import json
import boto3
from datetime import datetime, timedelta
bedrock = boto3.client("bedrock-runtime", region_name="ap-northeast-1")
logs = boto3.client("logs")
sns = boto3.client("sns")
model_id = "amazon.titan-text-express-v1"
def summarize_with_claude(log_messages):
messages = "\n".join(f"- {line}" for line in log_messages)
prompt = f"""以下はログの抜粋です。内容を簡潔に要約し原因や傾向があれば記述してください。
{messages}
要約:"""
body = {
"inputText": prompt,
"textGenerationConfig": {
"maxTokenCount": 300,
"temperature": 0.5
}
}
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["results"][0]["outputText"]
def lambda_handler(event, context):
log_group_name = "/aws/lambda/XXXXXXXXXXXXXXX"
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)
try:
response = logs.filter_log_events(
logGroupName=log_group_name,
startTime=int(start_time.timestamp() * 1000),
endTime=int(end_time.timestamp() * 1000)
)
log_messages = [event['message'] for event in response['events']]
if log_messages:
summary = summarize_with_claude(log_messages[-20:])
sns.publish(
TopicArn='arn:aws:sns:ap-northeast-1:111111111111:XXXXXXXX',
Subject='定期ログ解析結果',
Message=f"=== 定期ログ解析結果 ===\n{summary}"
)
print("解析完了:", summary)
else:
print("対象期間にログなし")
except Exception as e:
print("エラー:", e)
return {
'statusCode': 200,
'body': json.dumps('Log analysis completed.')
}