大模型应用的可观测性工程：构建LLM系统的监控、追踪与告警体系

当大语言模型从实验项目走向生产系统，它就不再只是一个”调用API拿到文本”的简单组件，而是一个涉及提示词管理、上下文编排、多轮对话状态、工具调用链路、向量检索质量和成本控制的复杂分布式子系统。传统应用的监控手段——CPU利用率、QPS、错误率——远远不够描述一个LLM系统的真实健康状态。本文从工程实战角度，系统拆解LLM应用可观测性的三层体系：指标（Metrics）、追踪（Tracing）和日志（Logs），并给出可落地的代码与配置。

一、为什么LLM应用需要全新的可观测性策略

传统Web应用的监控以请求为中心：一个HTTP请求进来，经过若干中间件，访问数据库，返回响应。整条链路的延迟、错误率、吞吐量构成了SRE的黄金信号。但LLM应用的调用链路有本质不同：

• 非确定性输出：同一个输入可能产生不同输出，传统的”请求-响应”模型无法描述输出质量。
• 延迟分布长尾严重：Token生成是流式的，首token延迟（TTFT）和整体生成延迟（E2E latency）差异巨大，P99可能是P50的5-10倍。
• 成本与Token强相关：一次请求的成本不取决于请求体大小，而取决于输入输出Token数量，传统按请求计费的监控模型失效。
• 质量退化隐蔽：模型版本更新、提示词修改、检索索引漂移都可能导致输出质量下降，但HTTP状态码仍然是200。
• 多跳调用链：一个用户请求可能触发RAG检索、工具调用、子Agent委派，形成深层嵌套的调用树。

这意味着，LLM应用的可观测性必须从”监控基础设施”升级为”监控语义质量”——不仅要知道系统在不在运行，还要知道系统运行得好不好。

二、指标层：从Token经济学到质量信号

2.1 核心指标定义

一个成熟的LLM监控体系至少需要以下四类指标：

2.2 用OpenTelemetry采集LLM指标

OpenTelemetry（OTel）已成为云原生可观测性的事实标准。对于LLM应用，我们可以通过自定义Meter来采集Token级指标。以下是一个基于Python的采集示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from opentelemetry import metrics

from opentelemetry.sdk.metrics import MeterProvider

from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader

from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

from opentelemetry.sdk.resources import Resource

import time, tiktoken



# 初始化OTel Meter

resource = Resource.create({"service.name": "llm-gateway"})

exporter = OTLPMetricExporter(endpoint="http://otel-collector:4317")

reader = PeriodicExportingMetricReader(exporter, export_interval_millis=15000)

meter_provider = MeterProvider(resource=resource, metric_readers=[reader])

metrics.set_meter_provider(meter_provider)

meter = metrics.get_meter("llm.observability")



# 定义指标

ttft_histogram = meter.create_histogram(

    name="llm.ttft.seconds",

    description="Time to first token in seconds",

    unit="s"

)

token_counter = meter.create_counter(

    name="llm.tokens.total",

    description="Total tokens consumed",

)

cost_counter = meter.create_counter(

    name="llm.cost.usd",

    description="LLM API cost in USD",

    unit="USD"

)



def instrument_llm_call(model, prompt, response_gen, provider):

    # 包装一个流式LLM调用，自动采集指标

    encoder = tiktoken.encoding_for_model(model) if "gpt" in model else None

    input_tokens = len(encoder.encode(prompt)) if encoder else 0

    token_counter.add(input_tokens, {"model": model, "type": "input", "provider": provider})



    first_token_time = None

    start = time.monotonic()

    output_tokens = 0

    full_response = []



    for chunk in response_gen:

        if first_token_time is None:

            first_token_time = time.monotonic() - start

            ttft_histogram.record(first_token_time, {"model": model, "provider": provider})

        text = chunk.choices[0].delta.content or ""

        if text:

            output_tokens += 1

            full_response.append(text)



    e2e = time.monotonic() - start

    token_counter.add(output_tokens, {"model": model, "type": "output", "provider": provider})



    # 按模型定价计算成本

    pricing = {"gpt-4o": (0.005, 0.015), "claude-sonnet-4": (0.003, 0.015)}

    in_price, out_price = pricing.get(model, (0, 0))

    cost = (input_tokens * in_price + output_tokens * out_price) / 1000

    cost_counter.add(cost, {"model": model, "provider": provider})



    return "".join(full_response), {

        "ttft": first_token_time, "e2e": e2e,

        "input_tokens": input_tokens, "output_tokens": output_tokens,

        "cost": cost

    }

这段代码的核心思路是将OpenTelemetry的Histogram和Counter嵌入到LLM调用的流式迭代器中。TTFT通过记录第一个chunk到达的时间差来计算，这对流式响应至关重要——它直接决定了用户体验的”首屏速度”。

2.3 Prometheus告警规则

将OTel指标导出到Prometheus后，可以配置关键告警规则：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# prometheus/rules/llm-alerts.yml

groups:

  - name: llm-observability

    rules:

      # TTFT超过3秒，用户感知明显卡顿

      - alert: LLMHighTTFT

        expr: |

          histogram_quantile(0.95,

            rate(llm_ttft_seconds_bucket[5m])) > 3

        for: 10m

        labels:

          severity: warning

        annotations:

          summary: "LLM首Token延迟P95超过3秒"



      # 错误率超过5%

      - alert: LLMHighErrorRate

        expr: |

          rate(llm_request_total{status="error"}[5m]) /

          rate(llm_request_total[5m]) > 0.05

        for: 5m

        labels:

          severity: critical

        annotations:

          summary: "LLM请求错误率超过5%"



      # 日成本超过预算阈值

      - alert: LLMCostBudgetExceeded

        expr: |

          sum(rate(llm_cost_usd_total[1h])) * 24 > 500

        for: 30m

        labels:

          severity: warning

        annotations:

          summary: "LLM日均成本预计超过500美元预算"

三、追踪层：还原多跳调用树

LLM应用的调用链通常远比传统微服务复杂。一个RAG请求可能涉及：用户请求 -> 意图分类 -> 查询改写 -> 向量检索 -> 重排序 -> 上下文组装 -> LLM生成 -> 工具调用 -> 二次生成 -> 响应后处理。如果其中任何一环出问题，没有分布式追踪就等于盲人摸象。

3.1 用OTel Span构建LLM调用树

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from opentelemetry import trace

from opentelemetry.trace import SpanKind



tracer = trace.get_tracer("llm.pipeline")



def rag_pipeline(user_query):

    # 带完整追踪的RAG流水线

    with tracer.start_as_current_span("rag_pipeline", kind=SpanKind.INTERNAL) as root:

        root.set_attribute("user.query", user_query)

        root.set_attribute("user.query.length", len(user_query))



        # 1. 意图分类

        with tracer.start_as_current_span("intent_classification") as span:

            intent = classify_intent(user_query)

            span.set_attribute("intent.result", intent)

            span.set_attribute("intent.confidence", 0.92)



        # 2. 查询改写

        with tracer.start_as_current_span("query_rewrite") as span:

            rewritten = rewrite_query(user_query, intent)

            span.set_attribute("query.original", user_query)

            span.set_attribute("query.rewritten", rewritten)



        # 3. 向量检索

        with tracer.start_as_current_span("vector_search", kind=SpanKind.CLIENT) as span:

            results = vector_db.search(rewritten, top_k=10)

            span.set_attribute("search.results_count", len(results))

            span.set_attribute("search.top_score", results[0].score if results else 0)

            span.set_attribute("search.latency_ms", 45)



        # 4. 重排序

        with tracer.start_as_current_span("rerank") as span:

            reranked = reranker.rerank(rewritten, results, top_k=3)

            span.set_attribute("rerank.input_count", len(results))

            span.set_attribute("rerank.output_count", len(reranked))



        # 5. LLM生成

        with tracer.start_as_current_span("llm_generation", kind=SpanKind.CLIENT) as span:

            context = assemble_context(reranked)

            response, call_metrics = instrument_llm_call(

                model="gpt-4o", prompt=context + user_query,

                response_gen=stream_completion(context, user_query),

                provider="openai"

            )

            span.set_attribute("llm.input_tokens", call_metrics["input_tokens"])

            span.set_attribute("llm.output_tokens", call_metrics["output_tokens"])

            span.set_attribute("llm.ttft_seconds", call_metrics["ttft"])

            span.set_attribute("llm.cost_usd", call_metrics["cost"])

            span.set_attribute("llm.context_docs", len(reranked))



        return response

在Jaeger或Tempo中查看这条Trace时，你会看到一个完整的瀑布图：每个Span的持续时间、属性、状态都清晰可见。当用户反馈”回答质量差”时，你可以快速定位是检索召回率低（top_score偏低）、重排序失效、还是LLM本身幻觉——而不是在日志海中捞针。

3.2 Span属性的最佳实践

LLM追踪的Span属性设计是决定可观测性质量的关键。以下是推荐的属性命名规范：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 标准化LLM Span属性（遵循OpenTelemetry GenAI语义约定）

span.set_attribute("gen_ai.system", "openai")

span.set_attribute("gen_ai.request.model", "gpt-4o")

span.set_attribute("gen_ai.request.temperature", 0.7)

span.set_attribute("gen_ai.request.max_tokens", 4096)

span.set_attribute("gen_ai.usage.prompt_tokens", input_tokens)

span.set_attribute("gen_ai.usage.completion_tokens", output_tokens)

span.set_attribute("gen_ai.response.id", response_id)



# RAG专用属性

span.set_attribute("rag.retrieval.top_k", 10)

span.set_attribute("rag.retrieval.score", 0.87)

span.set_attribute("rag.context.doc_ids", "doc1,doc2,doc3")

span.set_attribute("rag.context.total_chars", 12500)

遵循OpenTelemetry GenAI语义约定（gen_ai.*前缀）能确保你的追踪数据与生态工具兼容，包括Langfuse、Arize Phoenix、Datadog LLM Observability等。

四、质量评估层：超越基础设施监控

指标和追踪回答了”系统在做什么”，但没回答”系统做得好不好”。质量评估是LLM可观测性区别于传统APM的核心分水岭。

4.1 在线评估 vs 离线评估

4.2 实时质量门控

在LLM响应返回给用户之前，可以插入轻量级质量检查。以下是一个基于规则的实时评估器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import re

from dataclasses import dataclass



@dataclass

class QualityResult:

    passed: bool

    score: float

    reasons: list



class LLMQualityGate:

    def __init__(self):

        self.rules = [

            self._check_length,

            self._check_hallucination_markers,

            self._check_citation_coverage,

            self._check_language_consistency,

        ]



    def evaluate(self, response, context_docs, query):

        results = [rule(response, context_docs, query) for rule in self.rules]

        scores = [r[0] for r in results]

        reasons = [r[1] for r in results if not r[0]]

        overall = sum(scores) / len(scores)

        return QualityResult(passed=overall >= 0.7, score=overall, reasons=reasons)



    def _check_length(self, response, context_docs, query):

        # 过短响应可能信息不足

        if len(response) < 50:

            return (0.0, "响应过短，可能信息不足")

        return (1.0, "")



    def _check_hallucination_markers(self, response, context_docs, query):

        # 检测幻觉常见标记

        markers = ["根据我的知识", "据我所知", "通常情况下", "一般来说"]

        matches = sum(response.count(m) for m in markers)

        if matches > 2:

            return (0.3, "检测到%d个幻觉标记" % matches)

        return (1.0, "")



    def _check_citation_coverage(self, response, context_docs, query):

        # 检查引用是否覆盖了检索文档

        if not context_docs:

            return (0.5, "无上下文文档，无法验证引用")

        cited = sum(1 for doc in context_docs if doc.id in response)

        coverage = cited / len(context_docs)

        if coverage < 0.3:

            return (0.4, "引用覆盖率仅%.0f%%" % (coverage * 100))

        return (1.0, "")



    def _check_language_consistency(self, response, context_docs, query):

        # 语言一致性检查

        def is_chinese(text):

            return any("\u4e00" <= c <= "\u9fff" for c in text)

        if is_chinese(query) != is_chinese(response):

            return (0.2, "响应语言与查询语言不一致")

        return (1.0, "")

这个质量门控可以嵌入到追踪Span中，将质量分数作为Span属性记录，并在低于阈值时触发降级策略（如切换模型、追加上下文、返回兜底响应）。

4.3 LLM-as-Judge异步评估

对于无法实时运行的深度质量评估，可以使用LLM-as-Judge模式异步处理。以下是一个基于批处理的评估管线：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import asyncio, json

from openai import AsyncOpenAI



client = AsyncOpenAI()



JUDGE_PROMPT = """你是一个严格的技术内容质量评估专家。请对以下AI助手的回答进行评分。



用户问题：{query}

AI回答：{response}

参考文档：{context}



请从以下维度各打1-5分：

1. 准确性：回答是否与参考文档一致，有无事实错误

2. 完整性：是否充分回答了用户的问题

3. 引用准确性：引用的信息是否确实来自参考文档

4. 结构清晰度：回答的逻辑结构是否清晰易读



输出JSON格式如下：

{"accuracy": N, "completeness": N, "citation": N, "clarity": N, "overall": N, "reasoning": "..."}"""



async def judge_response(query, response, context):

    prompt = JUDGE_PROMPT.format(query=query, response=response, context=context)

    result = await client.chat.completions.create(

        model="gpt-4o-mini",

        messages=[{"role": "user", "content": prompt}],

        response_format={"type": "json_object"},

        temperature=0,

    )

    return json.loads(result.choices[0].message.content)



async def batch_evaluate(session_logs, batch_size=20):

    # 批量评估历史会话，用于趋势分析

    semaphore = asyncio.Semaphore(batch_size)



    async def eval_one(log):

        async with semaphore:

            return await judge_response(log["query"], log["response"], log["context"])



    results = await asyncio.gather(*[eval_one(log) for log in session_logs])

    n = len(results)

    return {

        "accuracy": sum(r["accuracy"] for r in results) / n,

        "completeness": sum(r["completeness"] for r in results) / n,

        "citation": sum(r["citation"] for r in results) / n,

        "clarity": sum(r["clarity"] for r in results) / n,

    }

将评估结果写入时序数据库（如Prometheus或InfluxDB），配合Grafana可以绘制质量趋势图。当某天的平均准确性分数突然下降0.5分，就说明可能有模型版本变更、提示词回归或知识库污染需要排查。

五、日志层：结构化LLM调用日志

LLM应用的日志不能只记录”请求来了、响应发了”，必须结构化记录完整的调用上下文，才能支持事后排查和审计。推荐使用JSON结构化日志：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import structlog, logging



logging.basicConfig(format="%(message)s")

structlog.configure(

    processors=[

        structlog.processors.TimeStamper(fmt="iso"),

        structlog.processors.add_log_level,

        structlog.processors.JSONRenderer(),

    ]

)

log = structlog.get_logger("llm.gateway")



def log_llm_call(trace_id, span_id, **kwargs):

    # 结构化记录LLM调用

    log.info("llm.call.complete",

        trace_id=trace_id,

        span_id=span_id,

        model=kwargs.get("model"),

        provider=kwargs.get("provider"),

        input_tokens=kwargs.get("input_tokens"),

        output_tokens=kwargs.get("output_tokens"),

        ttft_ms=kwargs.get("ttft_ms"),

        e2e_ms=kwargs.get("e2e_ms"),

        cost_usd=kwargs.get("cost"),

        prompt_hash=hash(kwargs.get("prompt", "")),

        response_quality_score=kwargs.get("quality_score"),

        fallback_triggered=kwargs.get("fallback", False),

    )

注意几个关键设计：第一，不记录原始Prompt和响应全文到日志系统——它们可能包含敏感用户数据，只记录hash和长度。全文应存储在专用的事务存储中（如Langfuse或S3），通过trace_id关联。第二，每条日志都带上trace_id和span_id，实现日志与追踪的交叉关联。第三，记录quality_score和fallback_triggered等业务语义字段，而不仅仅是技术指标。

六、成本可观测性：Token就是金钱

LLM应用的成本可观测性是很多团队忽视直到账单爆炸才重视的维度。一个良好的成本监控体系应该做到：

• 实时成本看板：每小时刷新一次的Token消耗和美元成本，按模型、按租户、按功能模块维度聚合。
• 成本异常检测：当某用户的Token消耗突然增长10倍时自动告警，可能是Prompt注入攻击或循环调用bug。
• 预算熔断：在网关层实现per-tenant的Token预算硬限制，超过预算自动降级到更便宜的模型。
• 成本归因：将每个用户请求的LLM成本精确归因到具体功能点，支持按功能计费或成本优化决策。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from dataclasses import dataclass

from collections import defaultdict

import time



@dataclass

class TokenBudget:

    tenant_id: str

    daily_limit: int

    hourly_limit: int



class TokenBudgetGuard:

    def __init__(self):

        self.hourly_usage = defaultdict(int)

        self.daily_usage = defaultdict(int)

        self.fallback_model = "gpt-4o-mini"



    def check_and_select_model(self, tenant_id, requested_model, estimated_tokens):

        hour_key = (tenant_id, int(time.time()) // 3600)

        day_key = (tenant_id, int(time.time()) // 86400)



        if self.hourly_usage[hour_key] + estimated_tokens > self._get_hourly_limit(tenant_id):

            log.warning("hourly_budget_exceeded",

                tenant=tenant_id, usage=self.hourly_usage[hour_key])

            return self.fallback_model



        if self.daily_usage[day_key] + estimated_tokens > self._get_daily_limit(tenant_id):

            log.error("daily_budget_exceeded",

                tenant=tenant_id, usage=self.daily_usage[day_key])

            raise BudgetExceededError("Daily budget exceeded for " + tenant_id)



        self.hourly_usage[hour_key] += estimated_tokens

        self.daily_usage[day_key] += estimated_tokens

        return requested_model



    def reconcile(self, tenant_id, estimated, actual):

        # 响应返回后修正预估Token与实际Token的差值

        diff = actual - estimated

        hour_key = (tenant_id, int(time.time()) // 3600)

        day_key = (tenant_id, int(time.time()) // 86400)

        self.hourly_usage[hour_key] += diff

        self.daily_usage[day_key] += diff

七、可观测性架构总览与工具选型

将以上各层整合起来，一个生产级LLM应用的可观测性架构如下：

• 采集层：OpenTelemetry SDK嵌入应用代码，统一采集Metrics、Traces、Logs。
• 传输层：OTel Collector作为统一网关，支持缓冲、重试、采样和敏感数据脱敏。
• 存储层：Prometheus存指标，Tempo/Jaeger存追踪，Loki/ELK存日志，Langfuse存LLM专用数据。
• 可视化层：Grafana统一看板，集成Prometheus、Tempo、Loki数据源；Langfuse提供LLM专用分析界面。
• 告警层：Alertmanager处理Prometheus告警路由，PagerDuty或Slack接收通知。

在工具选型上，如果你的LLM应用以RAG和Agent为主，Langfuse是性价比最高的LLM专用可观测性平台，它原生支持OpenAI、Anthropic、LangChain等主流框架的自动追踪，并内置了LLM-as-Judge评估能力。如果你已有Grafana技术栈，则直接用OTel + Prometheus + Tempo + Loki是最自然的选择。对于大规模部署，建议两者结合：Langfuse做LLM语义层分析，Grafana做基础设施层监控。

结语：可观测性是LLM应用上线的前提

很多团队把可观测性当作”上线之后再补”的nice-to-have，这在传统应用中或许勉强可行，但在LLM应用中是致命的。大模型的非确定性意味着你无法通过测试覆盖所有行为路径，唯一可靠的保障手段就是持续的可观测性——在真实流量中发现问题、定位问题、修复问题。

从指标采集到追踪链路、从质量评估到成本熔断，本文给出的方案并非空中楼阁，而是在真实生产环境中验证过的工程实践。可观测性的投入看起来是”额外成本”，但它能在一次模型版本回归中帮你省下数小时的排查时间，在一次Prompt注入攻击中帮你止损数千美元的Token消耗。在LLM从demo走向production的路上，可观测性不是可选项，而是必选项。