LLM 可观测性技术栈：Log/Trace/Metric 三位一体

为什么 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

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer(__name__)

class TracedLLMCall:
    """带链路追踪的 LLM 调用"""

    async def call(self, trace_id: str, model: str,
                   messages: list, **kwargs) -> dict:
        with tracer.start_as_current_span("llm_inference") as span:
            span.set_attribute("trace_id", trace_id)
            span.set_attribute("model", model)
            span.set_attribute("message_count", len(messages))
            span.set_attribute("temperature", kwargs.get("temperature", 1.0))

            # 子 span: 缓存检查
            with tracer.start_as_current_span("cache_check"):
                cached = await self._check_cache(messages)
                span.set_attribute("cache_hit", cached is not None)
                if cached:
                    span.set_attribute("cache.result", "hit")
                    return cached

            # 子 span: 模型调用
            with tracer.start_as_current_span("model_call") as model_span:
                model_span.set_attribute("model.endpoint", self._get_endpoint(model))
                response = await self._call_model(model, messages, **kwargs)
                model_span.set_attribute("input_tokens", response.usage.input_tokens)
                model_span.set_attribute("output_tokens", response.usage.output_tokens)
                model_span.set_attribute("latency_ms", response.latency_ms)

            # 子 span: 后处理
            with tracer.start_as_current_span("postprocess"):
                result = await self._postprocess(response)

            return result

Agent 链路追踪

class TracedAgent:
    """Agent 多步推理的链路追踪"""

    async def run(self, query: str) -> str:
        with tracer.start_as_current_span("agent_run") as span:
            span.set_attribute("query", query[:200])

            for step in range(self.max_steps):
                with tracer.start_as_current_span(f"step_{step}") as step_span:
                    step_span.set_attribute("step_number", step)

                    # 推理
                    with tracer.start_as_current_span("reasoning"):
                        action = await self._reason(query)
                        step_span.set_attribute("action", action.tool_name)

                    # 工具调用
                    with tracer.start_as_current_span("tool_call"):
                        result = await self._call_tool(action)
                        step_span.set_attribute("tool.success", result.success)

                    if action.finish:
                        span.set_attribute("total_steps", step + 1)
                        return result.output

三、关键指标

from prometheus_client import Counter, Histogram, Gauge, Summary

# === 请求指标 ===
REQUEST_COUNT = Counter(
    "llm_requests_total", "Total LLM requests",
    ["model", "status", "endpoint"]
)
REQUEST_LATENCY = Histogram(
    "llm_request_latency_seconds", "Request latency",
    ["model"],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60, 120]
)

# === Token 指标 ===
TOKEN_USAGE = Counter(
    "llm_tokens_total", "Token usage",
    ["type", "model"]  # type: input/output
)
TOKEN_COST = Counter(
    "llm_cost_usd_total", "LLM cost in USD",
    ["model", "user_tier"]
)

# === 质量指标 ===
ERROR_RATE = Gauge(
    "llm_error_rate", "Error rate",
    ["model", "error_type"]
)
TIMEOUT_RATE = Gauge(
    "llm_timeout_rate", "Timeout rate",
    ["model"]
)

# === 缓存指标 ===
CACHE_HIT_RATE = Gauge(
    "llm_cache_hit_rate", "Cache hit rate",
    ["cache_type"]  # exact / semantic
)

# === Agent 指标 ===
AGENT_STEPS = Histogram(
    "agent_steps", "Steps per agent run",
    ["agent_type"],
    buckets=[1, 2, 5, 10, 15, 20, 30]
)
TOOL_CALL_COUNT = Counter(
    "agent_tool_calls_total", "Tool call count",
    ["tool_name", "status"]
)

# === 并发指标 ===
ACTIVE_REQUESTS = Gauge(
    "llm_active_requests", "Currently active requests",
    ["model"]
)
QUEUE_SIZE = Gauge(
    "llm_queue_size", "Request queue size"
)

Grafana 看板配置

{
  "dashboard": {
    "title": "LLM Production Dashboard",
    "panels": [
      {
        "title": "Request Rate",
        "query": "rate(llm_requests_total[5m])",
        "type": "graph"
      },
      {
        "title": "Latency P50/P95/P99",
        "query": "histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m]))",
        "type": "graph"
      },
      {
        "title": "Token Usage by Model",
        "query": "sum by (model) (rate(llm_tokens_total[1h]))",
        "type": "pie"
      },
      {
        "title": "Cost (USD/hour)",
        "query": "rate(llm_cost_usd_total[1h]) * 3600",
        "type": "stat"
      },
      {
        "title": "Error Rate",
        "query": "llm_error_rate",
        "type": "gauge",
        "thresholds": [{"value": 0.01, "color": "green"},
                       {"value": 0.05, "color": "red"}]
      },
      {
        "title": "Cache Hit Rate",
        "query": "llm_cache_hit_rate",
        "type": "gauge"
      }
    ]
  }
}

四、工具选型

工具	定位	LLM 专项	价格	推荐
Langfuse	LLM 可观测性	原生	开源 + SaaS	⭐⭐⭐⭐⭐
LangSmith	LangChain 生态	原生	SaaS	⭐⭐⭐⭐
Datadog	通用 APM	插件	$$$	⭐⭐⭐
Arize Phoenix	ML 可观测性	原生	开源 + SaaS	⭐⭐⭐⭐
Grafana + Prometheus	通用监控	自定义	开源	⭐⭐⭐⭐
Helicone	LLM 代理层	原生	SaaS	⭐⭐⭐

Langfuse 集成示例

from langfuse import Langfuse
from langfuse.openai import openai

langfuse = Langfuse(
    public_key="pk-lf-xxx",
    secret_key="sk-lf-xxx",
    host="https://cloud.langfuse.com"
)

# 自动追踪 OpenAI 调用
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}],
    metadata={"trace_id": "xxx", "user_id": "user123"}
)

# 手动追踪自定义逻辑
with langfuse.trace(id="trace-xxx", name="agent_run") as trace:
    trace.generation(
        name="llm_call",
        model="gpt-4",
        input=messages,
        output=response.choices[0].message,
        usage={"input": 100, "output": 50},
        metadata={"temperature": 0.7}
    )
    trace.span(name="tool_call", input={"tool": "search"}, output={"result": "..."})

五、告警规则

# Prometheus AlertManager 规则
groups:
  - name: llm_alerts
    rules:
      # P99 延迟超过 10s
      - alert: HighLatencyP99
        expr: histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m])) > 10
        for: 5m
        labels: {severity: warning}
        annotations:
          summary: "LLM P99 latency > 10s for {{ $labels.model }}"

      # 错误率超过 5%
      - alert: HighErrorRate
        expr: llm_error_rate > 0.05
        for: 2m
        labels: {severity: critical}
        annotations:
          summary: "Error rate > 5% for {{ $labels.model }}"

      # 单用户小时成本超过 $100
      - alert: UserCostSpike
        expr: sum by (user_id) (rate(llm_cost_usd_total[1h])) * 3600 > 100
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "User {{ $labels.user_id }} cost > $100/hour"

      # 缓存命中率骤降
      - alert: CacheHitRateDrop
        expr: llm_cache_hit_rate < 0.3
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "Cache hit rate dropped below 30%"

      # Agent 步数异常
      - alert: AgentStepAnomaly
        expr: histogram_quantile(0.95, rate(agent_steps_bucket[10m])) > 15
        for: 5m
        labels: {severity: warning}
        annotations:
          summary: "Agent P95 steps > 15, possible loop"

关键指标速查表

指标	正常范围	告警阈值	紧急阈值
P50 延迟	0.5-2s	> 3s	> 5s
P99 延迟	2-5s	> 10s	> 30s
错误率	< 0.5%	> 1%	> 5%
缓存命中率	40-70%	< 30%	< 10%
Token 成本/小时	按预算	> 预算 120%	> 预算 200%
Agent 步数 P95	3-8	> 12	> 20
队列等待	< 1s	> 3s	> 10s

总结

LLM 可观测性的核心是 trace_id 贯穿全链路。Log 记录细节，Trace 展示结构，Metric 监控趋势。工具选型首选 Langfuse（LLM 原生 + 开源），配合 Prometheus/Grafana 做指标监控。告警遵循"快速发现、精准定位、自动恢复"原则，关键指标覆盖延迟、错误、成本、缓存、Agent 行为五个维度。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

为什么 LLM 需要专项可观测性？#

三位一体架构#

一、结构化日志#

日志查询示例#

二、分布式链路追踪#

Agent 链路追踪#

三、关键指标#

Grafana 看板配置#

四、工具选型#

Langfuse 集成示例#

五、告警规则#

关键指标速查表#

总结#

加入讨论#

📌 相关推荐

为什么 LLM 需要专项可观测性？

三位一体架构

一、结构化日志

日志查询示例

二、分布式链路追踪

Agent 链路追踪

三、关键指标

Grafana 看板配置

四、工具选型

Langfuse 集成示例

五、告警规则

关键指标速查表

总结

加入讨论