マルチモーダルLLMの安定性をDatadog Synthetic Testで担保する

参考になった

レビュー投稿日の情報になります

株式会社ワンキャリア / MikiPWata

メンバー / フロントエンドエンジニア

最終更新日2025/08/26投稿日2025/08/27

事業形態

B to B

B to C

事業形態	B to B B to C

アーキテクチャ

導入の背景・解決したかった問題

導入背景

ツール導入前の課題

システム障害発生時にインフラのメトリクスを確認することはできていたものの、根本的な問題の特定に時間がかかっていました。
特にパフォーマンスに起因する障害では、どのAPIがボトルネックになっているのか原因特定するのに数時間を要することもあり（3時間もかかることも）、対応の遅れが課題でした。また、SLO（Service Level Objective）の監視や対応が文化として根付いておらず、定義したSLOを可視化できない、あるいは監視業務に手作業が多く、運用コストが高いという問題も抱えていました。 E2Eテストツールは導入済みでしたが、コストの高さも悩みの種でした。

どのような状態を目指していたか

障害時の迅速な復旧体制
- 数時間かかっていた原因特定〜復旧対応までの時間を大幅に短縮する。
SLOの自動監視と運用定着
- SLOの状態をダッシュボードで可視化し、自動でアラートを受け取り、優先順位に基づいた対応を可能にする。
低コストかつ継続的なテスト運用
- 従来よりもコスト効率の良い方法で、自動テストを日常的に実行できる体制を目指す。

導入の成果

改善したかった課題はどれくらい解決されたか

SLO監視・対応における仕組みにはまだ改善余地もあるが、Datadogの導入により以下の課題は大きく改善されました。

システム障害時の復旧時間の短縮
自動テストツールのコスト削減

どのような成果が得られたか

APMの活用により、パフォーマンスのボトルネックの特定が容易に。障害発生から復旧までの時間が3分の1に短縮され、現在ではおおよそ1時間以内に復旧できるようになりました。また、SLOをDatadog上で定義し、ダッシュボードおよびSlack通知による監視体制を構築しました。稼働率・エラー率の見える化と、違反対応の文化が社内に根付きつつあります。

導入に向けた社内への説明

上長・チームへの説明

Datadogを選定した理由のひとつに「ユーザー単位の課金ではなく、リソース使用量に応じた課金体系」がありました。特定の機能に絞って利用すれば導入コストを抑えられることを、上長・チームに丁寧に説明しました。

また、APM（Application Performance Monitoring）やSynthetic Testなど、当時の課題に直結する機能にフォーカスし、解決策として具体的に提示しました。

活用方法

※応用的な活用方法のご紹介です

直近データチームでは、 GoogleのGemini を活用した開発を進めているのですが、時折 429 RESOURCE_EXHAUSTED エラーなど、外部API由来の不安定なレスポンスに悩まされていました。

大規模言語モデル（LLM）である Gemini は非常に強力なツールですが、外部サービスである以上、その応答速度や可用性には波があるのが実情です。特に開発中は、APIからのレスポンスが遅延したり、予期せぬエラーが返ってきたりするケースが散見されました。こういったエラーが本番環境で発生した場合、ユーザー体験の低下に直結してしまいます。そのため、Geminiの稼働状況をリアルタイムで把握し、問題が発生した際に迅速に検知・対応できる仕組みを構築することが急務であると考えました。

そこでサービスの安定運用に向けて、こういったエラーを早期に検知・対応できる体制が不可欠です。具体的には、以下のような2軸での監視体制を導入しました。

ヘルスチェックエンドポイントの実装:
Gemini の API を定期的に呼び出し、正常に応答するかどうかを確認する専用のエンドポイントを自前で用意。
Datadog Synthetic Test による外形監視:
作成したヘルスチェックエンドポイントに対して、Datadog から定期的にリクエストを送信し、レスポンスタイムやステータスコードを監視。これにより、外部から見たサービスの稼働状況を確認することができます。

この体制を選択した理由は、アプリケーション内部からの能動的なチェック（ヘルスチェックエンドポイント）と、外部からの受動的な監視（Datadog）を組み合わせることで、より多角的に Gemini の安定性を捉えられると考えたためです。

以下は全体的なアーキテクチャの概要になります。

sequenceDiagram
    participant Datadog
    participant HealthCheckServer
    participant Gemini
    participant VertexAI
    participant Slack

    Datadog->>HealthCheckServer: ヘルスチェックリクエスト送信
    HealthCheckServer->>Gemini: テキストモデルまたはマルチモーダルモデル呼び出し
    alt テキストデータ
        Gemini-->>HealthCheckServer: テキストレスポンス
    else オーディオデータ
        HealthCheckServer->>VertexAI: テキストを音声に変換
        VertexAI-->>HealthCheckServer: 音声データ
        HealthCheckServer->>Gemini: マルチモーダル処理
        Gemini-->>HealthCheckServer: マルチモーダルレスポンス
    end
    HealthCheckServer->>Datadog: JSONレスポンス(statusがsuccess)
    alt エラーレスポンス
        Datadog->>Slack: アラート送信
    else 正常レスポンス
        Datadog-->>HealthCheckServer: モニタリング継続
    end

具体的なアプローチ

アプローチ①：ヘルスチェックエンドポイントの実装

ヘルスチェックエンドポイントは、特定の URL にアクセスがあった際に、Gemini の API (テキストモデル用とマルチモーダルモデル用) を呼び出し、その成否を返すシンプルなものです。
設計思想としては、「実際の利用状況に近い形でテストすること」を重視しました。単に API サーバーとの疎通確認をするだけでなく、実際に簡単なプロンプトを送信し、意図した形式のレスポンスが返ってくるかまでを確認します。

テキストモデル (Gemini Pro) のヘルスチェック実装例(Django)

テキストモデルのヘルスチェックでは、簡単な質問を投げかけ、正常なテキスト応答が返ってくるかを確認します。

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import vertexai
from vertexai.generative_models import GenerativeModel, SafetySetting
import time
import random
import string

PROJECT_NAME = "xxx"
LOCATION = "asia-northeast1"

@csrf_exemptdef healthcheck_gemini(request):
    if request.method != 'POST':
        return JsonResponse({'status': 'error', 'message': 'Only POST method is allowed'}, status=405)

    model_name = request.POST.get('model_name', 'gemini-1.5-pro-002')

    # ランダムな10桁のシリアルナンバーを生成
    serial_no = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))

    prompt = f"This is a health check Seairl No: {serial_no}. Please respond with 'OK'."

    start_time = time.time()
    try:
        vertexai.init(project=PROJECT_NAME, location=LOCATION)
        model = GenerativeModel(model_name)

        # Set safety settings to allow all content for health check
        safety_settings = [
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HARASSMENT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
        ]

        response = model.generate_content(
            prompt,
            safety_settings=safety_settings
        )

        end_time = time.time()
        return JsonResponse({
            'status': 'success',
            'model': model_name,
            'response_time': end_time - start_time,
            'response': response.text
        })

    except Exception as e:
        import traceback
        traceback.print_exc()
        end_time = time.time()
        return JsonResponse({
            'status': 'error',
            'model': model_name,
            'response_time': end_time - start_time,
            'error': str(e)
        }, status=500)

また、このエンドポイントは以下のように curl で呼び出すことも可能です。model_name パラメータでテスト対象のモデル（例: gemini-1.5-pro-002, gemini-1.5-flash-001 など）を指定することができます。

# テキストモデルのヘルスチェックを実行 (モデル名を指定)
MODEL_NAME="gemini-1.5-flash-001"
curl -X POST <http://localhost:8000/v1/api/healthcheck/gemini/> -F "model_name=$MODEL_NAME" | jq .

# レスポンス例 (成功時)
# {
#   "status": "success",
#   "model": "gemini-1.5-flash-001",
#   "response_time": 1.23456789,
#   "response": "OK."
# }

マルチモーダルモデル (音声入力) のヘルスチェック実装例(Django)

マルチモーダルモデルのヘルスチェックでは、Vertex AI Text-to-Speech で動的に生成した音声データを入力とし、その内容を正しく文字起こしできるかを確認します。これにより、音声入力機能の正常性をテストします。

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import vertexai
from vertexai.generative_models import GenerativeModel, Part, SafetySetting
import base64
import time
import io
from google.cloud import texttospeech
import random
import string

PROJECT_NAME = "xxx"
LOCATION = "asia-northeast1"

@csrf_exemptdef healthcheck_gemini_audio(request):
    if request.method != 'POST':
        return JsonResponse({'status': 'error', 'message': 'Only POST method is allowed'}, status=405)

    model_name = request.POST.get('model_name', 'gemini-1.5-pro-002')
    prompt = "この音声を文字起こしして。環境音は不要です。"

    # ランダムな10桁のシリアルナンバーを生成
    serial_no = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))

    start_time = time.time()
    try:
        # 日本語音声を生成
        client = texttospeech.TextToSpeechClient()
        synthesis_input = texttospeech.SynthesisInput(text=f"ヘルスチェックです。シリアルナンバーは{serial_no}です。")

        voice = texttospeech.VoiceSelectionParams(
            language_code="ja-JP",
            name="ja-JP-Neural2-B"
        )

        audio_config = texttospeech.AudioConfig(
            audio_encoding=texttospeech.AudioEncoding.MP3
        )

        response = client.synthesize_speech(
            input=synthesis_input,
            voice=voice,
            audio_config=audio_config
        )

        # 音声をbase64にエンコード
        audio_bytes = io.BytesIO()
        audio_bytes.write(response.audio_content)
        base64_str = base64.b64encode(audio_bytes.getvalue()).decode('utf-8')

        # Geminiで処理
        vertexai.init(project=PROJECT_NAME, location=LOCATION)
        model = GenerativeModel(model_name)

        safety_settings = [
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
            SafetySetting(
                category=SafetySetting.HarmCategory.HARM_CATEGORY_HARASSMENT,
                threshold=SafetySetting.HarmBlockThreshold.BLOCK_NONE
            ),
        ]

        document_part = Part.from_data(
            mime_type="audio/mp3",
            data=base64.b64decode(base64_str),
        )

        response = model.generate_content(
            [prompt, document_part],
            safety_settings=safety_settings
        )

        end_time = time.time()
        return JsonResponse({
            'status': 'success',
            'model': model_name,
            'response_time': end_time - start_time,
            'response': response.text
        })

    except Exception as e:
        import traceback
        traceback.print_exc()
        end_time = time.time()
        return JsonResponse({
            'status': 'error',
            'model': model_name,
            'response_time': end_time - start_time,
            'error': str(e)
        }, status=500)

同様に、音声モデルのヘルスチェックも curl で実行できます。

# 音声モデルのヘルスチェックを実行 (モデル名を指定)
MODEL_NAME="gemini-1.5-pro-002"
curl -X POST <http://localhost:8000/v1/api/healthcheck/gemini/audio> -F "model_name=$MODEL_NAME" | jq .

# レスポンス例 (成功時)
# {
#   "status": "success",
#   "model": "gemini-1.5-pro-002",
#   "response_time": 3.45678912,
#   "response": "ヘルスチェックです。シリアルナンバーはXXXXXXXXXXです。"
# }

工夫点：音声によるマルチモーダルテストと Vertex AI Text-to-Speech

今回のマルチモーダルモデルのヘルスチェックでは、音声入力のテストに絞って工夫を行いました。利用モデルによりますが、Gemini はテキストや画像だけでなく、音声入力にも対応しています。毎回固定の音声ファイルを用意する代わりに、Vertex AI Text-to-Speech を活用し、テスト実行時に動的に音声データを生成しています。

具体的には、Text-to-Speech API を使って「ヘルスチェックです。シリアルナンバーは{serial_no}です。」のようなランダムなシリアルナンバーを含むテキストを音声データ（MP3形式）に変換します。そして、生成された音声データを Gemini モデル（コード例では gemini-1.5-pro-002 を使用）に入力し、「この音声を文字起こしして。環境音は不要です。」というプロンプトで文字起こしをさせます。
この一連の処理（テキスト生成 -> 音声生成 -> 音声認識・文字起こし）が正常に完了するかを確認することで、音声入力に関するマルチモーダル機能のヘルスチェックを実現しています。シリアルナンバーを毎回変えることで、キャッシュ等に頼らず、実際に処理が行われているかをより確実にテストすることが可能です。

アプローチ②：Datadog Synthetic Test による外形監視

アプローチ①で作成したヘルスチェックエンドポイントを、Datadog の Synthetic Test (API Test) を使って定期的に監視します。

Datadog Synthetic Testの設定は、Terraformで簡単に設定することが可能です。以下は、複数のエンドポイントを10分間隔でチェックし、assertionルールとしてステータスコード、JSON型のチェックを入れています。異常が発生すると、Slackチャンネルに通知されます。

terraform {
  required_providers {
    datadog = {
      source  = "Datadog/datadog"
      version = "~> 3.52.0"
    }
  }
}

variable "workflow_tests" {
  type = list(object({
    test_id    = string
    test_name  = string
    domain     = string
    model_name = string
  }))
  default = [
    {
      test_id    = "test1"
      test_name  = "Healthcheck Gemini 1.5 Pro 001 Text"
      domain     = "<https://example.com/xxx/gemini>"
      model_name = "gemini-1.5-pro-001"
    },
    {
      test_id    = "test2"
      test_name  = "Healthcheck Gemini 1.5 Pro 002 Text"
      domain     = "<https://example.com/xxx/gemini>"
      model_name = "gemini-1.5-pro-002"
    },
    {
      test_id    = "test3"
      test_name  = "Healthcheck Gemini 1.5 Flash 001 Text"
      domain     = "<https://example.com/xxx/gemini>"
      model_name = "gemini-1.5-flash-001"
    },
    {
      test_id    = "test4"
      test_name  = "Healthcheck Gemini 1.5 Flash 002 Text"
      domain     = "<https://example.com/xxx/gemini>"
      model_name = "gemini-1.5-flash-002"
    },
    {
      test_id    = "test5"
      test_name  = "Healthcheck Gemini 1.5 Pro 001 Audio"
      domain     = "<https://example.com/xxx/gemini/audio>"
      model_name = "gemini-1.5-pro-001"
    },
    {
      test_id    = "test6"
      test_name  = "Healthcheck Gemini 1.5 Pro 002 Audio"
      domain     = "<https://example.com/xxx/gemini/audio>"
      model_name = "gemini-1.5-pro-002"
    },
    {
      test_id    = "test7"
      test_name  = "Healthcheck Gemini 1.5 Flash 001 Audio"
      domain     = "<https://example.com/xxx/gemini/audio>"
      model_name = "gemini-1.5-flash-001"
    },
    {
      test_id    = "test8"
      test_name  = "Healthcheck Gemini 1.5 Flash 002 Audio"
      domain     = "<https://example.com/xxx/gemini/audio>"
      model_name = "gemini-1.5-flash-002"
    }
  ]
}

resource "datadog_synthetics_test" "workflow_test" {
  for_each  = { for test in var.workflow_tests : test.test_id => test }
  name      = "LLM Healthcheck: ${each.value.test_name}"
  type      = "api"
  subtype   = "http"
  status    = "live"
  message   = "Notify @slack-xxx-notifications"
  locations = ["aws:ap-northeast-1"]
  tags      = ["type:llm"]

  request_definition {
    method = "POST"
    url    = each.value.domain
    body = jsonencode({
      model_name = each.value.model_name
    })
  }

  request_headers = {
    Content-Type = "application/json"
  }

  assertion {
    type     = "statusCode"
    operator = "is"
    target   = "200"
  }

  assertion {
    type     = "body"
    operator = "validatesJSONPath"
    targetjsonpath {
      jsonpath    = "$.status"
      operator    = "is"
      targetvalue = "success"
    }
  }

  options_list {
    tick_every = 600
    retry {
      count    = 2
      interval = 300
    }
    monitor_options {
      renotify_interval = 120
    }
  }
}

これにより、ヘルスチェックエンドポイントへのアクセス状況やレスポンスタイムがメトリクスとして可視化され、異常があればすぐにアラートで検知できるようになります。

ヘルスチェックでエラー発生時、復旧時に、Slackに自動通知されます。

よく使う機能

APM（Application Performance Monitoring）
Synthetics Monitoring & Testing
SLO監視と通知

ツールの良い点

日本語ドキュメントが充実しており、導入時に大きな詰まりがなかった。
AWS/GCPなど外部サービスとの統合が容易。
UIが直感的で、初見でも扱いやすい。

ツールの課題点

Synthetic TestでのE2E監視時、一度失敗するとスケジュールを止めない限り実行が継続され、無駄なコストが発生する点。

ツールを検討されている方へ

日本語ドキュメントはあるものの、英語版のほうが最新情報が得られやすいため、英語ドキュメントも並行して参照するのがおすすめです。

今後の展望

このヘルスチェックと外形監視体制により、Geminiの不安定な挙動も格段に早く検知できるようになりました。これによって、問題発生時の初動が早まり、障害によるサービスへの影響を最小限に抑えられるようになりました。また、Datadog で収集したレスポンスタイムのメトリクスは、Gemini のパフォーマンス傾向を把握する上でも役立っています。

今後は、さらに詳細なエラー内容の分析や、モデル別パフォーマンスの比較などの監視項目への追加や、ヘルスチェックで検知した異常に対して自動的にフォールバック処理を行うような仕組みも将来的には実装していきたいと考えています。

株式会社ワンキャリア / MikiPWata

メンバー / フロントエンドエンジニア