AI アプリケーション監視アラート：SLI/SLO 指標定義とアラート戦略

AI API を本番環境に統合する際、監視とアラート体制の構築は可用性とユーザー体験の要です。本稿では、私自身が必要な API 監視基盤を構築した際に直面した具体的なエラーシナリオを起点として、SLI（Service Level Indicator）と SLO（Service Level Objective）の定義方法、そして効果的なアラート戦略について実践的に解説します。

1. 実際のエラーシナリオから始める

私が AI API 監視の実務で対応した最初の重大インシデントは、深夜に発生した「ConnectionError: timeout after 30 seconds」でした。API 応答が完全に停止したのではなく、レイテンシが急上昇してtimeoutを連発する状態でした。この経験から学んだのは、可用性だけでなくレイテンシ分布を監視する重要性です。

# 遭遇した典型的なエラーコード
HTTP Error: 401 Unauthorized
Request ID: req_abc123 - API キーが無効または期限切れ

HTTP Error: 429 Too Many Requests  
Rate limit exceeded: 60 requests per minute

HTTP Error: 500 Internal Server Error
upstream connect error: connection reset

ConnectionError: timeout after 30 seconds
応答時間が SLA の 3 秒を著しく超過

これらのエラーは単なる例外処理では不十分で、継続的な監視と主动的なアラートが必要です。

2. SLI 指標の設計

SLI は 서비스の、技術的な 측정指标です。AI API 監視において最も重要な SLI を以下に定義します。

2.1 主要 SLI 指標一覧

2.2 HolySheep AI での監視実装

HolySheep AI の API は 今すぐ登録 して取得した API キーを使用し、リアルタイムの監視データを収集できます。HolySheep は ¥1=$1 の為替レートで、公式 ¥7.3=$1 比で85%の節約を実現するため、コスト効率のよい監視アラート構築に適しています。

#!/usr/bin/env python3
"""
AI API 監視クライアント - HolySheep AI
ベースURL: https://api.holysheep.ai/v1
"""

import time
import httpx
import statistics
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime

@dataclass
class RequestMetrics:
    """単一リクエストのメトリクス"""
    request_id: str
    start_time: float
    end_time: float
    status_code: int
    response_time_ms: float
    model: str
    error_message: Optional[str] = None

    @property
    def success(self) -> bool:
        return 200 <= self.status_code < 300

    @property
    def is_timeout(self) -> bool:
        return self.error_message and "timeout" in self.error_message.lower()

@dataclass
class SLIMetrics:
    """SLI 計算用聚合メトリクス"""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    timeout_requests: int = 0
    rate_limited_requests: int = 0
    response_times: List[float] = field(default_factory=list)
    error_messages: List[str] = field(default_factory=list)

    def calculate_availability(self) -> float:
        """可用性 = 成功応答 / 総要求 × 100"""
        if self.total_requests == 0:
            return 100.0
        return (self.successful_requests / self.total_requests) * 100

    def calculate_error_rate(self) -> float:
        """エラー率 = (4xx + 5xx) / 総要求 × 100"""
        if self.total_requests == 0:
            return 0.0
        return (self.failed_requests / self.total_requests) * 100

    def calculate_p50_latency(self) -> float:
        if not self.response_times:
            return 0.0
        return statistics.median(self.response_times)

    def calculate_p95_latency(self) -> float:
        if not self.response_times:
            return 0.0
        sorted_times = sorted(self.response_times)
        index = int(len(sorted_times) * 0.95)
        return sorted_times[min(index, len(sorted_times) - 1)]

    def calculate_p99_latency(self) -> float:
        if not self.response_times:
            return 0.0
        sorted_times = sorted(self.response_times)
        index = int(len(sorted_times) * 0.99)
        return sorted_times[min(index, len(sorted_times) - 1)]

    def get_sli_report(self) -> dict:
        """SLI レポート生成"""
        return {
            "timestamp": datetime.utcnow().isoformat(),
            "availability": f"{self.calculate_availability():.2f}%",
            "error_rate": f"{self.calculate_error_rate():.3f}%",
            "p50_latency_ms": f"{self.calculate_p50_latency():.1f}",
            "p95_latency_ms": f"{self.calculate_p95_latency():.1f}",
            "p99_latency_ms": f"{self.calculate_p99_latency():.1f}",
            "rate_limited_rate": f"{(self.rate_limited_requests/self.total_requests*100) if self.total_requests > 0 else 0:.2f}%",
            "timeout_count": self.timeout_requests
        }

class HolySheepMonitor:
    """HolySheep AI API 監視クライアント"""

    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.metrics = SLIMetrics()
        self._client = httpx.Client(
            timeout=30.0,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )

    def _record_request(self, metric: RequestMetrics):
        """リクエストメトリクスを記録"""
        self.metrics.total_requests += 1
        self.metrics.response_times.append(metric.response_time_ms)

        if metric.success:
            self.metrics.successful_requests += 1
        else:
            self.metrics.failed_requests += 1

        if metric.status_code == 429:
            self.metrics.rate_limited_requests += 1

        if metric.is_timeout:
            self.metrics.timeout_requests += 1

        if metric.error_message:
            self.metrics.error_messages.append(metric.error_message)

    def chat_completion_with_monitoring(
        self,
        model: str,
        messages: List[dict],
        max_tokens: int = 1000
    ) -> dict:
        """監視付きの chat completion 呼び出し"""
        start_time = time.time()
        request_id = f"req_{int(start_time * 1000)}"

        try:
            response = self._client.post(
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
GPT-4o Realtime API 音声对话開発完全ガイド
AI 中转站 API 鉴权机制：API Key 安全配置详解
AI画像理解とマルチモーダル処理 パフォーマンス最適化テクニック