一、为什么需要 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, Im 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) }九、生产坑位#坑位现象解决方案1埋点性能损耗大量 Span 导致应用变慢采样策略头部采样/尾部采样 异步导出2指标基数爆炸高基数 Label如 user_id导致 Prometheus OOM只保留聚合指标明细走日志3日志泄露敏感信息Prompt 中包含用户身份证号等 PII日志脱敏过滤器 敏感字段掩码4存储成本过高每天 TB 级的 Trace/日志数据设置保留周期 采样率 冷热分离5告警风暴一个故障触发几十条告警告警聚合 依赖关系分析 静默期6质量评估不准规则误判导致无效告警渐进式规则 A/B 测试 人工标注反馈十、本讲小结┌─────────────────────────────────────────────────────────────┐ │ 第7讲 核心收获 │ │ │ │ ✅ 实现了完整的链路追踪Trace/Span/导出器 │ │ ✅ 搭建了 LLM 专有指标体系调用量/Tokens/延迟/成本 │ │ ✅ 实现了结构化日志分级/字段/调用链关联 │ │ ✅ 构建了质量评估系统拒绝检测/重复/长度/幻觉 │ │ ✅ 搭建了告警体系规则/通知/防重复 │ │ ✅ 实现了可观测性中间件自动埋点/日志/指标 │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念没有可观测性的 AI 系统就像黑盒出了问题只能靠猜 │ └─────────────────────────────────────────────────────────────┘下一讲预告《安全防护——防止你的 AI 被玩坏》你会学到Prompt 注入防御、敏感信息过滤、速率限制、内容审核、审计日志、安全架构设计。 
长河君 · 2026/10/31 2:05:04 · 阅读 1.2k 第7讲：可观测性与 AI 监控——你的 Agent 到底在干什么
一、为什么需要 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, Im 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) }九、生产坑位#坑位现象解决方案1埋点性能损耗大量 Span 导致应用变慢采样策略头部采样/尾部采样 异步导出2指标基数爆炸高基数 Label如 user_id导致 Prometheus OOM只保留聚合指标明细走日志3日志泄露敏感信息Prompt 中包含用户身份证号等 PII日志脱敏过滤器 敏感字段掩码4存储成本过高每天 TB 级的 Trace/日志数据设置保留周期 采样率 冷热分离5告警风暴一个故障触发几十条告警告警聚合 依赖关系分析 静默期6质量评估不准规则误判导致无效告警渐进式规则 A/B 测试 人工标注反馈十、本讲小结┌─────────────────────────────────────────────────────────────┐ │ 第7讲 核心收获 │ │ │ │ ✅ 实现了完整的链路追踪Trace/Span/导出器 │ │ ✅ 搭建了 LLM 专有指标体系调用量/Tokens/延迟/成本 │ │ ✅ 实现了结构化日志分级/字段/调用链关联 │ │ ✅ 构建了质量评估系统拒绝检测/重复/长度/幻觉 │ │ ✅ 搭建了告警体系规则/通知/防重复 │ │ ✅ 实现了可观测性中间件自动埋点/日志/指标 │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念没有可观测性的 AI 系统就像黑盒出了问题只能靠猜 │ └─────────────────────────────────────────────────────────────┘下一讲预告《安全防护——防止你的 AI 被玩坏》你会学到Prompt 注入防御、敏感信息过滤、速率限制、内容审核、审计日志、安全架构设计。