一、为什么需要 AI 可观测性

传统监控 vs AI 监控： 传统监控关心的问题： ┌──────────────────────────────────────────────┐ │ • QPS 多少？ │ │ • 延迟 P99 多少？ │ │ • 错误率多少？ │ │ • CPU/内存使用率？ │ └──────────────────────────────────────────────┘ AI 监控额外关心的问题： ┌──────────────────────────────────────────────┐ │ • LLM 返回了多少 Token？花了多少钱？ │ │ • 回答质量如何？有没有胡说八道？ │ │ • 用户满意度如何？ │ │ • 哪些 Prompt 效果最好？ │ │ • 模型有没有被注入攻击？ │ │ • 上下文中有没有泄露敏感信息？ │ └──────────────────────────────────────────────┘

AI 可观测性 = 传统监控 + LLM 专有指标 + 质量评估 + 安全审计

二、整体架构

┌──────────────────────────────────────────────────────────────┐ │ 应用层 │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ LLM 网关 │ │ RAG 引擎 │ │ FC 引擎 │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ ├───────┴──────────────┴──────────────┴────────────────────────┤ │ 埋点层 │ │ ┌──────────────────────────────────────────────┐ │ │ │ Span 采集器 │ Metric 采集器 │ Log 采集器 │ │ │ └──────────────────────┬───────────────────────┘ │ ├─────────────────────────┼───────────────────────────────────┤ │ ▼ │ │ ┌──────────────────────────────────────────────┐ │ │ │ 数据管道 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ │ │ Jaeger │ │ Prometheus│ │ Loki │ │ │ │ │ │ (Trace) │ │ (Metrics)│ │ (Logs) │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ └──────────────────────┬───────────────────────┘ │ ├─────────────────────────┼───────────────────────────────────┤ │ ▼ │ │ ┌──────────────────────────────────────────────┐ │ │ │ 可视化与告警 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ │ │ Grafana │ │ Alertmanager│ │ 看板 │ │ │ │ │ │ (仪表盘) │ │ (告警) │ │ (日报) │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ └──────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

三、链路追踪（Tracing）

// observability/trace.go package observability import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "log" "sync" "time" ) // TraceID 链路 ID type TraceID string // SpanID Span ID type SpanID string // SpanKind Span 类型 type SpanKind string const ( SpanKindInternal SpanKind = "internal" SpanKindClient SpanKind = "client" SpanKindServer SpanKind = "server" SpanKindProducer SpanKind = "producer" SpanKindConsumer SpanKind = "consumer" ) // Span 链路中的一个节点 type Span struct { TraceID TraceID `json:"trace_id"` SpanID SpanID `json:"span_id"` ParentSpan SpanID `json:"parent_span_id"` Name string `json:"name"` Kind SpanKind `json:"kind"` StartTime time.Time `json:"start_time"` EndTime time.Time `json:"end_time"` Duration time.Duration `json:"duration"` Status string `json:"status"` // ok / error Attributes map[string]interface{} `json:"attributes"` Events []SpanEvent `json:"events"` Resource map[string]string `json:"resource"` } // SpanEvent Span 内的事件 type SpanEvent struct { Timestamp time.Time `json:"timestamp"` Name string `json:"name"` Attributes map[string]interface{} `json:"attributes"` } // Tracer 链路追踪器 type Tracer struct { serviceName string exporter SpanExporter spans sync.Map } // SpanExporter Span 导出器接口 type SpanExporter interface { Export(spans []*Span) error Shutdown() error } // NewTracer 创建追踪器 func NewTracer(serviceName string, exporter SpanExporter) *Tracer { return &Tracer{ serviceName: serviceName, exporter: exporter, } } // StartSpan 开始一个 Span func (t *Tracer) StartSpan(ctx context.Context, name string, opts ...SpanOption) (context.Context, *Span) { options := &spanOptions{ kind: SpanKindInternal, } for _, opt := range opts { opt(options) } span := &Span{ TraceID: getOrCreateTraceID(ctx), SpanID: generateSpanID(), Name: name, Kind: options.kind, StartTime: time.Now(), Status: "ok", Attributes: make(map[string]interface{}), Resource: map[string]string{ "service": t.serviceName, }, } // 继承父 Span if parentSpan := ctx.Value(spanContextKey{}); parentSpan != nil { if ps, ok := parentSpan.(*Span); ok { span.ParentSpan = ps.SpanID } } t.spans.Store(span.SpanID, span) // 将当前 Span 放入 context ctx = context.WithValue(ctx, spanContextKey{}, span) ctx = context.WithValue(ctx, traceIDKey{}, span.TraceID) return ctx, span } // EndSpan 结束一个 Span func (t *Tracer) EndSpan(span *Span) { span.EndTime = time.Now() span.Duration = span.EndTime.Sub(span.StartTime) t.spans.Delete(span.SpanID) // 异步导出 go func() { if err := t.exporter.Export([]*Span{span}); err != nil { log.Printf("[Trace] Export error: %v", err) } }() } // AddEvent 添加事件 func (t *Tracer) AddEvent(span *Span, name string, attrs map[string]interface{}) { span.Events = append(span.Events, SpanEvent{ Timestamp: time.Now(), Name: name, Attributes: attrs, }) } // SetAttribute 设置属性 func (t *Tracer) SetAttribute(span *Span, key string, value interface{}) { span.Attributes[key] = value } // SetError 标记错误 func (t *Tracer) SetError(span *Span, err error) { span.Status = "error" span.Attributes["error"] = err.Error() } // ─── Span Option ─── type spanOptions struct { kind SpanKind } type SpanOption func(*spanOptions) func WithSpanKind(kind SpanKind) SpanOption { return func(o *spanOptions) { o.kind = kind } } // ─── Context Keys ─── type spanContextKey struct{} type traceIDKey struct{} func getOrCreateTraceID(ctx context.Context) TraceID { if tid, ok := ctx.Value(traceIDKey{}).(TraceID); ok { return tid } return generateTraceID() } // ─── ID 生成 ─── func generateTraceID() TraceID { b := make([]byte, 16) rand.Read(b) return TraceID(hex.EncodeToString(b)) } func generateSpanID() SpanID { b := make([]byte, 8) rand.Read(b) return SpanID(hex.EncodeToString(b)) } // ─── Console Exporter（开发用） ─── type ConsoleExporter struct{} func (e *ConsoleExporter) Export(spans []*Span) error { for _, span := range spans { data, _ := json.MarshalIndent(span, "", " ") log.Printf("[Trace] %s", string(data)) } return nil } func (e *ConsoleExporter) Shutdown() error { return nil } // ─── LLM Call Tracing ─── // LLMSpanAttributes LLM 调用的 Span 属性 type LLMSpanAttributes struct { Provider string `json:"llm.provider"` Model string `json:"llm.model"` RequestType string `json:"llm.request_type"` // chat / completion / embedding PromptTokens int `json:"llm.prompt_tokens"` CompletionsTokens int `json:"llm.completion_tokens"` TotalTokens int `json:"llm.total_tokens"` CostUSD float64 `json:"llm.cost_usd"` Temperature float64 `json:"llm.temperature"` MaxTokens int `json:"llm.max_tokens"` }

四、指标采集（Metrics）

// observability/metrics.go package observability import ( "sync" "sync/atomic" "time" ) // MetricType 指标类型 type MetricType string const ( MetricCounter MetricType = "counter" MetricGauge MetricType = "gauge" MetricHistogram MetricType = "histogram" ) // Metric 指标定义 type Metric struct { Name string `json:"name"` Type MetricType `json:"type"` Help string `json:"help"` Labels map[string]string `json:"labels"` Value float64 `json:"value,omitempty"` Buckets []float64 `json:"buckets,omitempty"` } // MetricsCollector 指标采集器 type MetricsCollector struct { counters map[string]*CounterMetric gauges map[string]*GaugeMetric histograms map[string]*HistogramMetric mu sync.RWMutex } // CounterMetric 计数器 type CounterMetric struct { Name string Help string Labels map[string]string value atomic.Int64 } // GaugeMetric 仪表盘 type GaugeMetric struct { Name string Help string Labels map[string]string value atomic.Int64 } // HistogramMetric 直方图 type HistogramMetric struct { Name string Help string Labels map[string]string Buckets []float64 counts map[int]atomic.Int64 sum atomic.Int64 mu sync.Mutex } // NewMetricsCollector 创建指标采集器 func NewMetricsCollector() *MetricsCollector { return &MetricsCollector{ counters: make(map[string]*CounterMetric), gauges: make(map[string]*GaugeMetric), histograms: make(map[string]*HistogramMetric), } } // NewCounter 创建计数器 func (mc *MetricsCollector) NewCounter(name, help string, labels map[string]string) *CounterMetric { mc.mu.Lock() defer mc.mu.Unlock() cm := &CounterMetric{ Name: name, Help: help, Labels: labels, } mc.counters[name] = cm return cm } // Inc 计数器加 1 func (cm *CounterMetric) Inc() { cm.value.Add(1) } // Add 计数器增加值 func (cm *CounterMetric) Add(v int64) { cm.value.Add(v) } // Value 获取当前值 func (cm *CounterMetric) Value() int64 { return cm.value.Load() } // NewGauge 创建仪表盘 func (mc *MetricsCollector) NewGauge(name, help string, labels map[string]string) *GaugeMetric { mc.mu.Lock() defer mc.mu.Unlock() gm := &GaugeMetric{ Name: name, Help: help, Labels: labels, } mc.gauges[name] = gm return gm } // Set 设置仪表盘值 func (gm *GaugeMetric) Set(v int64) { gm.value.Store(v) } // Value 获取当前值 func (gm *GaugeMetric) Value() int64 { return gm.value.Load() } // NewHistogram 创建直方图 func (mc *MetricsCollector) NewHistogram(name, help string, buckets []float64, labels map[string]string) *HistogramMetric { mc.mu.Lock() defer mc.mu.Unlock() hm := &HistogramMetric{ Name: name, Help: help, Labels: labels, Buckets: buckets, counts: make(map[int]atomic.Int64), } mc.histograms[name] = hm return hm } // Observe 观察一个值 func (hm *HistogramMetric) Observe(v float64) { hm.sum.Add(int64(v * 1000)) // 保留三位小数 hm.mu.Lock() for i, bucket := range hm.Buckets { if v <= bucket { hm.counts[i].Add(1) } } hm.mu.Unlock() } // ─── LLM 专有指标 ─── // LLMMetrics LLM 相关指标集合 type LLMMetrics struct { // 调用量 TotalRequests *CounterMetric SuccessRequests *CounterMetric FailedRequests *CounterMetric // Token 用量 PromptTokens *CounterMetric CompletionTokens *CounterMetric TotalTokens *CounterMetric // 延迟 LatencyMs *HistogramMetric TTFTMs *HistogramMetric // Time To First Token // 成本 TotalCostUSD *CounterMetric // 并发 ActiveRequests *GaugeMetric QueueDepth *GaugeMetric // 质量 UserRatings *HistogramMetric HallucinationCnt *CounterMetric } // InitLLMMetrics 初始化 LLM 指标 func InitLLMMetrics(mc *MetricsCollector) *LLMMetrics { return &LLMMetrics{ TotalRequests: mc.NewCounter("llm_requests_total", "Total LLM requests", nil), SuccessRequests: mc.NewCounter("llm_requests_success", "Successful LLM requests", nil), FailedRequests: mc.NewCounter("llm_requests_failed", "Failed LLM requests", nil), PromptTokens: mc.NewCounter("llm_prompt_tokens_total", "Total prompt tokens", nil), CompletionTokens: mc.NewCounter("llm_completion_tokens_total", "Total completion tokens", nil), TotalTokens: mc.NewCounter("llm_total_tokens_total", "Total tokens", nil), LatencyMs: mc.NewHistogram("llm_latency_ms", "LLM request latency ms", []float64{100, 500, 1000, 2000, 5000, 10000}, nil), TTFTMs: mc.NewHistogram("llm_ttft_ms", "Time to first token ms", []float64{200, 500, 1000, 2000, 5000}, nil), TotalCostUSD: mc.NewCounter("llm_cost_usd_total", "Total cost in USD", nil), ActiveRequests: mc.NewGauge("llm_active_requests", "Active LLM requests", nil), QueueDepth: mc.NewGauge("llm_queue_depth", "LLM request queue depth", nil), UserRatings: mc.NewHistogram("llm_user_rating", "User rating (1-5)", []float64{1, 2, 3, 4, 5}, nil), HallucinationCnt: mc.NewCounter("llm_hallucinations_total", "Detected hallucinations", nil), } }

五、日志采集（Logging）

// observability/log.go package observability import ( "encoding/json" "fmt" "io" "log" "os" "runtime" "time" ) // LogLevel 日志级别 type LogLevel string const ( LogDebug LogLevel = "debug" LogInfo LogLevel = "info" LogWarn LogLevel = "warn" LogError LogLevel = "error" LogFatal LogLevel = "fatal" ) // LogEntry 日志条目 type LogEntry struct { Timestamp string `json:"timestamp"` Level LogLevel `json:"level"` Message string `json:"message"` Service string `json:"service"` TraceID TraceID `json:"trace_id,omitempty"` SpanID SpanID `json:"span_id,omitempty"` Caller string `json:"caller,omitempty"` Fields map[string]interface{} `json:"fields,omitempty"` Error string `json:"error,omitempty"` StackTrace string `json:"stack_trace,omitempty"` } // StructuredLogger 结构化日志记录器 type StructuredLogger struct { service string writer io.Writer minLevel LogLevel } // NewStructuredLogger 创建结构化日志记录器 func NewStructuredLogger(service string, writer io.Writer) *StructuredLogger { if writer == nil { writer = os.Stdout } return &StructuredLogger{ service: service, writer: writer, minLevel: LogInfo, } } func (l *StructuredLogger) Debug(msg string, fields map[string]interface{}) { l.log(LogDebug, msg, fields, "") } func (l *StructuredLogger) Info(msg string, fields map[string]interface{}) { l.log(LogInfo, msg, fields, "") } func (l *StructuredLogger) Warn(msg string, fields map[string]interface{}) { l.log(LogWarn, msg, fields, "") } func (l *StructuredLogger) Error(msg string, err error, fields map[string]interface{}) { errMsg := "" stackTrace := "" if err != nil { errMsg = err.Error() stackTrace = getStackTrace() } l.log(LogError, msg, fields, errMsg, stackTrace) } func (l *StructuredLogger) Fatal(msg string, err error, fields map[string]interface{}) { errMsg := "" stackTrace := "" if err != nil { errMsg = err.Error() stackTrace = getStackTrace() } l.log(LogFatal, msg, fields, errMsg, stackTrace) os.Exit(1) } func (l *StructuredLogger) log(level LogLevel, msg string, fields map[string]interface{}, extras ...string) { if level < l.minLevel { return } entry := LogEntry{ Timestamp: time.Now().Format(time.RFC3339Nano), Level: level, Message: msg, Service: l.service, Fields: fields, } // 从 context 中获取 Trace/Span ID // 实际使用时通过中间件注入 if len(extras) > 0 { entry.Error = extras[0] } if len(extras) > 1 { entry.StackTrace = extras[1] } // 获取调用者信息 if pc, file, line, ok := runtime.Caller(2); ok { entry.Caller = fmt.Sprintf("%s:%d %s", file, line, runtime.FuncForPC(pc).Name()) } data, _ := json.Marshal(entry) fmt.Fprintln(l.writer, string(data)) } func getStackTrace() string { buf := make([]byte, 4096) n := runtime.Stack(buf, false) return string(buf[:n]) } // ─── LLM 专用日志 ─── // LLMLogEntry LLM 调用日志 type LLMLogEntry struct { *LogEntry LLMProvider string `json:"llm_provider"` Model string `json:"model"` Prompt string `json:"prompt,omitempty"` Completion string `json:"completion,omitempty"` PromptTokens int `json:"prompt_tokens"` CompTokens int `json:"completion_tokens"` LatencyMs int64 `json:"latency_ms"` CostUSD float64 `json:"cost_usd"` UserID string `json:"user_id,omitempty"` SessionID string `json:"session_id,omitempty"` }

六、质量评估

// observability/quality.go package observability import ( "strings" "sync" "time" ) // QualityEvaluator 质量评估器 type QualityEvaluator struct { rules []QualityRule mu sync.RWMutex } // QualityRule 质量规则 type QualityRule struct { Name string Description string Check func(prompt, completion string) (bool, float64, string) } // QualityReport 质量报告 type QualityReport struct { Timestamp time.Time `json:"timestamp"` OverallScore float64 `json:"overall_score"` Checks []CheckResult `json:"checks"` } // CheckResult 检查结果 type CheckResult struct { RuleName string `json:"rule_name"` Passed bool `json:"passed"` Score float64 `json:"score"` Detail string `json:"detail,omitempty"` } // NewQualityEvaluator 创建质量评估器 func NewQualityEvaluator() *QualityEvaluator { qe := &QualityEvaluator{} qe.registerDefaultRules() return qe } func (qe *QualityEvaluator) registerDefaultRules() { qe.rules = []QualityRule{ { Name: "拒绝回答检测", Description: "检测模型是否拒绝回答（如'我不能回答这个问题'）", Check: func(prompt, completion string) (bool, float64, string) { refusalPhrases := []string{ "我不能", "无法回答", "很抱歉", "对不起", "I cannot", "I'm sorry", "I apologize", } for _, phrase := range refusalPhrases { if strings.Contains(completion, phrase) { return false, 0.0, "检测到拒绝回答模式" } } return true, 1.0, "" }, }, { Name: "重复检测", Description: "检测回答中是否存在大量重复内容", Check: func(prompt, completion string) (bool, float64, string) { sentences := strings.Split(completion, "。") seen := make(map[string]int) for _, s := range sentences { s = strings.TrimSpace(s) if len(s) > 10 { seen[s]++ if seen[s] > 2 { return false, 0.3, "检测到重复句子" } } } return true, 1.0, "" }, }, { Name: "长度合理性", Description: "检测回答长度是否合理（非空且不过短）", Check: func(prompt, completion string) (bool, float64, string) { length := len([]rune(completion)) if length < 5 { return false, 0.2, "回答过短" } if length > 8000 { return false, 0.5, "回答过长" } return true, 1.0, "" }, }, { Name: "幻觉检测（简单版）", Description: "检测是否包含常见的幻觉模式", Check: func(prompt, completion string) (bool, float64, string) { hallucinationSignals := []string{ "据我所知", "我记得", "我了解到", "根据我的知识", "在我的记忆中", } count := 0 for _, signal := range hallucinationSignals { count += strings.Count(completion, signal) } if count > 3 { return false, 0.5, "检测到可能的幻觉模式" } return true, 1.0, "" }, }, } } // Evaluate 评估回答质量 func (qe *QualityEvaluator) Evaluate(prompt, completion string) *QualityReport { qe.mu.RLock() defer qe.mu.RUnlock() report := &QualityReport{ Timestamp: time.Now(), } totalScore := 0.0 for _, rule := range qe.rules { passed, score, detail := rule.Check(prompt, completion) result := CheckResult{ RuleName: rule.Name, Passed: passed, Score: score, Detail: detail, } report.Checks = append(report.Checks, result) totalScore += score } report.OverallScore = totalScore / float64(len(qe.rules)) return report } // ─── 成本核算 ─── // CostCalculator 成本计算器 type CostCalculator struct { prices map[string]ModelPrice } // ModelPrice 模型价格 type ModelPrice struct { PromptPricePer1K float64 // 每千 token 输入价格（USD） CompletionPricePer1K float64 // 每千 token 输出价格（USD） } func NewCostCalculator() *CostCalculator { return &CostCalculator{ prices: map[string]ModelPrice{ "gpt-4": {0.03, 0.06}, "gpt-4-turbo": {0.01, 0.03}, "gpt-3.5-turbo": {0.001, 0.002}, "claude-3-opus": {0.015, 0.075}, "claude-3-sonnet": {0.003, 0.015}, "claude-3-haiku": {0.00025, 0.00125}, "ERNIE-Bot": {0.008, 0.012}, }, } } func (cc *CostCalculator) Calculate(model string, promptTokens, completionTokens int) float64 { price, ok := cc.prices[model] if !ok { return 0 } promptCost := float64(promptTokens) / 1000 * price.PromptPricePer1K completionCost := float64(completionTokens) / 1000 * price.CompletionPricePer1K return promptCost + completionCost }

七、告警体系

// observability/alert.go package observability import ( "fmt" "log" "sync" "time" ) // AlertSeverity 告警级别 type AlertSeverity string const ( AlertInfo AlertSeverity = "info" AlertWarning AlertSeverity = "warning" AlertCritical AlertSeverity = "critical" ) // AlertRule 告警规则 type AlertRule struct { Name string Description string Severity AlertSeverity Interval time.Duration EvalFunc func(metrics *LLMMetrics) (bool, string) lastAlert time.Time mu sync.Mutex } // AlertManager 告警管理器 type AlertManager struct { rules []*AlertRule notifiers []AlertNotifier metrics *LLMMetrics alertHistory []AlertEvent mu sync.RWMutex } // AlertNotifier 告警通知接口 type AlertNotifier interface { Notify(alert AlertEvent) error } // AlertEvent 告警事件 type AlertEvent struct { Timestamp time.Time `json:"timestamp"` RuleName string `json:"rule_name"` Severity AlertSeverity `json:"severity"` Message string `json:"message"` ResolvedAt time.Time `json:"resolved_at,omitempty"` } // NewAlertManager 创建告警管理器 func NewAlertManager(metrics *LLMMetrics) *AlertManager { am := &AlertManager{ metrics: metrics, } am.registerDefaultRules() return am } func (am *AlertManager) registerDefaultRules() { am.rules = []*AlertRule{ { Name: "high_error_rate", Description: "错误率过高", Severity: AlertCritical, Interval: 5 * time.Minute, EvalFunc: func(m *LLMMetrics) (bool, string) { total := m.TotalRequests.Value() failed := m.FailedRequests.Value() if total > 100 { rate := float64(failed) / float64(total) if rate > 0.05 { return true, fmt.Sprintf("错误率 %.1f%% 超过阈值 5%%", rate*100) } } return false, "" }, }, { Name: "high_latency", Description: "P99 延迟过高", Severity: AlertWarning, Interval: 10 * time.Minute, EvalFunc: func(m *LLMMetrics) (bool, string) { // 简化版：实际应从直方图中计算 P99 return false, "" }, }, { Name: "cost_spike", Description: "成本异常飙升", Severity: AlertWarning, Interval: 15 * time.Minute, EvalFunc: func(m *LLMMetrics) (bool, string) { cost := m.TotalCostUSD.Value() if cost > 1000 { return true, fmt.Sprintf("累计成本 $%.2f 超过阈值 $1000", float64(cost)/100) } return false, "" }, }, { Name: "low_quality", Description: "回答质量持续偏低", Severity: AlertWarning, Interval: 30 * time.Minute, EvalFunc: func(m *LLMMetrics) (bool, string) { // 实际应从质量评估结果中计算 return false, "" }, }, } } // AddNotifier 添加通知渠道 func (am *AlertManager) AddNotifier(n AlertNotifier) { am.notifiers = append(am.notifiers, n) } // Evaluate 评估所有规则 func (am *AlertManager) Evaluate() { for _, rule := range am.rules { rule.mu.Lock() if time.Since(rule.lastAlert) < rule.Interval { rule.mu.Unlock() continue } rule.mu.Unlock() triggered, message := rule.EvalFunc(am.metrics) if !triggered { continue } event := AlertEvent{ Timestamp: time.Now(), RuleName: rule.Name, Severity: rule.Severity, Message: message, } am.mu.Lock() am.alertHistory = append(am.alertHistory, event) am.mu.Unlock() rule.mu.Lock() rule.lastAlert = time.Now() rule.mu.Unlock() // 发送通知 for _, notifier := range am.notifiers { if err := notifier.Notify(event); err != nil { log.Printf("[Alert] Notify error: %v", err) } } } } // StartAlertLoop 启动告警循环 func (am *AlertManager) StartAlertLoop(interval time.Duration, stop chan struct{}) { ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ticker.C: am.Evaluate() case <-stop: return } } } // ─── Slack Notifier ─── type SlackNotifier struct { WebhookURL string } func (sn *SlackNotifier) Notify(alert AlertEvent) error { color := "#36a64f" if alert.Severity == AlertWarning { color = "#ffcc00" } else if alert.Severity == AlertCritical { color = "#ff0000" } message := fmt.Sprintf(`{ "attachments": [{ "color": "%s", "title": "[%s] %s", "text": "%s", "footer": "AI Observability", "ts": %d }] }`, color, alert.Severity, alert.RuleName, alert.Message, alert.Timestamp.Unix()) // 实际发送 HTTP POST 到 webhook URL log.Printf("[Slack] Would send: %s", message) return nil }

八、集成中间件

// observability/middleware.go package observability import ( "net/http" "time" ) // ObservabilityMiddleware 可观测性中间件 type ObservabilityMiddleware struct { tracer *Tracer metrics *LLMMetrics logger *StructuredLogger quality *QualityEvaluator } // NewObservabilityMiddleware 创建中间件 func NewObservabilityMiddleware(tracer *Tracer, metrics *LLMMetrics, logger *StructuredLogger) *ObservabilityMiddleware { return &ObservabilityMiddleware{ tracer: tracer, metrics: metrics, logger: logger, quality: NewQualityEvaluator(), } } // WrapHandler 包装 HTTP 处理器 func (m *ObservabilityMiddleware) WrapHandler(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() // 创建根 Span ctx, span := m.tracer.StartSpan(r.Context(), r.URL.Path, WithSpanKind(SpanKindServer)) m.tracer.SetAttribute(span, "http.method", r.Method) m.tracer.SetAttribute(span, "http.url", r.URL.String()) m.tracer.SetAttribute(span, "http.user_agent", r.UserAgent()) // 指标：活跃请求 +1 m.metrics.ActiveRequests.Inc() m.metrics.ActiveRequests.Add(1) // 包装 ResponseWriter 以捕获状态码 wrapped := &responseWriter{ResponseWriter: w, statusCode: 200} // 执行下一个处理器 next.ServeHTTP(wrapped, r.WithContext(ctx)) // 记录日志 duration := time.Since(start) fields := map[string]interface{}{ "method": r.Method, "path": r.URL.Path, "status": wrapped.statusCode, "duration_ms": duration.Milliseconds(), } if wrapped.statusCode >= 500 { m.logger.Error("Request failed", nil, fields) m.metrics.FailedRequests.Inc() } else { m.logger.Info("Request completed", fields) m.metrics.SuccessRequests.Inc() } // 指标：活跃请求 -1 m.metrics.ActiveRequests.Add(-1) m.metrics.TotalRequests.Inc() // 延迟 m.metrics.LatencyMs.Observe(float64(duration.Milliseconds())) // 结束 Span if wrapped.statusCode >= 500 { m.tracer.SetError(span, nil) } m.tracer.EndSpan(span) }) } // WrapLLMCall 包装 LLM 调用 func (m *ObservabilityMiddleware) WrapLLMCall( ctx context.Context, provider, model string, promptTokens, completionTokens int, latency time.Duration, err error, ) { // 指标 m.metrics.TotalRequests.Inc() m.metrics.PromptTokens.Add(int64(promptTokens)) m.metrics.CompletionTokens.Add(int64(completionTokens)) m.metrics.TotalTokens.Add(int64(promptTokens + completionTokens)) m.metrics.LatencyMs.Observe(float64(latency.Milliseconds())) // 成本计算 calculator := NewCostCalculator() cost := calculator.Calculate(model, promptTokens, completionTokens) m.metrics.TotalCostUSD.Add(int64(cost * 100)) // 保留两位小数 if err != nil { m.metrics.FailedRequests.Inc() } else { m.metrics.SuccessRequests.Inc() } // 日志 fields := map[string]interface{}{ "provider": provider, "model": model, "prompt_tokens": promptTokens, "completion_tokens": completionTokens, "total_tokens": promptTokens + completionTokens, "latency_ms": latency.Milliseconds(), "cost_usd": cost, } if err != nil { m.logger.Error("LLM call failed", err, fields) } else { m.logger.Info("LLM call succeeded", fields) } } type responseWriter struct { http.ResponseWriter statusCode int } func (rw *responseWriter) WriteHeader(code int) { rw.statusCode = code rw.ResponseWriter.WriteHeader(code) }

九、生产坑位

十、本讲小结

┌─────────────────────────────────────────────────────────────┐ │ 第7讲 核心收获 │ │ │ │ ✅ 实现了完整的链路追踪（Trace/Span/导出器） │ │ ✅ 搭建了 LLM 专有指标体系（调用量/Tokens/延迟/成本） │ │ ✅ 实现了结构化日志（分级/字段/调用链关联） │ │ ✅ 构建了质量评估系统（拒绝检测/重复/长度/幻觉） │ │ ✅ 搭建了告警体系（规则/通知/防重复） │ │ ✅ 实现了可观测性中间件（自动埋点/日志/指标） │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念：没有可观测性的 AI 系统就像黑盒，出了问题只能靠猜 │ └─────────────────────────────────────────────────────────────┘

下一讲预告：《安全防护——防止你的 AI 被玩坏》

你会学到：Prompt 注入防御、敏感信息过滤、速率限制、内容审核、审计日志、安全架构设计。