生成 AI アプリケーションをデプロイすると、コストはもちろんインシデントにつながる入出力の監視も重要になります。次のスライドでは 2024 年に生成 AI を使用したチャットボットやエージェントで発生した代表的なインシデントを挙げています。回答の誤りはもちろん、ユーザー側の作為的な入力が運用停止につながりうることがわかります。
AI Agent を始める前のチェックリスト:実例から学ぶ期待効果の見積りと備えるべきリスク より引用
本記事では、AWS のマネージドな運用監視サービスである Amazon CloudWatch を使用してリスクの検知・トレースを行う方法を示します。生成 AI アプリケーションの監視といえば Langfuse といったオープンソースツールを思い浮かべる方もいるかもしれません。これらのツールは洗練された機能を提供しますが、その分ホスティングするコストや学習コストもかかります。そのため、今回は AWS を利用していれば追加の手間・コストがほぼなしに使える CloudWatch を使って記録していきます。安全性の検知には、執筆時点でまだ日本語に対応してませんが Amazon Bedrock Guardrails を使用します。
はじめに安全性をモニタリングするイメージを紹介し、そのあとの具体的な実装を紹介します。
安全性のモニタリングシナリオ
あなたが生成 AI アプリケーションをモニタリングしている担当者とします。昼頃ダッシュボードを見てみると、Guardrails でブロックされたインプットが閾値 (この場合時間当たり 7) を超えていました。
何事か、ということでガードレールにかかった場合の HTTP Status (400: Bad Request) のトレースを参照してみます。
トレースを見ると、Error のスパンで良からぬ入力がされていました (黒塗りにしています。余談ですが、ガードレール系のテストではフィルタにかかるような有害文言を考えるのが結構ストレスです)。
セッション ID や IP などを見て、リクエストを遮断するなどの措置を取りる必要がありそうです (本記事では扱いませんが、こういう場合にどう対応するかを決めておく必要があります)。
こうした一連の対応を行うための実装方法を解説します。
安全性のモニタリングを実現する実装
今回のアプリケーションは全体として次の構造になっています。
- React アプリケーションから Amazon CloudFront 経由で AWS Lambda (Function URL) に API アクセス
- AWS Lambda で、会話履歴を Amazon DynamoDB、返答を Amazon Bedrock で推論。入力・出力は Amazon Bedrock Guardrails で確認
- リクエストは CloudWatch の Application Signals でトレースし、X-Ray で参照
ポイントは Application Signals の自動計装の仕組みで、基本的にトレースをオンにするだけで AWS のサービスをまたいだリクエスト処理の過程を自動で記録してくれます。アプリケーションのコードに手を入れることなくトレースを一定行ってくれるのはとても手軽です。
もちろん、自動で記録されないトレースを行いたい場合もあります。Application Signals はオープンソースの Observability フレームワークである OpenTelemetry と互換性があり、OpenTelemetry で記録した内容も統合的に CloudWatch で参照できます。
以下セクションでは、記録をするための具体的な実装について解説していきます。実装は下記で公開していますが、シンプルな実装なのでそのまま使用するというより実装の参考にしていただければ幸いです。
フロントエンド
今回、フロントエンドの実装は生成 AI に任せて React x Materialize で作りました。構築には Create React App を使用していますが、こちらは 2025/2/14 に非推奨の宣言が行われています。そのため VITE などを利用すべきですが、今回フロントエンドは実験を行うためのインターフェースなので一旦移行は棚に上げています (ただ、依存パッケージの vulneravility は 0 にしています)。
実際の画面はこんな感じです。ガードレールにかかった場合、スコアがわかるようにしています (Guardrails は具体的なスコアでなく Confidence を NONE, LOW, MEDIUM, HIGH で返すのでそれを数値にしています)。
フロントエンドから Function URL により公開した AWS Lambda を叩いています。この際、アプリケーションからのリクエストのみ通すよう CloudFront を通し Lambda@Edge で AWS Lambda を呼び出すための AWS_IAM 認証を行っています。注意点として、PUT/POST のリクエストについては x-amz-content-sha256 ヘッダーに SHA256 で計算したリクエストボディのハッシュ値を設定する必要があります。CloudFront の Origin Access Control でサポートされているリクエストに対する AWS Signature Version 4 (SigV4) 署名 は執筆時点で GET のみで、PUT/POST は自前での実装が必要です。実装には下記記事を参照させて頂きました。
下記は Lambda@Edge の実装です。上記の記事で紹介されている実装とほぼ同じですが、SHA256 の署名については createHash 一発で出来ます。
import { CloudFrontRequestEvent, CloudFrontRequestHandler } from "aws-lambda";
import { createHash } from "crypto";
/**
* Calculate SHA256 hash of a payload string
* @param payload - The payload to hash
* @returns SHA256 hash as a hexadecimal string
*/
const hashPayload = async (payload: string): Promise<string> => {
return createHash('sha256').update(payload).digest('hex');
};
/**
* Lambda@Edge handler for CloudFront viewer requests
* Adds x-amz-content-sha256 header with SHA256 hash of the request body
* This is required for Lambda function URLs with AWS_IAM auth when accessed through CloudFront
*/
export const handler: CloudFrontRequestHandler = async (
event: CloudFrontRequestEvent
) => {
const request = event.Records[0].cf.request;
console.log("Original request:", JSON.stringify(request));
// If there's no body, return the request as is
if (!request.body?.data) {
console.log("No request body found, skipping hash calculation");
return request;
}
try {
// Decode the base64-encoded body
const body = request.body.data;
const decodedBody = Buffer.from(body, "base64").toString("utf-8");
// Calculate SHA256 hash of the body
const contentHash = await hashPayload(decodedBody);
console.log(`Calculated content hash: ${contentHash}`);
// Add the x-amz-content-sha256 header
request.headers["x-amz-content-sha256"] = [
{ key: "x-amz-content-sha256", value: contentHash }
];
console.log("Modified request:", JSON.stringify(request));
return request;
} catch (error) {
console.error("Error processing request:", error);
// In case of error, return the original request
// This allows the request to proceed, though it might fail at the Lambda function URL
return request;
}
};
バックエンド
バックエンドは AWS Lambda です。今回、1) Application Signals で自動トレースをするために tracing: Tracing.ACTIVE を設定し、2) OpenTelemetry で追加の記録をするために AWS Distro for OpenTelemetry (ADOT) を使用しています。こちらは OpenTelemetry で記録するためのエージェントやツールセットを一括で提供するもので、AWS Lambda の場合 Lambda Layer の形で提供されています。AWS_LAMBDA_EXEC_WRAPPER: '/opt/otel-handler'を指定することで OpenTelemetry 側のハンドラで自動計装がされます (※ Tracing.ACTIVE にした場合はデフォルト X-Ray へ記録されます。ADOT の Collector はデフォルト X-Ray へ送信するようになっているので経路は異なりますが見る場所は X-Ray になります)。
// Create a Lambda Layer with AWS Distro for OpenTelemetry Lambda
// https://aws-otel.github.io/docs/getting-started/lambda/lambda-js
const adotLayer = LayerVersion.fromLayerVersionArn(
this,
'ADOTJSLayer',
`arn:aws:lambda:${Stack.of(this).region}:901920570463:layer:aws-otel-nodejs-amd64-ver-1-30-1:2`
);
// Create Lambda function
const chatFunction = new Function(this, 'ChatFunction', {
runtime: Runtime.NODEJS_22_X,
code: new AssetCode(backendPath),
handler: 'index.handler',
timeout: Duration.minutes(5), // Extended for streaming responses
tracing: Tracing.ACTIVE, // Enable tracing
environment: {
TABLE_NAME: chatTable.tableName,
GUARDRAIL_ID: guardrails.guardrailId,
GUARDRAIL_VERSION: guardrailVersion,
AWS_LAMBDA_EXEC_WRAPPER: '/opt/otel-handler'
}, // Thanks to AWS_LAMBDA_EXEC_WRAPPER, Lambda execution is traced by otel-hander.
layers: [adotLayer]
});
リスク検知を担う Guardrails は generative-ai-cdk-constructs を使用すると非常に簡単に作ることが出来ます。
// Create Bedrock Guardrails
const guardrails = new Guardrail(this, 'ChatGuardrails', {
name: 'llm-observability-guardrails',
description: 'Guardrails for LLM observability demo',
});
// Add content filters
guardrails.addContentFilter({
type: ContentFilterType.HATE,
inputStrength: ContentFilterStrength.LOW,
outputStrength: ContentFilterStrength.LOW,
});
AWS Lambda での呼び出しは以下のような実装になります。返却の型が結構複雑なので AWS ドキュメントの MCP を使い実装しました (Claude が)。最終的に確認しているんですが、ちゃんと型が取れているのは驚きました。
const command = new ApplyGuardrailCommand({
guardrailIdentifier: GUARDRAIL_ID,
guardrailVersion: GUARDRAIL_VERSION,
content: [{
text: {
text: content
}
}],
source: source,
outputScope: 'FULL' // Get full output for enhanced debugging
});
const response = await bedrockClient.send(command);
カスタムメトリクスの記録
独自に記録したいスパンがある場合、次のように実装しています。自動計装をしていると自動と手動で同じ範囲に 2 つ似たスパンができるため、細かくとりたい場合は自動を OFF にしてもいいかもしれません (Application Signals の思想と逆ですが。。。)。
import * as api from '@opentelemetry/api';
...
const tracer = api.trace.getTracer(TRACE_NAME);
const currentSpan = api.trace.getActiveSpan() || tracer.startSpan(TRACE_NAME);
...
// Apply guardrails to user message
const userGuardrailsResult = await tracer.startActiveSpan('Guardrails-INPUT', async (span : api.Span) => {
span.setAttribute('guardrails.id', GUARDRAIL_ID);
span.setAttribute('guardrails.version', GUARDRAIL_VERSION);
const result = await applyGuardrails(message, 'INPUT');
const totalScore = Object.values(result.contentFilterResults || {}).reduce((acc, filter) => acc + (filter.score || 0), 0);
if (totalScore > 0) {
span.setStatus({ code: api.SpanStatusCode.ERROR, message: 'User message filtered' });
span.setAttribute('guardrails.input', message);
}
span.end();
return result;
});
入出力やトークン数など、状況の把握や対応をするために必要な属性やエラー/正常のステータスを設定できます。
ダッシュボードの作成
ダッシュボードの作成は CDK でなく手動で行いました。Bedrock は Bedrock と Guardrails が分かれているので注意です (本記事執筆中に Agent のメトリクスも取れるようになったとアナウンスがあったので、 Agent の起動回数なども取れると思います)。
Guardrails のブロック数は "InvocationsIntervened" で取ることが出来ます。"Period" でモニタリングしたい期間を設定し、閾値を決めることで設定期間内の数を集計できます。
Trace は X-Ray のページから "Add Dashboard" してしまうのが楽です。
おわりに
今回のサンプル実装は 8 割ぐらい生成 AI で作成しました。良い点として、不慣れな言語でもかなり迅速に書くことが出来ること、MCP を使えばかなり正確に型も当てられるようになる点がありました。課題と感じたのは Node のバージョン、脆弱性あるパッケージの禁止、ディレクトリ構成といった前提情報の徹底と、当然ではありますが情報がそもそもない領域では無力というところでした。生成させてみて気づくこともあるので、最初からがっちり rule を書くより Cline であれば Plan モードで一定詰めてから (実装も例示として書いてもらう) md 形式にしてコンテキストとして渡すのがワークする印象です。
本記事が皆さんの安全な生成 AI アプリケーションライフのお役に立てば幸いです！