当前位置：首页 > news >正文

AI应用的可观测性建设：从日志到链路追踪

news 2026/5/22 14:32:17

AI应用的可观测性建设：从日志到链路追踪

前言

在大厂时，我们有专门的可观测性平台（日志、监控、链路追踪），我以为这是大公司的专利。创业后第一次遇到线上问题时，我们只有打印的日志，查了3个小时才定位到问题。

那次之后我意识到：可观测性不是奢侈品，而是必需品。一个没有可观测性的系统，就像一个没有仪表盘的飞机，出了问题都不知道在哪。

今天，分享我们是如何从零开始搭建 AI 应用的可观测性体系的。

一、可观测性的三大支柱

1.1 可观测性 vs 监控

维度	传统监控	可观测性
思维模式	预设问题，针对检查	未知问题，自由探索
关注点	指标告警	数据关联分析
调试方式	告警 -> 排查	数据 -> 探索 -> 根因
数据要求	结构化指标	多维度原始数据

1.2 三大支柱

可观测性 = 日志（Logs）+ 指标（Metrics）+ 链路追踪（Traces）

类型	说明	工具
日志	离散的事件记录	ELK, Loki
指标	聚合的数值数据	Prometheus, InfluxDB
链路追踪	请求的完整路径	Jaeger, Zipkin

二、日志体系建设

2.1 日志规范

import logging import json from datetime import datetime from enum import Enum class LogLevel(Enum): DEBUG = "DEBUG" INFO = "INFO" WARNING = "WARNING" ERROR = "ERROR" CRITICAL = "CRITICAL" class StructuredLogger: def __init__(self, service_name: str): self.service_name = service_name self.logger = logging.getLogger(service_name) self.logger.setLevel(logging.INFO) # JSON formatter handler = logging.StreamHandler() handler.setFormatter(self._create_formatter()) self.logger.addHandler(handler) def _create_formatter(self): """创建 JSON 格式化器""" def formatter(record): log_entry = { "timestamp": datetime.utcnow().isoformat(), "level": record.levelname, "service": self.service_name, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno } # 添加异常信息 if record.exc_info: log_entry["exception"] = self.logger.exception( record.exc_info, exc_info=record.exc_info ) # 添加额外字段 if hasattr(record, "extra"): log_entry.update(record.extra) return json.dumps(log_entry) return formatter def log(self, level: LogLevel, message: str, **kwargs): """记录日志""" extra = {"extra": kwargs} if kwargs else {} getattr(self.logger, level.value.lower())(message, extra=extra) def info(self, message: str, **kwargs): self.log(LogLevel.INFO, message, **kwargs) def error(self, message: str, **kwargs): self.log(LogLevel.ERROR, message, **kwargs) def warning(self, message: str, **kwargs): self.log(LogLevel.WARNING, message, **kwargs)

2.2 AI 应用的日志最佳实践

class AILogger: def __init__(self, logger: StructuredLogger): self.logger = logger def log_model_request(self, request_id: str, model: str, prompt_length: int): """记录模型请求""" self.logger.info( "模型请求开始", request_id=request_id, model=model, prompt_length=prompt_length, event_type="model_request_start" ) def log_model_response(self, request_id: str, response_length: int, latency_ms: float, tokens_used: int): """记录模型响应""" self.logger.info( "模型请求完成", request_id=request_id, response_length=response_length, latency_ms=latency_ms, tokens_used=tokens_used, event_type="model_response_complete" ) def log_model_error(self, request_id: str, error: str, error_type: str): """记录模型错误""" self.logger.error( "模型请求失败", request_id=request_id, error=error, error_type=error_type, event_type="model_error" )

三、指标体系建设

3.1 指标采集

from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry class AIMetrics: def __init__(self, registry: CollectorRegistry = None): self.registry = registry or CollectorRegistry() # 请求计数器 self.request_total = Counter( "ai_request_total", "Total number of AI requests", ["model", "status"], registry=self.registry ) # 请求延迟 self.request_duration = Histogram( "ai_request_duration_seconds", "AI request duration in seconds", ["model", "operation"], buckets=[0.1, 0.5, 1, 2, 5, 10, 30], registry=self.registry ) # Token 使用 self.tokens_used = Counter( "ai_tokens_used_total", "Total tokens used", ["model", "type"], # type: prompt/completion registry=self.registry ) # 当前请求数 self.active_requests = Gauge( "ai_active_requests", "Number of active requests", ["model"], registry=self.registry ) def record_request(self, model: str, status: str, duration: float, tokens: int): """记录请求指标""" self.request_total.labels(model=model, status=status).inc() self.request_duration.labels(model=model, operation="inference").observe(duration) self.tokens_used.labels(model=model, type="prompt").inc(tokens) def increment_active(self, model: str): """增加活跃请求数""" self.active_requests.labels(model=model).inc() def decrement_active(self, model: str): """减少活跃请求数""" self.active_requests.labels(model=model).dec()

3.2 自定义指标

class BusinessMetrics: def __init__(self, registry: CollectorRegistry): # 用户指标 self.active_users = Gauge( "app_active_users", "Number of active users", ["period"], # hourly, daily registry=registry ) # 功能使用指标 self.feature_usage = Counter( "feature_usage_total", "Total feature usage", ["feature_name"], registry=registry ) # 业务转化指标 self.conversion_rate = Gauge( "conversion_rate", "Business conversion rate", ["stage"], # trial_to_paid, etc registry=registry )

四、链路追踪

4.1 分布式追踪基础

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.jaeger.thrift import JaegerExporter class TracingSetup: def __init__(self, service_name: str): self.service_name = service_name self.setup_tracing() def setup_tracing(self): """配置链路追踪""" # 创建 tracer provider provider = TracerProvider() # 添加 Jaeger exporter jaeger_exporter = JaegerExporter( agent_host_name="localhost", agent_port=6831, ) # 添加 batch span processor provider.add_span_processor( BatchSpanProcessor(jaeger_exporter) ) # 设置全局 tracer provider trace.set_tracer_provider(provider) # 获取 tracer self.tracer = trace.get_tracer(self.service_name) def create_span(self, name: str, attributes: dict = None): """创建 span""" return self.tracer.start_as_current_span( name, attributes=attributes or {} )

4.2 AI 应用的链路追踪

class AIDistributedTracing: def __init__(self, tracing: TracingSetup): self.tracing = tracing self.tracer = tracing.tracer def trace_ai_request(self, user_id: str, prompt: str, model: str): """追踪 AI 请求""" with self.tracer.start_as_current_span( "ai.request", attributes={ "user_id": user_id, "model": model, "prompt_length": len(prompt) } ) as span: try: # 模拟 AI 调用 response = self._call_model(prompt, model) # 记录响应信息 span.set_attribute("response_length", len(response)) span.set_attribute("status", "success") return response except Exception as e: span.set_attribute("status", "error") span.set_attribute("error.message", str(e)) raise def _call_model(self, prompt: str, model: str) -> str: """调用模型（实际应用中替换为真实调用）""" import time time.sleep(0.1) # 模拟调用延迟 return f"Response to: {prompt[:50]}..."

五、可观测性集成

5.1 统一日志上下文

from contextvars import ContextVar # 上下文变量 request_id: ContextVar[str] = ContextVar('request_id', default='') user_id: ContextVar[str] = ContextVar('user_id', default='') class UnifiedLogger: def __init__(self): self.logger = StructuredLogger("app") def _get_context(self) -> dict: """获取上下文信息""" return { "request_id": request_id.get(), "user_id": user_id.get() } def info(self, message: str, **kwargs): context = self._get_context() context.update(kwargs) self.logger.info(message, **context)

5.2 告警配置

# alertmanager.yml groups: - name: ai_alerts rules: - alert: HighErrorRate expr: rate(ai_request_total{status="error"}[5m]) > 0.1 for: 5m labels: severity: critical annotations: summary: "AI 请求错误率过高" description: "当前错误率: ${{ $value }}" - alert: HighLatency expr: histogram_quantile(0.95, ai_request_duration_seconds) > 10 for: 5m labels: severity: warning annotations: summary: "AI 请求延迟过高" description: "P95 延迟: ${{ $value }}s" - alert: APIKeyNearlyExhausted expr: api_usage_percentage > 90 for: 1h labels: severity: warning annotations: summary: "API 额度即将耗尽"

六、实战案例：问题排查

6.1 问题场景

用户反馈："AI 客服响应很慢，有时还会失败"

6.2 排查流程

# 1. 查看错误率指标 # prometheus query: rate(ai_request_total{status="error"}[5m]) # 2. 查看延迟分布 # prometheus query: histogram_quantile(0.95, ai_request_duration_seconds) # 3. 查看具体错误日志 # loki query: {service="ai-service"} |= "error" | json # 4. 查看链路追踪 # jaeger query: service=ai-service operation=/api/chat