本课学习可观测性(Observability)——生产环境LLM应用的眼睛。没有可观测性，线上问题就是黑盒：不知道哪里慢、哪里错、花了多少钱。可观测性三支柱：日志(Logs) + 指标(Metrics) + 追踪(Traces)。
第31课: 可观测性
├── 三支柱
│ ├── Logs: 结构化日志
│ ├── Metrics: 指标聚合
│ └── Traces: 分布式追踪
├── LLMObservability
│ ├── LLM调用记录
│ ├── Token用量追踪
│ ├── 延迟分析
│ └── 成本估算
└── 告警系统
├── 错误率告警
├── 延迟告警
└── 成本告警
# 三支柱: Logs/Metrics/Traces
import json
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class LogEntry:
timestamp: float
level: str # INFO/WARN/ERROR
component: str
message: str
metadata: dict = field(default_factory=dict)
@dataclass
class MetricPoint:
timestamp: float
name: str
value: float
tags: Dict[str, str] = field(default_factory=dict)
@dataclass
class Span:
"""分布式追踪的一个跨度"""
trace_id: str
span_id: str
parent_span_id: Optional[str] = None
operation: str = ""
start_time: float = 0.0
end_time: float = 0.0
status: str = "ok" # ok/error
attributes: Dict[str, Any] = field(default_factory=dict)
class LoggingSystem:
"""结构化日志系统"""
def __init__(self): self.entries: List[LogEntry] = []
def log(self, level, component, message, **meta):
self.entries.append(LogEntry(timestamp=time.time(), level=level, component=component, message=message, metadata=meta))
def query(self, level=None, component=None, since=None):
results = self.entries
if level: results = [e for e in results if e.level == level]
if component: results = [e for e in results if e.component == component]
if since: results = [e for e in results if e.timestamp >= since]
return results
class MetricsSystem:
"""指标系统"""
def __init__(self): self.metrics: List[MetricPoint] = []
def record(self, name, value, **tags):
self.metrics.append(MetricPoint(timestamp=time.time(), name=name, value=value, tags=tags))
def aggregate(self, name, window_seconds=60):
now = time.time()
points = [m for m in self.metrics if m.name == name and now - m.timestamp < window_seconds]
if not points: return {}
values = [p.value for p in points]
return {"count": len(values), "avg": sum(values)/len(values), "min": min(values), "max": max(values), "p99": sorted(values)[int(len(values)*0.99)]}
class TracingSystem:
"""分布式追踪系统"""
def __init__(self): self.traces: Dict[str, List[Span]] = defaultdict(list)
def start_span(self, trace_id, span_id, operation, parent=None, **attrs):
span = Span(trace_id=trace_id, span_id=span_id, parent_span_id=parent, operation=operation, start_time=time.time(), attributes=attrs)
self.traces[trace_id].append(span)
return span
def end_span(self, trace_id, span_id, status="ok"):
for span in self.traces.get(trace_id, []):
if span.span_id == span_id:
span.end_time = time.time()
span.status = status
break
def get_trace(self, trace_id):
return self.traces.get(trace_id, [])✅ 验证通过：三支柱系统完整实现——LoggingSystem/MetricsSystem/TracingSystem
# LLMObservability: LLM应用专用可观测性
from openai import OpenAI
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
client = OpenAI()
@dataclass
class LLMCallRecord:
"""LLM调用记录"""
call_id: str
model: str
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
latency_ms: float = 0.0
status: str = "success" # success/error/timeout
error: Optional[str] = None
timestamp: float = field(default_factory=time.time)
class LLMObservability:
"""LLM应用可观测性 - 专用的指标/追踪/告警"""
def __init__(self):
self.call_records: List[LLMCallRecord] = []
self.alerts: List[dict] = []
def record_call(self, call_id, model, usage, latency_ms, status="success", error=None):
record = LLMCallRecord(
call_id=call_id, model=model,
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
total_tokens=usage.get("total_tokens", 0),
latency_ms=latency_ms, status=status, error=error
)
self.call_records.append(record)
self._check_alerts(record)
return record
def _check_alerts(self, record):
"""检查是否触发告警"""
if record.latency_ms > 10000:
self.alerts.append({"type": "high_latency", "call_id": record.call_id, "value": record.latency_ms, "threshold": 10000})
if record.status == "error":
self.alerts.append({"type": "call_error", "call_id": record.call_id, "error": record.error})
if record.total_tokens > 4000:
self.alerts.append({"type": "high_token_usage", "call_id": record.call_id, "value": record.total_tokens})
def get_metrics(self, window_minutes=60) -> dict:
"""获取聚合指标"""
now = time.time()
recent = [r for r in self.call_records if now - r.timestamp < window_minutes * 60]
if not recent: return {"total_calls": 0}
latencies = [r.latency_ms for r in recent if r.status == "success"]
return {
"total_calls": len(recent),
"success_rate": sum(1 for r in recent if r.status == "success") / len(recent) * 100,
"avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
"p99_latency_ms": sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0,
"total_tokens": sum(r.total_tokens for r in recent),
"avg_tokens_per_call": sum(r.total_tokens for r in recent) / len(recent),
}
def get_cost_estimate(self, price_per_1k_input=0.00015, price_per_1k_output=0.0006):
"""估算成本"""
total_input = sum(r.prompt_tokens for r in self.call_records)
total_output = sum(r.completion_tokens for r in self.call_records)
return {
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"estimated_cost_usd": total_input / 1000 * price_per_1k_input + total_output / 1000 * price_per_1k_output,
}✅ 验证通过：LLMObservability实现了LLM调用记录、指标聚合和成本估算
# 告警系统: 多级告警与通知
from typing import Callable, Dict, List
from dataclasses import dataclass, field
from enum import Enum
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class AlertRule:
name: str
condition: Callable
severity: AlertSeverity
message_template: str
cooldown_seconds: int = 300
@dataclass
class Alert:
rule_name: str
severity: AlertSeverity
message: str
timestamp: float = field(default_factory=__import__("time").time)
class AlertSystem:
"""多级告警系统"""
def __init__(self):
self.rules: List[AlertRule] = []
self.alerts: List[Alert] = []
self.last_fired: Dict[str, float] = {}
def add_rule(self, name, condition, severity, message_template, cooldown=300):
self.rules.append(AlertRule(name=name, condition=condition, severity=severity, message_template=message_template, cooldown_seconds=cooldown))
def evaluate(self, metrics: dict):
"""评估所有规则"""
import time
now = time.time()
for rule in self.rules:
if rule.name in self.last_fired and now - self.last_fired[rule.name] < rule.cooldown_seconds:
continue
if rule.condition(metrics):
alert = Alert(rule_name=rule.name, severity=rule.severity, message=rule.message_template.format(**metrics))
self.alerts.append(alert)
self.last_fired[rule.name] = now
print(f"[{alert.severity.value.upper()}] {alert.message}")
# 预定义规则
alert_system = AlertSystem()
alert_system.add_rule("high_error_rate", lambda m: m.get("success_rate", 100) < 95, AlertSeverity.CRITICAL, "错误率过高: {success_rate:.1f}%")
alert_system.add_rule("high_latency", lambda m: m.get("p99_latency_ms", 0) > 5000, AlertSeverity.WARNING, "P99延迟过高: {p99_latency_ms:.0f}ms")
alert_system.add_rule("high_cost", lambda m: m.get("estimated_cost_usd", 0) > 10, AlertSeverity.WARNING, "成本过高: ${estimated_cost_usd:.2f}")✅ 验证通过：告警系统实现了多级规则评估、冷却时间和自动通知
| 支柱 | 回答什么 | 成本 | 用途 |
|---|---|---|---|
| Logs | 发生了什么 | 高(存储) | 审计/调试 |
| Metrics | 什么趋势 | 低(聚合) | 监控/告警 |
| Traces | 请求经过了哪里 | 中 | 性能分析 |
# 挑战: 构建LLM应用监控Dashboard
# - 实时Token用量
# - 错误率趋势
# - P50/P95/P99延迟
# - 成本追踪实现异常检测——自动发现指标异常并告警
可观测性不是"记录一切"，而是在正确的位置记录正确的信息。以下是生产环境的实战经验：
不是所有指标都值得监控。以下是最关键的LLM应用指标：
| 指标类别 | 具体指标 | 告警阈值 | 监控频率 |
|---|---|---|---|
| 性能 | TTFT(首字延迟) | > 2s | 实时 |
| 性能 | 总延迟P99 | > 10s | 实时 |
| 质量 | 成功率 | < 95% | 5分钟 |
| 质量 | 用户满意度 | < 3.5/5 | 每小时 |
| 成本 | 每请求Token | > 5000 | 实时 |
| 成本 | 日均成本 | > 预算120% | 每天 |
| 安全 | 注入尝试次数 | > 10/min | 实时 |