Pemantauan dan Makluman untuk Pipeline Gesaan
Papan pemuka, pengesanan anomali dan makluman petugas untuk gesaan pengeluaran.
Pemantauan dan Makluman untuk Pipeline Gesaan ialah pelajaran Kejuruteraan Arahan AI percuma di CoddyKit. Ini ialah pelajaran 4 daripada 4. Sebanyak 3 pelajaran dalam laluan pembelajaran ini boleh dibaca sepenuhnya secara percuma — selepas itu, CoddyKit PRO membuka akses kepada semua pelajaran, serta latihan praktikal dengan penyunting kod terbina dalam dan tutor kecerdasan buatan yang tersedia 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Kejuruteraan Arahan AI, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Kejuruteraan Arahan AI merangkumi sejumlah 4 pelajaran.
Saluran Paip Gesaan Produksi Memerlukan Pemantauan
Saluran paip gesaan dalam produksi ialah infrastruktur — ia memerlukan papan pemuka, amaran dan panduan operasi sama seperti perkhidmatan lain. Tanpa pemantauan, lonjakan kos, kemerosotan kualiti dan peningkatan mendadak kependaman tidak akan disedari sehingga pengguna mengadu atau bil tiba.
Metrik Teras: Persentil Kependaman
Pantau kependaman pada P50, P95 dan P99. Purata menyembunyikan tingkah laku hujung — P99 selama 30 saat bermaksud 1% pengguna menunggu setengah minit, walaupun P50 ialah 2 saat. Kependaman LLM sememangnya berubah-ubah kerana meningkat mengikut panjang output.
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())Metrik Papan Pemuka Kos Harian
Pantau kos API harian sebagai metrik utama papan pemuka. Kirakannya daripada penggunaan token secara masa nyata dan bandingkan dengan purata mingguan bergerak untuk mengesan lonjakan kos lebih awal.
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())Pemantauan Kadar Ralat
Pantau kadar ralat sebagai peratusan daripada jumlah permintaan. Ralat merangkumi kegagalan API, tamat masa, output tidak sah yang gagal dihuraikan dan penolakan model. Asingkan jenis ralat untuk menghasilkan amaran yang boleh diambil tindakan.
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())Penjejakan Trend Skor Kualiti
Pantau skor kualiti sebagai siri masa bergerak supaya trend dapat dilihat. Kemerosotan kualiti secara beransur-ansur selama beberapa hari lebih sukar dikesan berbanding penurunan mendadak, tetapi boleh menjejaskan kepercayaan pengguna dengan sama teruk.
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())Reka Bentuk Panel Papan Pemuka
Papan pemuka pemantauan yang direka bentuk dengan baik mengumpulkan metrik ke dalam bahagian yang logik. Tentukan empat panel teras yang diperlukan oleh setiap papan pemuka saluran paip gesaan.
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}')Pelaksanaan Peraturan Amaran
Amaran dicetuskan apabila metrik melanggar ambang. Laksanakan amaran sebagai semakan berkala ringkas yang dijalankan mengikut jadual dan menghantar pemberitahuan kepada PagerDuty, Slack atau e-mel.
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 firedPenghantaran Pemberitahuan
Pemberitahuan amaran hendaklah dihalakan berdasarkan tahap keterukan: amaran CRITICAL menghubungi jurutera bertugas serta-merta; amaran WARNING dihantar ke Slack; amaran INFO direkodkan dalam fail log. Gunakan pemasa peningkatan untuk amaran kritikal yang belum diakui.
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')Struktur Panduan Operasi Bertugas
Setiap amaran hendaklah mempunyai panduan operasi yang sepadan dan menerangkan dengan tepat perkara yang perlu dilakukan oleh jurutera bertugas. Panduan operasi yang ditulis dengan baik mengurangkan MTTR (Purata Masa Penyelesaian) daripada beberapa jam kepada beberapa minit.
# 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], '...')Pengelogan Berstruktur untuk Saluran Paip Gesaan
Log berstruktur (JSON bagi setiap baris) membolehkan penapisan dan pengagregatan yang berkuasa dalam sistem pengurusan log seperti Datadog, Splunk atau CloudWatch. Setiap panggilan LLM hendaklah menghasilkan satu entri log berstruktur.
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)Seni Bina Sistem Pemantauan
Satukan semua komponen pemantauan menjadi sistem padu yang berjalan bersama saluran paip gesaan. Gelung tinjauan ringkas mengendalikan penilaian dan penghantaran amaran.
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.')Semakan Ringkas
Kependaman P50 saluran paip gesaan anda ialah 1.5 saat tetapi P99 ialah 28 saat. Apakah yang diberitahu oleh keadaan ini tentang tingkah laku dalam produksi?
Ringkasan Pemantauan dan Amaran
Pemantauan saluran paip gesaan produksi memerlukan lima kategori pengukuran serta peraturan amaran yang sepadan:
- Kependaman: pantau P50/P95/P99 — beri amaran apabila P95 > 10s
- Kos: kos harian berbanding purata mingguan — beri amaran apabila berlaku lonjakan kos 2x
- Kadar ralat: jumlah % dan pecahan jenis ralat — beri amaran apabila > 5%
- Skor kualiti: trend purata bergerak — beri amaran apabila jatuh di bawah 3.5/5
- Amaran: CRITICAL → PagerDuty + Slack; WARNING → Slack sahaja
- Panduan operasi: panduan penyelesaian langkah demi langkah untuk setiap jenis amaran
- Papan pemuka: empat panel yang meliputi ketersediaan, prestasi, kos dan kualiti
Pelajari Kejuruteraan Arahan AI dengan tutor kecerdasan buatan — percuma
Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.
- Kursus
- 53
- Pelajaran
- 199
Soalan Lazim
Adakah pelajaran “Pemantauan dan Makluman untuk Pipeline Gesaan” percuma?
Ya — sebanyak 3 pelajaran dalam laluan pembelajaran Kejuruteraan Arahan AI, termasuk “Pemantauan dan Makluman untuk Pipeline Gesaan”, boleh dibaca sepenuhnya secara percuma di web ini. Selepas itu, CoddyKit PRO membuka akses kepada semua pelajaran, serta latihan interaktif dengan penyunting kod terbina dalam dan tutor kecerdasan buatan yang tersedia 24/7. Kursus Kejuruteraan Arahan AI merangkumi sejumlah 4 pelajaran.
Apakah yang akan saya pelajari dalam “Pemantauan dan Makluman untuk Pipeline Gesaan”?
Papan pemuka, pengesanan anomali dan makluman petugas untuk gesaan pengeluaran. Anda berlatih Kejuruteraan Arahan AI menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.
Adakah saya memerlukan pengalaman untuk memulakan Kejuruteraan Arahan AI?
Tiada pengalaman terdahulu diperlukan. Pembelajaran Kejuruteraan Arahan AI di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 4 daripada 4.
Berapa lamakah pelajaran “Pemantauan dan Makluman untuk Pipeline Gesaan” diambil?
Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.
Bolehkah saya menulis dan menjalankan kod dalam pelajaran Kejuruteraan Arahan AI ini?
Ya. Setiap pelajaran Kejuruteraan Arahan AI menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.
Semua pelajaran dalam kursus ini
- Strategi Penyimpanan Cache untuk Gesaan
- Pemprosesan Kelompok dan Pelaksanaan Tak Segerak
- Pengimbangan Beban Merentas Model
- Pemantauan dan Makluman untuk Pipeline Gesaan