Surveillance et alertes pour les pipelines de prompts
Tableaux de bord, détection d’anomalies et alertes d’astreinte pour les prompts en production.
Surveillance et alertes pour les pipelines de prompts est une leçon AI Prompt Engineering gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Prompt Engineering, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Prompt Engineering comprend 4 leçons au total.
Les chaînes de traitement des instructions en production nécessitent une surveillance
Une chaîne de traitement des instructions en production est une infrastructure : elle nécessite des tableaux de bord, des alertes et des guides opérationnels, comme tout autre service. Sans surveillance, les pics de coûts, les régressions de qualité et les fortes hausses de latence passent inaperçus jusqu'à ce que les utilisateurs se plaignent ou que les factures arrivent.
Métriques essentielles : percentiles de latence
Suivez la latence aux niveaux P50, P95 et P99. La moyenne masque les comportements extrêmes : un P99 de 30 secondes signifie que 1 % des utilisateurs attendent une demi-minute, même si le P50 est de 2 secondes. La latence des LLM est intrinsèquement variable, car elle augmente avec la longueur de la sortie.
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())Coût quotidien : métrique du tableau de bord
Suivez le coût quotidien de l'API comme métrique principale du tableau de bord. Calculez-le en temps réel à partir de l'utilisation des jetons et comparez-le à la moyenne quotidienne glissante de la semaine afin de détecter rapidement les pics de coûts.
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())Surveillance du taux d'erreur
Suivez le taux d'erreur en pourcentage du nombre total de requêtes. Les erreurs comprennent les échecs de l'API, les délais d'expiration, les sorties malformées qui échouent lors de l'analyse et les refus du modèle. Distinguez les types d'erreurs pour permettre des alertes exploitables.
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())Suivi de la tendance du score de qualité
Suivez le score de qualité sous la forme d'une série chronologique glissante afin de rendre les tendances visibles. Une baisse progressive de la qualité sur plusieurs jours est plus difficile à remarquer qu'une chute brutale, mais peut être tout aussi préjudiciable à la confiance des utilisateurs.
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())Conception des panneaux du tableau de bord
Un tableau de bord de surveillance bien conçu regroupe les métriques en sections logiques. Définissez les quatre panneaux essentiels dont tout tableau de bord de chaîne de traitement des instructions a besoin.
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}')Mise en œuvre des règles d'alerte
Les alertes se déclenchent lorsque les métriques dépassent les seuils. Mettez en œuvre les alertes sous forme de simples vérifications périodiques exécutées selon un calendrier, qui envoient des notifications à PagerDuty, Slack ou par e-mail.
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 firedDiffusion des notifications
Les notifications d'alerte doivent être acheminées selon leur gravité : les alertes CRITICAL alertent immédiatement l'astreinte ; les alertes WARNING sont publiées dans Slack ; les alertes INFO sont envoyées dans un fichier journal. Utilisez des minuteurs d'escalade pour les alertes critiques qui ne sont pas reconnues.
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')Structure des guides d'astreinte
Chaque alerte doit avoir un guide opérationnel correspondant, qui indique précisément à l'ingénieur d'astreinte quoi faire. Des guides bien rédigés réduisent le MTTR (temps moyen de résolution) de plusieurs heures à quelques minutes.
# 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], '...')Journalisation structurée des chaînes de traitement des instructions
Les journaux structurés, au format JSON par ligne, permettent un filtrage et une agrégation puissants dans des systèmes de gestion des journaux tels que Datadog, Splunk ou CloudWatch. Chaque appel de LLM doit produire une entrée de journal structurée.
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)Architecture du système de surveillance
Assemblez tous les composants de surveillance en un système cohérent qui s'exécute aux côtés de la chaîne de traitement des instructions. Une simple boucle d'interrogation suffit à évaluer et à diffuser les alertes.
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.')Vérification rapide
La latence P50 de votre chaîne de traitement des instructions est de 1,5 seconde, mais sa latence P99 est de 28 secondes. Que vous apprend cette différence sur le comportement en production ?
Résumé de la surveillance et des alertes
La surveillance d'une chaîne de traitement des instructions en production nécessite cinq catégories de mesures et les règles d'alerte correspondantes :
- Latence : suivre P50/P95/P99 et déclencher une alerte lorsque P95 > 10 s
- Coût : coût quotidien par rapport à la moyenne hebdomadaire et alerte en cas de pic de coûts égal à 2 fois la moyenne
- Taux d'erreur : pourcentage total et répartition par type d'erreur, avec alerte lorsque le taux dépasse 5 %
- Score de qualité : tendance de la moyenne glissante et alerte lorsqu'elle passe sous 3,5/5
- Alertes : CRITICAL → PagerDuty + Slack ; WARNING → Slack uniquement
- Guides opérationnels : guides de résolution étape par étape pour chaque type d'alerte
- Tableau de bord : quatre panneaux couvrant la disponibilité, les performances, les coûts et la qualité
Questions Fréquemment Posées
La leçon « Surveillance et alertes pour les pipelines de prompts » est-elle gratuite ?
Oui — le texte complet de « Surveillance et alertes pour les pipelines de prompts » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Prompt Engineering, passe à CoddyKit PRO. Le cours AI Prompt Engineering comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Surveillance et alertes pour les pipelines de prompts » ?
Tableaux de bord, détection d’anomalies et alertes d’astreinte pour les prompts en production. Tu pratiques AI Prompt Engineering avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Prompt Engineering ?
Aucune expérience préalable n'est requise. AI Prompt Engineering sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Surveillance et alertes pour les pipelines de prompts » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Prompt Engineering ?
Oui. Chaque leçon AI Prompt Engineering inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Stratégies de mise en cache des prompts
- Traitement par lots et exécution asynchrone
- Équilibrage de charge entre les modèles
- Surveillance et alertes pour les pipelines de prompts