운영 환경에서 프롬프트 성능 모니터링

프롬프트 성능을 모니터링하는 이유

프롬프트 버전별 핵심 지표

계측: 지표 기록

import time
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def instrumented_call(prompt_id, version, messages, model):
    start = time.time()
    error = False
    response = None
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages
        )
    except Exception as e:
        error = True
        raise
    finally:
        latency_ms = (time.time() - start) * 1000
        usage = response.usage if response else None
        record_metric({
            'prompt_id': prompt_id,
            'version': version,
            'latency_ms': latency_ms,
            'input_tokens': usage.prompt_tokens if usage else 0,
            'output_tokens': usage.completion_tokens if usage else 0,
            'error': error,
            'timestamp': time.time()
        })
    return response

호출당 비용 계산

# pricing.py — model pricing table (per 1M tokens)
PRICING = {
    'gpt-4o': {'input': 2.50, 'output': 10.00},
    'gpt-4o-mini': {'input': 0.15, 'output': 0.60},
    'claude-3-5-sonnet-20241022': {'input': 3.00, 'output': 15.00},
}

def cost_usd(model, input_tokens, output_tokens):
    p = PRICING.get(model)
    if not p:
        return None
    cost = (input_tokens / 1_000_000) * p['input'] \
         + (output_tokens / 1_000_000) * p['output']
    return round(cost, 6)

# Example
c = cost_usd('gpt-4o-mini', input_tokens=800, output_tokens=200)
print(f'Cost per call: ${c}')   # Cost per call: $0.000240

# Daily cost estimate
calls_per_day = 50_000
print(f'Daily cost: ${round(c * calls_per_day, 2)}')  # $12.00

자동 품질 점수 산정

import random

JUDGE_SAMPLE_RATE = 0.05  # score 5% of calls

JUDGE_PROMPT = '''Rate the quality of this AI response on a scale of 1-5.
1=Very poor, 3=Acceptable, 5=Excellent.
Return only the integer score.

User input: {input}
AI response: {output}'''

def maybe_score_quality(user_input, ai_output, model='gpt-4o-mini'):
    if random.random() > JUDGE_SAMPLE_RATE:
        return None  # not sampled

    judge_messages = [{'role': 'user', 'content':
        JUDGE_PROMPT.format(input=user_input, output=ai_output)}]
    response = client.chat.completions.create(
        model=model, messages=judge_messages, max_tokens=5
    )
    try:
        score = int(response.choices[0].message.content.strip())
        return max(1, min(5, score))  # clamp to 1-5
    except ValueError:
        return None

시계열 데이터베이스에 지표 저장

-- TimescaleDB / PostgreSQL schema for prompt metrics
CREATE TABLE prompt_metrics (
  ts              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  prompt_id       VARCHAR(100),
  version         VARCHAR(20),
  model           VARCHAR(50),
  latency_ms      FLOAT,
  input_tokens    INT,
  output_tokens   INT,
  cost_usd        FLOAT,
  quality_score   FLOAT,   -- NULL if not sampled
  error           BOOLEAN DEFAULT FALSE
);

-- Convert to hypertable (TimescaleDB)
SELECT create_hypertable('prompt_metrics', 'ts');

-- Query: hourly P95 latency by version
SELECT
  date_trunc('hour', ts) AS hour,
  version,
  PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) AS p95_latency
FROM prompt_metrics
WHERE prompt_id = 'summarize-article'
  AND ts > NOW() - INTERVAL '24 hours'
GROUP BY 1, 2
ORDER BY 1;

지표 대시보드 구축

# Simple dashboard query aggregation in Python
def get_dashboard_stats(conn, prompt_id, hours=24):
    with conn.cursor() as cur:
        cur.execute('''
            SELECT
              version,
              COUNT(*) AS calls,
              AVG(latency_ms) AS avg_latency,
              PERCENTILE_CONT(0.95)
                WITHIN GROUP (ORDER BY latency_ms) AS p95_latency,
              AVG(cost_usd) AS avg_cost,
              AVG(quality_score) FILTER (WHERE quality_score IS NOT NULL) AS avg_quality,
              SUM(error::int)::float / COUNT(*) AS error_rate
            FROM prompt_metrics
            WHERE prompt_id = %s
              AND ts > NOW() - INTERVAL %s
            GROUP BY version
            ORDER BY MAX(ts) DESC
        ''', (prompt_id, f'{hours} hours'))
        return cur.fetchall()

프롬프트 성능 저하를 위한 이상 감지

import statistics

def detect_anomaly(recent_values, baseline_values, threshold_std=2.0):
    if len(baseline_values) < 10:
        return False  # not enough data
    baseline_mean = statistics.mean(baseline_values)
    baseline_std = statistics.stdev(baseline_values)
    recent_mean = statistics.mean(recent_values)

    if baseline_std == 0:
        return False
    z_score = abs(recent_mean - baseline_mean) / baseline_std
    return z_score > threshold_std

# Example: detect quality regression
baseline_quality = [4.1, 4.0, 4.2, 4.1, 4.0, 3.9, 4.1, 4.2, 4.0, 4.1]
recent_quality   = [3.2, 3.1, 3.4, 3.0]

if detect_anomaly(recent_quality, baseline_quality):
    print('ALERT: Quality score anomaly detected!')
    # fire_pagerduty_alert(...)
else:
    print('Quality within normal range')

경고 규칙과 임계값

# alerts.yaml (Grafana alerting or custom)
alerts:
  - name: HighErrorRate
    condition: error_rate > 0.05
    for: 5m
    severity: critical
    message: 'Prompt {prompt_id} error rate {error_rate:.1%} exceeds 5%'

  - name: HighLatencyP95
    condition: p95_latency_ms > 10000
    for: 3m
    severity: warning
    message: 'P95 latency {p95_latency_ms}ms exceeds 10s for {prompt_id}'

  - name: CostSpike
    condition: daily_cost_usd > weekly_avg_daily_cost * 2
    for: 1h
    severity: warning
    message: 'Daily cost ${daily_cost_usd:.2f} is 2x the weekly average'

  - name: QualityRegression
    condition: rolling_quality_avg < baseline_quality_avg * 0.80
    for: 15m
    severity: critical
    message: 'Quality dropped {pct_drop:.0%} for prompt {prompt_id}'

사용자 만족도 신호

# feedback_collector.py
def record_feedback(prompt_id, version, call_id, signal_type, value):
    '''
    signal_type: 'thumbs' | 'retry' | 'copy' | 'edit'
    value: for thumbs: 1 (up) or -1 (down); others: 1 (occurred)
    '''
    db.execute(
        'INSERT INTO prompt_feedback '
        '(prompt_id, version, call_id, signal_type, value, ts) '
        'VALUES (%s, %s, %s, %s, %s, NOW())',
        (prompt_id, version, call_id, signal_type, value)
    )

# Aggregate satisfaction score
def satisfaction_score(prompt_id, version):
    rows = db.fetch(
        'SELECT signal_type, AVG(value) as avg_val FROM prompt_feedback '
        'WHERE prompt_id=%s AND version=%s GROUP BY signal_type',
        (prompt_id, version)
    )
    return {r['signal_type']: round(r['avg_val'], 3) for r in rows}

버전 비교 보고서

def version_comparison_report(prompt_id, version_a, version_b, hours=48):
    stats_a = get_dashboard_stats_for_version(prompt_id, version_a, hours)
    stats_b = get_dashboard_stats_for_version(prompt_id, version_b, hours)

    print(f'=== Comparison: {version_a} vs {version_b} ===')
    metrics = ['avg_latency', 'p95_latency', 'avg_cost', 'avg_quality', 'error_rate']
    for m in metrics:
        va = stats_a.get(m, 0)
        vb = stats_b.get(m, 0)
        delta = vb - va
        pct = (delta / va * 100) if va else 0
        direction = '+' if delta > 0 else ''
        print(f'{m:20s}: {va:.4f} -> {vb:.4f}  ({direction}{pct:.1f}%)')

# Example output:
# avg_latency         : 1234.5000 -> 1189.2000  (-3.7%)
# p95_latency         : 3210.0000 -> 3050.0000  (-5.0%)
# avg_cost            : 0.000240 -> 0.000255  (+6.2%)
# avg_quality         : 4.1000 -> 4.3000  (+4.9%)
# error_rate          : 0.0120 -> 0.0080  (-33.3%)

빠른 확인

모니터링 요약