レイテンシ、コスト、品質低下へのアラート
p99レイテンシ、リクエストごとのコスト、自動品質スコアにアラート閾値を設定し、LLMパイプラインの品質が低下した際にSlackまたはPagerDutyへアラートを送ります。
「レイテンシ、コスト、品質低下へのアラート」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。
LLMシステムでアラートが重要な理由
LLMアプリケーションの障害は、微妙かつ徐々に進行する形で発生します。プロンプトの変更によって平均レイテンシが30%増加することもあれば、検索機能の改善によって回答品質がわずかに低下することもあります。また、利用急増によって1日のコストが予算の5倍に達することもあります。積極的なアラートがなければ、ユーザーから苦情を受けたり月々の請求書が届いたりするまで、こうした問題に気付けません。アラートによって、受け身の障害対応を能動的な運用へと変えられます。
3種類のアラートカテゴリ
LLMアプリケーションのアラートは、3つのカテゴリに分けられます。レイテンシアラートは、応答時間がユーザー体験のしきい値（例：p99 > 10秒）を超えたときに発生します。コストアラートは、リクエストごとのコストまたは1日の支出が予算のしきい値を超えたときに発生し、予想外の高額請求を防ぎます。品質アラートは、自動品質メトリクス（LLM-as-Judgeのスコア、ユーザー満足度、忠実性スコア）が許容可能な下限を下回ったときに発生します。各カテゴリには、異なる計測方法とアラートチャネルが必要です。
from dataclasses import dataclass
@dataclass
class AlertThresholds:
# Latency (milliseconds)
p50_latency_ms: int = 2000 # median should be under 2s
p99_latency_ms: int = 10000 # 99th percentile under 10s
# Cost (USD)
max_cost_per_request: float = 0.05 # alert if one request costs > 5 cents
max_daily_spend: float = 50.00 # alert if daily spend exceeds $50
# Quality (0.0 to 1.0 scale)
min_quality_score: float = 0.75 # alert if rolling avg drops below 75%
min_faithfulness: float = 0.80 # alert if RAGAS faithfulness drops below 80%
DEFAULT_THRESHOLDS = AlertThresholds()レイテンシメトリクスの収集
レイテンシに対してアラートを発生させるには、まず一貫した方法でレイテンシを収集する必要があります。3つのレイテンシ要素を個別に計測します。ストリーミングUXにとって重要な最初のトークンまでの時間（TTFT）、応答全体の時間、検索と生成にかかるステップごとのレイテンシです。これらを時系列データ（リクエストごとに1つのデータポイント）として保存すると、過去N分またはN時間のパーセンタイルや移動平均を計算できます。
import time
from collections import deque
from statistics import quantiles
class LatencyTracker:
def __init__(self, window_size=100):
self.window = deque(maxlen=window_size) # rolling window of latencies
def record(self, latency_ms: float):
self.window.append(latency_ms)
def p50(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[49]
def p99(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[98]
def check_alert(self, thresholds: AlertThresholds) -> list[str]:
alerts = []
if self.p99() > thresholds.p99_latency_ms:
alerts.append(f'p99 latency {self.p99():.0f}ms exceeds {thresholds.p99_latency_ms}ms')
if self.p50() > thresholds.p50_latency_ms:
alerts.append(f'p50 latency {self.p50():.0f}ms exceeds {thresholds.p50_latency_ms}ms')
return alerts
latency_tracker = LatencyTracker()コストの追跡とアラート
コスト監視では、複数の粒度で支出を追跡する必要があります。高コストな外れ値クエリを検出するリクエストごとのコスト、利用急増を検出する時間別・日別の合計、アプリケーションのどの部分が最も高コストかを特定する機能別のコストです。リクエストごとのコスト異常にはすぐにアラートを発生させ、1日の予算上限にはスケジュールされたチェックを使用します。
from datetime import datetime, date
import threading
class CostTracker:
def __init__(self):
self._lock = threading.Lock()
self._daily_spend = {} # date -> total USD
self._per_request = [] # list of (timestamp, cost)
def record(self, cost_usd: float) -> list[str]:
today = date.today().isoformat()
alerts = []
with self._lock:
# Track daily spend
self._daily_spend[today] = self._daily_spend.get(today, 0) + cost_usd
self._per_request.append((datetime.now(), cost_usd))
# Alert on expensive single requests
if cost_usd > DEFAULT_THRESHOLDS.max_cost_per_request:
alerts.append(f'Expensive request: ${cost_usd:.4f} (threshold: ${DEFAULT_THRESHOLDS.max_cost_per_request})')
# Alert on daily budget exceeded
daily_total = self._daily_spend[today]
if daily_total > DEFAULT_THRESHOLDS.max_daily_spend:
alerts.append(f'Daily budget exceeded: ${daily_total:.2f} > ${DEFAULT_THRESHOLDS.max_daily_spend}')
return alerts
cost_tracker = CostTracker()品質スコアのアラート
品質アラートは、インフラストラクチャのメトリクスだけでは品質を測定できず、意味的な評価が必要になるため、最も複雑なカテゴリです。最もスケーラブルな方法はサンプリングした自動評価です。本番リクエストのランダムサンプル（5～10%）に対してLLM-as-Judge評価器を非同期で実行し、スコアを保存して移動平均を計算します。移動平均が品質の下限を下回ったときにアラートを発生させます。
import random
from openai import OpenAI
client = OpenAI()
def evaluate_response_quality(question: str, answer: str) -> float:
judge_prompt = f'''Rate the quality of this AI assistant response on a scale from 0 to 1.
Question: {question}
Answer: {answer}
Return only a JSON object: {{"score": 0.85, "reason": "brief reason"}}
Scoring guide:
1.0 = Perfect, accurate, helpful
0.75 = Good, minor issues
0.5 = Acceptable but incomplete
0.25 = Poor, major gaps
0.0 = Completely wrong or harmful'''
response = client.chat.completions.create(
model='gpt-4o-mini', # cheaper model for evaluation
messages=[{'role': 'user', 'content': judge_prompt}],
response_format={'type': 'json_object'}
)
import json
result = json.loads(response.choices[0].message.content)
return float(result['score'])
def maybe_evaluate(question: str, answer: str, sample_rate=0.1):
if random.random() < sample_rate:
score = evaluate_response_quality(question, answer)
quality_tracker.record(score)
return score
return Noneトレンド検出のための移動平均
1件の悪い回答だけでは、システム全体の問題とは判断できません。移動平均を一定の時間枠（例：直近1時間または直近100リクエスト）で計算し、傾向を検出します。移動平均がしきい値を超え、その状態が10分以上続く場合は実際の問題である可能性が高く、一時的な急上昇はノイズかもしれません。指数加重移動平均（EWMA）は、単純な移動平均よりも直近の変化に速く反応します。
class RollingQualityMonitor:
def __init__(self, window=50, alert_threshold=0.75, ewma_alpha=0.1):
self.scores = []
self.window = window
self.alert_threshold = alert_threshold
self.alpha = ewma_alpha # EWMA decay factor
self.ewma_score = None
def record(self, score: float):
self.scores.append(score)
if len(self.scores) > self.window:
self.scores.pop(0)
# Update exponentially weighted moving average
if self.ewma_score is None:
self.ewma_score = score
else:
self.ewma_score = self.alpha * score + (1 - self.alpha) * self.ewma_score
def should_alert(self) -> bool:
if len(self.scores) < 10: # not enough data yet
return False
return self.ewma_score < self.alert_threshold
def summary(self) -> dict:
if not self.scores:
return {}
return {
'rolling_avg': sum(self.scores) / len(self.scores),
'ewma': self.ewma_score,
'sample_count': len(self.scores),
'alert': self.should_alert()
}
quality_tracker = RollingQualityMonitor()アラートの振り分け：SlackとPagerDuty
アラートは、適切なチャネルを通じて適切な担当者に届いて初めて役立ちます。重大度に応じてアラートを振り分けます。品質低下やコスト予算のアラートは、営業時間中にチームが監視するSlackチャンネルに送ります。重大なしきい値を超えるレイテンシ（例：p99 > 30秒）や急激なコスト急増（例：5分間で通常の10倍）は、オンコール担当者へ即座にエスカレーションするためPagerDutyに送ります。
import requests
def send_slack_alert(message: str, severity: str, webhook_url: str):
color = {'critical': '#ff0000', 'warning': '#ff9900', 'info': '#36a64f'}[severity]
payload = {
'attachments': [{
'color': color,
'title': f'LLM Alert [{severity.upper()}]',
'text': message,
'footer': 'AI Pipeline Monitor'
}]
}
requests.post(webhook_url, json=payload)
def send_pagerduty_alert(title: str, details: str, routing_key: str):
payload = {
'routing_key': routing_key,
'event_action': 'trigger',
'payload': {
'summary': title,
'severity': 'critical',
'source': 'ai-pipeline-monitor',
'custom_details': {'details': details}
}
}
requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)
def route_alert(alert_text: str, severity: str):
send_slack_alert(alert_text, severity, SLACK_WEBHOOK_URL)
if severity == 'critical':
send_pagerduty_alert(alert_text, alert_text, PAGERDUTY_ROUTING_KEY)アラート評価ループ
アラートロジックは、リクエスト処理にインラインで組み込むのではなく、スケジュールされたループで実行する必要があります。アラートチェックを非同期で実行すると、本番リクエストへのレイテンシ追加を防げます。ほとんどのアラート要件では、60秒ごとに実行するバックグラウンドスレッドまたはスケジュールジョブで十分です。リクエスト処理中にメトリクスを収集し、バックグラウンドループでしきい値を評価します。
import threading
import time
def alert_evaluation_loop(interval_seconds=60):
while True:
try:
# Check latency
latency_alerts = latency_tracker.check_alert(DEFAULT_THRESHOLDS)
for alert in latency_alerts:
route_alert(f'LATENCY: {alert}', 'warning')
# Check quality
quality_summary = quality_tracker.summary()
if quality_summary.get('alert'):
msg = f'QUALITY DEGRADATION: Rolling avg {quality_summary["ewma"]:.2f} below threshold {DEFAULT_THRESHOLDS.min_quality_score}'
route_alert(msg, 'warning')
print(f'Alert check complete. Quality: {quality_summary.get("ewma", "N/A")}')
except Exception as e:
print(f'Alert evaluation error: {e}')
time.sleep(interval_seconds)
# Start in background thread
alert_thread = threading.Thread(target=alert_evaluation_loop, daemon=True)
alert_thread.start()アラート疲れとしきい値の調整
アラート疲れは、アラートが頻繁に発生しすぎて、チームが無視し始める状態です。これを防ぐには、まず控えめな（高い）しきい値から始め、基準値を把握しながら徐々に厳しくします。また、発生前に複数回の評価サイクルにわたって継続することを要求し、関連するアラートを1つの通知にまとめ、意味のある対応につながらないアラートを定期的に見直して廃止します。
class AlertDebouncer:
def __init__(self, required_consecutive_fires=3):
self.required = required_consecutive_fires
self.fire_counts = {} # alert_name -> consecutive fires
self.already_firing = set() # alert_names currently active
def should_fire(self, alert_name: str, condition: bool) -> bool:
if condition:
self.fire_counts[alert_name] = self.fire_counts.get(alert_name, 0) + 1
if self.fire_counts[alert_name] >= self.required and alert_name not in self.already_firing:
self.already_firing.add(alert_name)
return True # fire the alert (first time condition sustained)
else:
if self.fire_counts.get(alert_name, 0) > 0:
self.fire_counts[alert_name] = 0
self.already_firing.discard(alert_name) # condition cleared
return False # either not sustained or already firing (no duplicate alert)
debouncer = AlertDebouncer(required_consecutive_fires=3)統計手法によるコスト異常の監視
単純なしきい値アラートでは、2日間でコストが徐々に50%増加するような微妙なコスト異常を見逃します。統計的な異常検出を使用します。時間ごとのコストの移動平均と標準偏差を計算し、現在の時間のコストが平均よりN標準偏差を超えて高い場合にアラートを発生させます（Zスコアアラート）。これにより、平日と週末のトラフィックの違いのような自然な利用パターンにも自動的に適応できます。
import statistics
def compute_z_score(recent_value: float, historical_values: list[float]) -> float:
if len(historical_values) < 5:
return 0 # not enough data
mean = statistics.mean(historical_values)
stdev = statistics.stdev(historical_values)
if stdev == 0:
return 0
return (recent_value - mean) / stdev
def check_cost_anomaly(current_hour_cost: float, historical_hourly_costs: list[float]) -> str | None:
z = compute_z_score(current_hour_cost, historical_hourly_costs)
if z > 3.0: # more than 3 standard deviations above mean
mean = statistics.mean(historical_hourly_costs)
return f'Cost anomaly: ${current_hour_cost:.2f} this hour (normal: ${mean:.2f}, z={z:.1f})'
return None運用ダッシュボードの構築
アラートが受動的なレイヤーであるのに対し、リアルタイムの運用ダッシュボードは能動的なレイヤーです。現在のp50/p99レイテンシ、1分あたりのリクエスト数、現在の品質スコア（移動EWMA）、現時点までの1日のコストと予算との比較、直近1時間で最もコストが高いリクエスト上位5件など、リアルタイムメトリクスを表示するシンプルなダッシュボードを構築します。これにより、アラートが発生するのを待たずに、システムの状態を一目で把握できます。
クイックチェック
このレッスンで学んだ、LLMアプリケーションのメトリクスに対するアラートについて、理解度を確認しましょう。
レッスンのまとめ
このレッスンでは、3つのアラートカテゴリでLLMシステムをカバーできること（レイテンシ、コスト、品質）、それぞれに異なる計装が必要であることを学びました。また、移動平均とEWMAは、個々のリクエストに対するしきい値チェックよりも、品質の緩やかな低下を検出するのに適しています。さらに、アラートのデバウンスは、発報前に複数の評価サイクルにわたって条件が継続することを要求し、アラート疲れを防ぎます。次は、プロンプトインジェクション攻撃とAIセキュリティについて詳しく見ていきます。
よくある質問
「レイテンシ、コスト、品質低下へのアラート」レッスンは無料ですか？
はい。「レイテンシ、コスト、品質低下へのアラート」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。
「レイテンシ、コスト、品質低下へのアラート」で何を学びますか？
p99レイテンシ、リクエストごとのコスト、自動品質スコアにアラート閾値を設定し、LLMパイプラインの品質が低下した際にSlackまたはPagerDutyへアラートを送ります。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Engineering Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「レイテンシ、コスト、品質低下へのアラート」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Engineering Academyレッスンでコードを書いて実行できますか？
はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- LLMアプリのデバッグが難しい理由
- LangSmithによるトレーシング
- モデルに依存しない可観測性のためのLangfuse
- レイテンシ、コスト、品質低下へのアラート