การตรวจติดตามและแจ้งเตือนสำหรับไปป์ไลน์พรอมต์

กระบวนการพรอมต์ในระบบจริงจำเป็นต้องมีการเฝ้าติดตาม

ตัวชี้วัดหลัก: เปอร์เซ็นไทล์ของเวลาแฝง

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())

ตัวชี้วัดค่าใช้จ่ายรายวันบนแผงควบคุม

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())

การเฝ้าติดตามอัตราข้อผิดพลาด

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())

การออกแบบส่วนแสดงผลบนแผงควบคุม

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}')

การนำกฎการแจ้งเตือนไปใช้

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

การส่งการแจ้งเตือน

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')

โครงสร้างคู่มือปฏิบัติการสำหรับวิศวกรเวร

# 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], '...')

การบันทึกแบบมีโครงสร้างสำหรับกระบวนการพรอมต์

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.')

ตรวจสอบความเข้าใจ

สรุปการเฝ้าติดตามและการแจ้งเตือน