针对延迟、成本和质量下降设置告警
为 p99 延迟、单次请求成本和自动质量分数定义告警阈值，并在 LLM 流程性能下降时将告警发送到 Slack 或 PagerDuty。
针对延迟、成本和质量下降设置告警 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
为什么 LLM 系统需要告警
LLM 应用可能以隐蔽且渐进的方式发生故障。提示词的修改可能使平均延迟增加 30%，检索功能的升级可能导致回答质量略有下降，或者使用量激增可能使每日成本达到预算的 5 倍。如果没有主动告警，您只能等到用户投诉或收到月度账单时才会发现这些问题。告警可以将被动救火转变为主动运维。
三类告警
LLM 应用告警分为三类。延迟告警会在响应时间超过用户体验阈值时触发（例如 p99 > 10 秒）。成本告警会在单次请求成本或每日支出超过预算阈值时触发，从而避免账单带来的冲击。质量告警会在自动质量指标（LLM 作为评审员的评分、用户满意度比率、忠实性评分）低于可接受下限时触发。每一类告警都需要不同的监测方式和告警渠道。
from dataclasses import dataclass
@dataclass
class AlertThresholds:
# Latency (milliseconds)
p50_latency_ms: int = 2000 # median should be under 2s
p99_latency_ms: int = 10000 # 99th percentile under 10s
# Cost (USD)
max_cost_per_request: float = 0.05 # alert if one request costs > 5 cents
max_daily_spend: float = 50.00 # alert if daily spend exceeds $50
# Quality (0.0 to 1.0 scale)
min_quality_score: float = 0.75 # alert if rolling avg drops below 75%
min_faithfulness: float = 0.80 # alert if RAGAS faithfulness drops below 80%
DEFAULT_THRESHOLDS = AlertThresholds()收集延迟指标
要针对延迟设置告警，首先需要持续一致地收集延迟数据。请分别测量三个延迟组成部分：首个令牌时间（TTFT，对流式用户体验至关重要）、总响应时间，以及检索和生成各步骤的延迟。将这些数据存储为时间序列数据（每个请求对应一个数据点），这样您就可以计算过去 N 分钟或数小时的百分位数和滚动平均值。
import time
from collections import deque
from statistics import quantiles
class LatencyTracker:
def __init__(self, window_size=100):
self.window = deque(maxlen=window_size) # rolling window of latencies
def record(self, latency_ms: float):
self.window.append(latency_ms)
def p50(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[49]
def p99(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[98]
def check_alert(self, thresholds: AlertThresholds) -> list[str]:
alerts = []
if self.p99() > thresholds.p99_latency_ms:
alerts.append(f'p99 latency {self.p99():.0f}ms exceeds {thresholds.p99_latency_ms}ms')
if self.p50() > thresholds.p50_latency_ms:
alerts.append(f'p50 latency {self.p50():.0f}ms exceeds {thresholds.p50_latency_ms}ms')
return alerts
latency_tracker = LatencyTracker()跟踪成本并设置告警
成本监控需要按多个粒度跟踪支出：单次请求成本（用于发现成本异常高的请求）、每小时和每日总额（用于发现使用量激增），以及按功能统计的成本（用于确定应用中成本最高的部分）。请立即针对单次请求成本异常发出告警，并使用计划任务检查每日预算上限。
from datetime import datetime, date
import threading
class CostTracker:
def __init__(self):
self._lock = threading.Lock()
self._daily_spend = {} # date -> total USD
self._per_request = [] # list of (timestamp, cost)
def record(self, cost_usd: float) -> list[str]:
today = date.today().isoformat()
alerts = []
with self._lock:
# Track daily spend
self._daily_spend[today] = self._daily_spend.get(today, 0) + cost_usd
self._per_request.append((datetime.now(), cost_usd))
# Alert on expensive single requests
if cost_usd > DEFAULT_THRESHOLDS.max_cost_per_request:
alerts.append(f'Expensive request: ${cost_usd:.4f} (threshold: ${DEFAULT_THRESHOLDS.max_cost_per_request})')
# Alert on daily budget exceeded
daily_total = self._daily_spend[today]
if daily_total > DEFAULT_THRESHOLDS.max_daily_spend:
alerts.append(f'Daily budget exceeded: ${daily_total:.2f} > ${DEFAULT_THRESHOLDS.max_daily_spend}')
return alerts
cost_tracker = CostTracker()质量评分告警
质量告警是最复杂的一类，因为仅凭基础设施指标无法衡量质量——这需要语义评估。最具扩展性的方式是进行抽样自动评估：从生产请求中随机抽取样本（5% 至 10%），异步运行 LLM 作为评审员的评估器，存储评分并计算滚动平均值。当滚动平均值低于质量下限时触发告警。
import random
from openai import OpenAI
client = OpenAI()
def evaluate_response_quality(question: str, answer: str) -> float:
judge_prompt = f'''Rate the quality of this AI assistant response on a scale from 0 to 1.
Question: {question}
Answer: {answer}
Return only a JSON object: {{"score": 0.85, "reason": "brief reason"}}
Scoring guide:
1.0 = Perfect, accurate, helpful
0.75 = Good, minor issues
0.5 = Acceptable but incomplete
0.25 = Poor, major gaps
0.0 = Completely wrong or harmful'''
response = client.chat.completions.create(
model='gpt-4o-mini', # cheaper model for evaluation
messages=[{'role': 'user', 'content': judge_prompt}],
response_format={'type': 'json_object'}
)
import json
result = json.loads(response.choices[0].message.content)
return float(result['score'])
def maybe_evaluate(question: str, answer: str, sample_rate=0.1):
if random.random() < sample_rate:
score = evaluate_response_quality(question, answer)
quality_tracker.record(score)
return score
return None使用滚动平均值检测趋势
单个糟糕的响应并不表示存在系统性问题。请在一个时间窗口内使用滚动平均值（例如最近 1 小时或最近 100 个请求）来检测趋势。滚动平均值如果越过阈值并持续 10 分钟以上，说明存在真实问题；而短暂的峰值可能只是噪声。指数加权移动平均值（EWMA）对近期变化的反应速度比简单滚动平均值更快。
class RollingQualityMonitor:
def __init__(self, window=50, alert_threshold=0.75, ewma_alpha=0.1):
self.scores = []
self.window = window
self.alert_threshold = alert_threshold
self.alpha = ewma_alpha # EWMA decay factor
self.ewma_score = None
def record(self, score: float):
self.scores.append(score)
if len(self.scores) > self.window:
self.scores.pop(0)
# Update exponentially weighted moving average
if self.ewma_score is None:
self.ewma_score = score
else:
self.ewma_score = self.alpha * score + (1 - self.alpha) * self.ewma_score
def should_alert(self) -> bool:
if len(self.scores) < 10: # not enough data yet
return False
return self.ewma_score < self.alert_threshold
def summary(self) -> dict:
if not self.scores:
return {}
return {
'rolling_avg': sum(self.scores) / len(self.scores),
'ewma': self.ewma_score,
'sample_count': len(self.scores),
'alert': self.should_alert()
}
quality_tracker = RollingQualityMonitor()告警路由：Slack 与 PagerDuty
只有通过正确的渠道将告警发送给正确的人，告警才有用。请按严重程度路由告警：质量下降和成本预算告警发送到团队在工作时间监控的 Slack 频道。超过关键阈值的延迟告警（例如 p99 > 30 秒）以及突发的成本峰值（例如 5 分钟内达到正常水平的 10 倍）发送到 PagerDuty，以便立即升级给值班人员。
import requests
def send_slack_alert(message: str, severity: str, webhook_url: str):
color = {'critical': '#ff0000', 'warning': '#ff9900', 'info': '#36a64f'}[severity]
payload = {
'attachments': [{
'color': color,
'title': f'LLM Alert [{severity.upper()}]',
'text': message,
'footer': 'AI Pipeline Monitor'
}]
}
requests.post(webhook_url, json=payload)
def send_pagerduty_alert(title: str, details: str, routing_key: str):
payload = {
'routing_key': routing_key,
'event_action': 'trigger',
'payload': {
'summary': title,
'severity': 'critical',
'source': 'ai-pipeline-monitor',
'custom_details': {'details': details}
}
}
requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)
def route_alert(alert_text: str, severity: str):
send_slack_alert(alert_text, severity, SLACK_WEBHOOK_URL)
if severity == 'critical':
send_pagerduty_alert(alert_text, alert_text, PAGERDUTY_ROUTING_KEY)告警评估循环
告警逻辑应在计划循环中运行，而不应与请求处理同步执行。异步运行告警检查可以避免它们增加生产请求的延迟。对于大多数告警需求，每 60 秒运行一次的后台线程或计划任务就足够了。在请求处理路径中收集指标，在后台循环中评估阈值。
import threading
import time
def alert_evaluation_loop(interval_seconds=60):
while True:
try:
# Check latency
latency_alerts = latency_tracker.check_alert(DEFAULT_THRESHOLDS)
for alert in latency_alerts:
route_alert(f'LATENCY: {alert}', 'warning')
# Check quality
quality_summary = quality_tracker.summary()
if quality_summary.get('alert'):
msg = f'QUALITY DEGRADATION: Rolling avg {quality_summary["ewma"]:.2f} below threshold {DEFAULT_THRESHOLDS.min_quality_score}'
route_alert(msg, 'warning')
print(f'Alert check complete. Quality: {quality_summary.get("ewma", "N/A")}')
except Exception as e:
print(f'Alert evaluation error: {e}')
time.sleep(interval_seconds)
# Start in background thread
alert_thread = threading.Thread(target=alert_evaluation_loop, daemon=True)
alert_thread.start()告警疲劳与阈值调整
当告警触发得过于频繁，以至于团队开始忽略它们时，就会出现告警疲劳。您可以通过以下方式避免这种情况：先从保守的（较高）阈值开始，了解基线后再逐步收紧；要求告警在多个评估周期内持续存在后才触发；将相关告警合并为一条通知；定期审查并停用那些从未促成有意义行动的告警。
class AlertDebouncer:
def __init__(self, required_consecutive_fires=3):
self.required = required_consecutive_fires
self.fire_counts = {} # alert_name -> consecutive fires
self.already_firing = set() # alert_names currently active
def should_fire(self, alert_name: str, condition: bool) -> bool:
if condition:
self.fire_counts[alert_name] = self.fire_counts.get(alert_name, 0) + 1
if self.fire_counts[alert_name] >= self.required and alert_name not in self.already_firing:
self.already_firing.add(alert_name)
return True # fire the alert (first time condition sustained)
else:
if self.fire_counts.get(alert_name, 0) > 0:
self.fire_counts[alert_name] = 0
self.already_firing.discard(alert_name) # condition cleared
return False # either not sustained or already firing (no duplicate alert)
debouncer = AlertDebouncer(required_consecutive_fires=3)使用统计方法监控成本异常
简单的阈值告警可能无法发现细微的成本异常，例如两天内成本逐渐增加 50%。请使用统计异常检测：计算每小时成本的滚动平均值和标准差，并在当前小时的成本高于平均值 N 个标准差时发出告警（Z 分数告警）。这种方法可以自动适应工作日与周末流量等自然使用模式。
import statistics
def compute_z_score(recent_value: float, historical_values: list[float]) -> float:
if len(historical_values) < 5:
return 0 # not enough data
mean = statistics.mean(historical_values)
stdev = statistics.stdev(historical_values)
if stdev == 0:
return 0
return (recent_value - mean) / stdev
def check_cost_anomaly(current_hour_cost: float, historical_hourly_costs: list[float]) -> str | None:
z = compute_z_score(current_hour_cost, historical_hourly_costs)
if z > 3.0: # more than 3 standard deviations above mean
mean = statistics.mean(historical_hourly_costs)
return f'Cost anomaly: ${current_hour_cost:.2f} this hour (normal: ${mean:.2f}, z={z:.1f})'
return None构建运维仪表板
告警是被动层，实时运维仪表板则是主动层。请构建一个简单的仪表板，显示实时指标：当前 p50/p99 延迟、每分钟请求数、当前质量评分（滚动 EWMA）、截至目前的每日成本与预算的对比，以及过去一小时内成本最高的 5 个请求。这样，团队无需等待告警触发，就能一目了然地了解系统健康状况。
快速检查
请通过本课内容检验您对 LLM 应用指标告警的理解。
课程回顾
在本课中，您学到了：三类告警涵盖 LLM 系统——延迟、成本和质量，每类告警都需要不同的监测；滚动平均值和 EWMA比针对单个请求进行阈值检查更擅长检测逐渐发生的质量下降；而且，告警去抖要求条件在多个评估周期内持续满足后才触发告警，从而避免告警疲劳。接下来，我们将深入学习提示词注入攻击和 AI 安全。
常见问题解答
「针对延迟、成本和质量下降设置告警」课时是免费的吗？
是的 — 「针对延迟、成本和质量下降设置告警」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「针对延迟、成本和质量下降设置告警」这节课中我会学到什么？
为 p99 延迟、单次请求成本和自动质量分数定义告警阈值，并在 LLM 流程性能下降时将告警发送到 Slack 或 PagerDuty。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「针对延迟、成本和质量下降设置告警」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- LLM 应用为何难以调试
- 使用 LangSmith 进行追踪
- 使用 Langfuse 实现与模型无关的可观测性
- 针对延迟、成本和质量下降设置告警