プロンプトパイプラインの監視とアラート
本番環境のプロンプト向けに、ダッシュボード、異常検知、オンコールアラートを構築します。
「プロンプトパイプラインの監視とアラート」はCoddyKit上の無料AI Prompt Engineeringレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Prompt Engineering学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Prompt Engineeringコースには全4レッスンが含まれています。
本番環境のプロンプトパイプラインにはモニタリングが必要
本番環境のプロンプトパイプラインはインフラストラクチャです。他のサービスと同様に、ダッシュボード、アラート、ランブックが必要です。モニタリングを行わないと、ユーザーから苦情が来たり請求書が届いたりするまで、コストの急増、品質の低下、レイテンシの急増に気付けません。
主要指標：レイテンシのパーセンタイル
レイテンシをP50、P95、P99で追跡します。平均値ではテールの挙動が隠れてしまいます。P99が30秒ということは、P50が2秒でもユーザーの1%が30秒待つことを意味します。LLMのレイテンシは、出力の長さに応じて変化するため、本質的にばらつきがあります。
import time
import statistics
from collections import deque
class LatencyTracker:
def __init__(self, window_size=1000):
self.samples = deque(maxlen=window_size)
def record(self, latency_ms):
self.samples.append(latency_ms)
def percentile(self, p):
if not self.samples:
return None
sorted_samples = sorted(self.samples)
idx = int(len(sorted_samples) * p / 100)
return sorted_samples[min(idx, len(sorted_samples) - 1)]
def report(self):
if not self.samples:
return {}
return {
'count': len(self.samples),
'p50_ms': self.percentile(50),
'p95_ms': self.percentile(95),
'p99_ms': self.percentile(99),
'max_ms': max(self.samples)
}
tracker = LatencyTracker()
for ms in [1200, 1100, 1300, 1150, 8500, 1200, 1250, 15000, 1100, 1300]:
tracker.record(ms)
print(tracker.report())ダッシュボード指標としての日次コスト
日次のAPIコストを主要なダッシュボード指標として追跡します。トークン使用量からリアルタイムに計算し、週次の移動平均と比較して、コストの急増を早期に検出します。
from datetime import datetime, timedelta
from collections import defaultdict
class CostTracker:
def __init__(self):
self.daily_costs = defaultdict(float) # date: total_cost_usd
def record_call(self, model, input_tokens, output_tokens):
pricing = {
'gpt-4o-mini': (0.15, 0.60),
'gpt-4o': (2.50, 10.00),
'claude-opus-4-5': (15.00, 75.00), # per 1M tokens
'claude-haiku-4-5': (0.25, 1.25)
}
if model not in pricing:
return
input_price, output_price = pricing[model]
cost = (input_tokens / 1_000_000 * input_price +
output_tokens / 1_000_000 * output_price)
today = datetime.utcnow().date().isoformat()
self.daily_costs[today] += cost
def today_cost(self):
today = datetime.utcnow().date().isoformat()
return round(self.daily_costs[today], 4)
def weekly_avg_daily_cost(self):
dates = sorted(self.daily_costs.keys())[-7:]
if not dates:
return 0
return round(sum(self.daily_costs[d] for d in dates) / len(dates), 4)
cost_tracker = CostTracker()
cost_tracker.record_call('gpt-4o-mini', 800, 200)
print('Today cost:', cost_tracker.today_cost())エラー率のモニタリング
エラー率を総リクエスト数に対する割合として追跡します。エラーには、APIの失敗、タイムアウト、解析に失敗する不正な形式の出力、モデルによる拒否が含まれます。実行可能なアラートを設定できるよう、エラーの種類を分けて管理します。
from collections import Counter
class ErrorRateTracker:
ERROR_TYPES = [
'api_error', 'timeout', 'rate_limit',
'parse_failure', 'model_refusal', 'context_length_exceeded'
]
def __init__(self, window_size=1000):
self.total = 0
self.errors = Counter()
self.recent = deque(maxlen=window_size) # True=error, False=success
def record(self, success, error_type=None):
self.total += 1
self.recent.append(not success)
if not success and error_type:
self.errors[error_type] += 1
def error_rate(self):
if not self.recent:
return 0.0
return sum(self.recent) / len(self.recent)
def report(self):
return {
'error_rate': round(self.error_rate(), 4),
'total_requests': self.total,
'error_breakdown': dict(self.errors.most_common())
}
err_tracker = ErrorRateTracker()
for i in range(100):
if i % 20 == 0:
err_tracker.record(False, 'timeout')
else:
err_tracker.record(True)
print(err_tracker.report())品質スコアの傾向追跡
品質スコアを移動時系列として追跡し、傾向を把握できるようにします。数日かけて徐々に品質が低下する場合、急激な低下より気付きにくい一方で、ユーザーの信頼に同じように大きな損害を与える可能性があります。
from datetime import datetime
import statistics
class QualityTrendTracker:
def __init__(self, window_minutes=60):
self.window_seconds = window_minutes * 60
self.samples = [] # (timestamp, score)
def record(self, score):
now = time.time()
self.samples.append((now, score))
# Purge old samples outside window
cutoff = now - self.window_seconds
self.samples = [(t, s) for t, s in self.samples if t >= cutoff]
def rolling_avg(self):
if not self.samples:
return None
return round(statistics.mean(s for _, s in self.samples), 3)
def trend(self):
if len(self.samples) < 10:
return 'insufficient_data'
mid = len(self.samples) // 2
first_half_avg = statistics.mean(s for _, s in self.samples[:mid])
second_half_avg = statistics.mean(s for _, s in self.samples[mid:])
delta = second_half_avg - first_half_avg
if delta > 0.1:
return 'improving'
elif delta < -0.1:
return 'declining'
return 'stable'
qt = QualityTrendTracker(window_minutes=60)
for score in [4.2, 4.1, 4.0, 3.9, 3.8, 3.7, 3.6, 3.5, 3.4, 3.3]:
qt.record(score)
print('Rolling avg:', qt.rolling_avg(), '| Trend:', qt.trend())ダッシュボードパネルの設計
適切に設計されたモニタリングダッシュボードでは、指標を論理的なセクションにまとめます。すべてのプロンプトパイプラインのダッシュボードに必要な4つの主要パネルを定義します。
DASHBOARD_PANELS = {
'Panel 1: Availability': [
'Error rate (%) — last 1h, 24h, 7d',
'Error type breakdown (timeout vs API vs parse)',
'P99 latency (alert if > 10s)',
'Success rate by prompt_id and version'
],
'Panel 2: Performance': [
'P50 / P95 / P99 latency (time series)',
'Latency by model and prompt version',
'Time to first token (streaming)',
'Latency heatmap by hour of day'
],
'Panel 3: Cost': [
'Daily cost USD (actual vs budget)',
'Cost per request by model',
'Cost trend (7-day rolling)',
'Top 10 most expensive prompt_ids'
],
'Panel 4: Quality': [
'Average quality score (rolling 1h)',
'Quality trend by prompt version',
'User satisfaction (thumbs, retry rate)',
'Low-quality alert rate'
]
}
for panel, metrics in DASHBOARD_PANELS.items():
print(f'\n{panel}:')
for m in metrics:
print(f' - {m}')アラートルールの実装
指標がしきい値を超えるとアラートが発生します。スケジュールに従って実行され、PagerDuty、Slack、またはメールに通知を送る単純なポーリングチェックとしてアラートを実装します。
ALERT_RULES = [
{
'name': 'HighErrorRate',
'condition': lambda m: m['error_rate'] > 0.05,
'severity': 'CRITICAL',
'message': 'Error rate {error_rate:.1%} exceeds 5% threshold',
'for_minutes': 5
},
{
'name': 'HighLatencyP95',
'condition': lambda m: m.get('p95_latency_ms', 0) > 10000,
'severity': 'WARNING',
'message': 'P95 latency {p95_latency_ms}ms exceeds 10s threshold',
'for_minutes': 3
},
{
'name': 'CostSpike',
'condition': lambda m: m.get('today_cost', 0) > m.get('weekly_avg', 1) * 2,
'severity': 'WARNING',
'message': 'Daily cost ${today_cost:.2f} is 2x weekly average',
'for_minutes': 60
},
{
'name': 'QualityRegression',
'condition': lambda m: m.get('quality_avg', 5) < 3.5,
'severity': 'CRITICAL',
'message': 'Quality score {quality_avg:.2f} below 3.5 threshold',
'for_minutes': 15
}
]
def check_alerts(metrics):
fired = []
for rule in ALERT_RULES:
if rule['condition'](metrics):
msg = rule['message'].format(**metrics)
fired.append({'name': rule['name'], 'severity': rule['severity'],
'message': msg})
return fired通知の振り分け
アラート通知は重大度に応じて振り分けます。CRITICALアラートはオンコール担当者に直ちにページ通知し、WARNINGアラートはSlackに投稿し、INFOアラートはログファイルに記録します。確認されないCRITICALアラートにはエスカレーションタイマーを使用します。
import requests
SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
PAGERDUTY_API_KEY = 'YOUR_PD_KEY'
def send_slack_alert(message, severity='WARNING'):
emoji = ':rotating_light:' if severity == 'CRITICAL' else ':warning:'
payload = {'text': f'{emoji} *{severity}*: {message}'}
try:
requests.post(SLACK_WEBHOOK, json=payload, timeout=5)
print(f'Slack alert sent: {message[:60]}')
except Exception as e:
print(f'Slack notification failed: {e}')
def send_pagerduty_alert(summary, severity='critical'):
payload = {
'routing_key': PAGERDUTY_API_KEY,
'event_action': 'trigger',
'payload': {
'summary': summary,
'severity': severity,
'source': 'prompt-pipeline-monitor'
}
}
try:
response = requests.post(
'https://events.pagerduty.com/v2/enqueue',
json=payload, timeout=10
)
print(f'PagerDuty alert: {response.status_code}')
except Exception as e:
print(f'PagerDuty notification failed: {e}')
def dispatch_alert(alert):
if alert['severity'] == 'CRITICAL':
send_pagerduty_alert(alert['message'])
send_slack_alert(alert['message'], 'CRITICAL')
else:
send_slack_alert(alert['message'], 'WARNING')オンコール用ランブックの構成
すべてのアラートには、オンコール担当のエンジニアが具体的に何をすべきかを示す対応するランブックが必要です。適切に記述されたランブックにより、MTTR（平均復旧時間）を数時間から数分に短縮できます。
# Runbook template for HighErrorRate alert
HIGH_ERROR_RATE_RUNBOOK = '''
## Alert: HighErrorRate
### Trigger: Error rate > 5% for > 5 minutes
### Severity: CRITICAL
## Immediate Actions (< 5 minutes)
1. Check error type breakdown in dashboard: Panel 1 > Error type breakdown
- timeout errors -> see Timeout Runbook
- api_error -> check LLM provider status page
- parse_failure -> check if model output format changed
2. Check if this is related to a recent deployment:
python manage.py prompt list-recent-activations --last-hours 2
3. If error rate > 20%, trigger emergency rollback:
python manage.py prompt activate --prompt-id <id> --version <last-stable>
## Investigation (< 30 minutes)
4. Sample failed requests from log:
grep error_rate /var/log/prompt-pipeline.log | tail -100
5. Check model provider status:
- OpenAI: https://status.openai.com
- Anthropic: https://status.anthropic.com
## Resolution
6. If provider outage: activate fallback model routing
7. If prompt change: rollback to previous version
8. If code change: rollback deployment
9. Document in post-mortem after resolution
'''
print(HIGH_ERROR_RATE_RUNBOOK[:400], '...')プロンプトパイプラインの構造化ログ
構造化ログ（1行ごとのJSON）を使うと、Datadog、Splunk、CloudWatchなどのログ管理システムで強力なフィルタリングと集計を行えます。すべてのLLM呼び出しで、構造化されたログエントリを1件生成する必要があります。
import json
import time
from datetime import datetime
def log_llm_call(request_id, prompt_id, version, model, messages,
response_text, latency_ms, input_tokens,
output_tokens, error=None):
log_entry = {
'ts': datetime.utcnow().isoformat() + 'Z',
'level': 'ERROR' if error else 'INFO',
'service': 'prompt-pipeline',
'request_id': request_id,
'prompt_id': prompt_id,
'version': version,
'model': model,
'latency_ms': round(latency_ms),
'input_tokens': input_tokens,
'output_tokens': output_tokens,
'error': str(error) if error else None,
'response_preview': response_text[:100] if response_text else None
}
print(json.dumps(log_entry))
# In production: ship to log aggregator
# logger.info(json.dumps(log_entry))
# Example log output:
# {"ts":"2024-08-15T10:00:01Z","level":"INFO",
# "prompt_id":"summarize-article","version":"1.2.0",
# "model":"gpt-4o-mini","latency_ms":1234,
# "input_tokens":800,"output_tokens":150,...}
log_llm_call('req-001', 'summarize-article', '1.2.0', 'gpt-4o-mini',
[], 'Summary text...', 1234, 800, 150)モニタリングシステムのアーキテクチャ
すべてのモニタリングコンポーネントを、プロンプトパイプラインと並行して動作する一貫したシステムに組み立てます。単純なポーリングループで、アラートの評価と通知を処理できます。
import time
class PromptPipelineMonitor:
def __init__(self):
self.latency = LatencyTracker()
self.errors = ErrorRateTracker()
self.quality = QualityTrendTracker()
self.cost = CostTracker()
def record(self, model, latency_ms, input_tokens, output_tokens,
quality_score=None, error=None, error_type=None):
self.latency.record(latency_ms)
self.errors.record(error is None, error_type)
self.cost.record_call(model, input_tokens, output_tokens)
if quality_score:
self.quality.record(quality_score)
def current_metrics(self):
lat = self.latency.report()
err = self.errors.report()
return {
**lat, **err,
'quality_avg': self.quality.rolling_avg() or 5.0,
'quality_trend': self.quality.trend(),
'today_cost': self.cost.today_cost(),
'weekly_avg': self.cost.weekly_avg_daily_cost()
}
def run_alert_check(self):
metrics = self.current_metrics()
alerts = check_alerts(metrics)
for alert in alerts:
dispatch_alert(alert)
return alerts
monitor = PromptPipelineMonitor()
print('Monitor initialized. Call monitor.record() on each LLM call.')理解度チェック
プロンプトパイプラインのP50レイテンシは1.5秒ですが、P99は28秒です。これは本番環境での挙動について何を示していますか。
モニタリングとアラートのまとめ
本番環境のプロンプトパイプラインをモニタリングするには、5つの測定カテゴリと、それぞれに対応するアラートルールが必要です：
- レイテンシ：P50/P95/P99を追跡し、P95 > 10sでアラートを発生させます
- コスト：日次コストと週次平均を比較し、コストが2倍に急増したらアラートを発生させます
- エラー率：合計の割合とエラー種別の内訳を追跡し、5%を超えたらアラートを発生させます
- 品質スコア：移動平均の傾向を追跡し、3.5/5を下回ったらアラートを発生させます
- アラート：CRITICAL → PagerDuty + Slack、WARNING → Slackのみ
- ランブック：すべてのアラート種別について、解決手順を段階的に示します
- ダッシュボード：可用性、パフォーマンス、コスト、品質をカバーする4つのパネルを配置します
よくある質問
「プロンプトパイプラインの監視とアラート」レッスンは無料ですか？
はい。「プロンプトパイプラインの監視とアラート」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Prompt Engineeringコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Prompt Engineeringコースには全4レッスンが含まれています。
「プロンプトパイプラインの監視とアラート」で何を学びますか？
本番環境のプロンプト向けに、ダッシュボード、異常検知、オンコールアラートを構築します。 ブラウザで直接実行するハンズオンコードでAI Prompt Engineeringを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Prompt Engineeringを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Prompt Engineeringは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「プロンプトパイプラインの監視とアラート」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Prompt Engineeringレッスンでコードを書いて実行できますか？
はい。すべてのAI Prompt Engineeringレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- プロンプトのキャッシュ戦略
- バッチ処理と非同期実行
- モデル間の負荷分散
- プロンプトパイプラインの監視とアラート