监控

为什么 LLM 需要专项可观测性？ 传统 APM 不够：LLM 有 Token 计费、Prompt 变体、模型路由、工具调用链等特有维度。一个请求可能涉及 3 个模型 + 5 个工具调用 + 2 次检索，没有 Tracing 根本无法定位问题。 三位一体架构 ┌──────────────────────────────────────────────────┐ │ 用户请求 │ │ trace_id = xxx │ └──────────────────────┬───────────────────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Logs │ │ Traces │ │ Metrics │ │ 结构化 │ │ 链路 │ │ 聚合 │ │ 日志 │ │ 追踪 │ │ 指标 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ ELK / │ │ Jaeger /│ │Prometheus│ │ Loki │ │ Langfuse│ │ + Grafana│ └─────────┘ └─────────┘ └─────────┘ │ │ │ └──────────────┼──────────────┘ ▼ ┌─────────────────┐ │ AlertManager │ │ 告警 + 通知 │ └─────────────────┘ 一、结构化日志 import structlog import json # 配置 structlog structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], ) logger = structlog.get_logger() class LLMLogger: """LLM 专用结构化日志""" def log_request(self, trace_id: str, user_id: str, model: str, prompt: str, **kwargs): logger.info("llm_request", trace_id=trace_id, user_id=user_id, model=model, prompt_length=len(prompt), prompt_tokens=kwargs.get("input_tokens"), max_tokens=kwargs.get("max_tokens"), temperature=kwargs.get("temperature", 1.0), tools=kwargs.get("tools"), timestamp=datetime.utcnow().isoformat(), ) def log_response(self, trace_id: str, response: str, input_tokens: int, output_tokens: int, latency_ms: float, model: str, **kwargs): logger.info("llm_response", trace_id=trace_id, model=model, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, latency_ms=latency_ms, finish_reason=kwargs.get("finish_reason"), cost_usd=self._calc_cost(model, input_tokens, output_tokens), ) def log_tool_call(self, trace_id: str, tool_name: str, params: dict, result: dict, latency_ms: float): logger.info("tool_call", trace_id=trace_id, tool=tool_name, params_keys=list(params.keys()), result_status="success" if result.get("success") else "failed", latency_ms=latency_ms, ) def log_error(self, trace_id: str, error: Exception, context: dict): logger.error("llm_error", trace_id=trace_id, error_type=type(error).__name__, error_message=str(error), context=context, ) 日志查询示例 # ELK / Loki 查询：查找高延迟请求 # Kibana KQL: # llm_response AND latency_ms > 5000 AND model: "gpt-4" # Grafana Loki LogQL: # {app="llm-service"} |= "llm_response" | json | latency_ms > 5000 二、分布式链路追踪 LLM 请求的典型链路：API → Router → Cache → Model → Tool → Model → Response ...

为什么 AI 应用需要可观测性 传统应用监控关注 CPU、内存、QPS。AI 应用需要关注：LLM 调用了几次？每次延迟多少？Token 消耗多少？检索结果是否相关？Agent 的推理链路是什么？没有可观测性，AI 应用就是黑盒，出问题只能猜。 三支柱模型 支柱 传统应用 AI 应用 工具 Tracing 请求链路 LLM 调用链 + Agent 推理步骤 + 检索过程 LangSmith/Langfuse Logging 日志 Prompt/Response/中间状态 结构化日志 Metrics QPS/延迟 Token消耗/检索召回率/幻觉率 Prometheus+Grafana 关键指标定义 from dataclasses import dataclass from typing import Optional @dataclass class LLMCallMetrics: # 延迟指标 time_to_first_token: float # 首 token 延迟(ms) total_latency: float # 总延迟(ms) # 成本指标 prompt_tokens: int completion_tokens: int total_cost: float # 质量指标 model_name: str temperature: float finish_reason: str # 检索指标(RAG场景) retrieval_count: Optional[int] = None retrieval_relevance: Optional[float] = None context_utilization: Optional[float] = None @dataclass class AgentTrace: trace_id: str spans: list[dict] # 每个步骤一个 span # span 结构: name, start, end, input, output, metadata Langfuse 集成实现 from langfuse import Langfuse from langfuse.decorators import langfuse_context, observe import functools, time langfuse = Langfuse( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="https://your-langfuse.com" ) def trace_llm(name="llm_call"): def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): start = time.time() span = langfuse_context.start_span(name=name, input=kwargs) try: result = await func(*args, **kwargs) ttft = time.time() - start span.end( output=result, metadata={"ttft_ms": ttft * 1000, "model": kwargs.get("model", "unknown")} ) langfuse_context.update_current_observation( usage={ "prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0), "completion_tokens": result.get("usage", {}).get("completion_tokens", 0), "total_tokens": result.get("usage", {}).get("total_tokens", 0), }, metadata={"latency_ms": (time.time() - start) * 1000} ) return result except Exception as e: span.end(level="ERROR", status_message=str(e)) raise return wrapper return decorator @observe(name="rag_pipeline") async def rag_pipeline(query: str): docs = await retrieve(query) answer = await generate(query, docs) return answer @observe(name="retrieve") async def retrieve(query: str): results = await vector_search(query) return {"query": query, "docs": results} @trace_llm(name="generate") async def generate(query: str, docs: list): resp = await llm_client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"Q: {query}\nContext: {docs}"}] ) return resp 工具对比 工具 部署方式 核心优势 适用场景 价格 LangSmith SaaS LangChain 深度集成 LangChain 用户 $39/月起 Langfuse 自部署/SaaS 开源，框架无关 通用 AI 应用 免费/自部署 Phoenix 自部署 Arize 出品，评估强 需要评估体系 开源 OpenTelemetry 自部署 标准化，通用 已有 OTel 体系 免费 选型建议：不用 LangChain 选 Langfuse（开源、框架无关），用 LangChain 选 LangSmith（集成最好），需要深度评估选 Phoenix。 ...