프롬프트 파이프라인 모니터링 및 알림
운영 환경의 프롬프트를 위한 대시보드, 이상 탐지, 당직자 알림을 다룹니다.
레슨 4/413개 단계
프롬프트 파이프라인 모니터링 및 알림은(는) CoddyKit의 무료 AI Prompt Engineering 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Prompt Engineering 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Prompt Engineering 강의에는 총 4개의 강의가 포함되어 있습니다.
프로덕션 프롬프트 파이프라인에는 모니터링이 필요합니다
프로덕션의 프롬프트 파이프라인은 인프라의 일부이므로 다른 서비스와 마찬가지로 대시보드, 경고, 운영 절차서가 필요합니다. 모니터링이 없으면 비용 급증, 품질 저하, 지연 시간 급증을 사용자가 불만을 제기하거나 청구서가 도착할 때까지 알아차리지 못합니다.
핵심 지표: 지연 시간 백분위수
P50, P95, P99에서 지연 시간을 추적합니다. 평균만 보면 꼬리 구간의 동작이 가려집니다. P50이 2초이더라도 P99가 30초라면 사용자의 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())대시보드 패널 설계
잘 설계된 모니터링 대시보드는 지표를 논리적인 섹션으로 묶습니다. 모든 프롬프트 파이프라인 대시보드에 필요한 네 가지 핵심 패널을 정의합니다.
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 경고는 기록 파일에 남깁니다. 확인되지 않은 심각한 경고에는 단계적 에스컬레이션 타이머를 사용합니다.
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], '...')프롬프트 파이프라인을 위한 구조화된 기록
구조화된 기록(줄마다 하나의 JSON)을 사용하면 Datadog, Splunk, CloudWatch와 같은 기록 관리 시스템에서 강력한 필터링과 집계가 가능합니다. 모든 LLM 호출은 구조화된 기록 항목 하나를 생성해야 합니다.
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초입니다. 이것은 프로덕션 환경의 동작에 대해 무엇을 알려 주나요?
모니터링 및 경고 요약
프로덕션 프롬프트 파이프라인 모니터링에는 다섯 가지 측정 범주와 이에 맞는 경고 규칙이 필요합니다:
- 지연 시간: P50/P95/P99를 추적하고 P95 > 10s이면 경고합니다
- 비용: 일일 비용과 주간 평균을 비교하고 비용이 2배 급증하면 경고합니다
- 오류율: 전체 비율과 오류 유형별 분류를 추적하고 5%를 초과하면 경고합니다
- 품질 점수: 이동 평균 추세를 추적하고 3.5/5 아래로 떨어지면 경고합니다
- 경고: CRITICAL → PagerDuty + Slack, WARNING → Slack만 사용합니다
- 운영 절차서: 모든 경고 유형에 대해 단계별 해결 안내를 제공합니다
- 대시보드: 가용성, 성능, 비용, 품질을 다루는 네 개의 패널로 구성합니다
무료로 시작
AI 튜터와 함께 AI Prompt Engineering을(를) 배우세요 — 무료
브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.
- 코스
- 53
- 레슨
- 199
자주 묻는 질문
“프롬프트 파이프라인 모니터링 및 알림” 강의는 무료인가요?
네 — “프롬프트 파이프라인 모니터링 및 알림” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Prompt Engineering 강의 전체를 잠금 해제할 수 있습니다. AI Prompt Engineering 강의에는 총 4개의 강의가 포함되어 있습니다.
“프롬프트 파이프라인 모니터링 및 알림”에서 뭘 배우나요?
운영 환경의 프롬프트를 위한 대시보드, 이상 탐지, 당직자 알림을 다룹니다. 브라우저에서 직접 실행하는 실습 코드로 AI Prompt Engineering을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Prompt Engineering을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Prompt Engineering은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.
“프롬프트 파이프라인 모니터링 및 알림” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Prompt Engineering 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Prompt Engineering 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 프롬프트 캐싱 전략
- 일괄 처리 및 비동기 실행
- 모델 간 부하 분산
- 프롬프트 파이프라인 모니터링 및 알림