FastAPI 大模型接口监控：延迟、Token 与错误率

大模型接口上线后，只加一个请求耗时 Histogram 就算完成监控，是很容易走偏的地方。总耗时只能回答“这次请求用了多久”，回答不了“首 Token 等了多久”“流式响应中途断没断”“Token 是 Provider 返回的还是本地估算的”。更完整的链路是：用模块级 Counter + Histogram 记录请求、错误、延迟和 Token，再通过 /metrics 交给 Prometheus 定时拉取，最后用 PromQL 分别计算总延迟、TTFT、Token 增速和错误率。

最容易卡住的地方通常不在 PromQL，而在采集链路本身：/metrics 返回 404，说明路由没有注册、路径被代理改写，或应用监听地址不对；Prometheus Target 显示 up=0，则说明 Prometheus 当前无法从目标地址完成抓取。先用 curl 从应用所在网络直连 /metrics，确认能拿到 Prometheus 文本格式，再检查监听端口、Docker 服务名和代理路径，排查顺序会比反复修改告警规则更快。

先把“慢”和“失败”拆开

一次大模型请求至少包含四类信号：FastAPI 收到请求后的端到端耗时、应用等待 Provider 的耗时、流式响应的首 Token 时间，以及请求最终以什么方式结束。普通非流式接口还可以用 HTTP 状态码概括结果，流式接口却经常在响应头返回 200 后才发生超时、解析失败或客户端断开，因此 HTTP 状态码不能代替最终错误计数。

这些指标可以用 Prometheus 的多维标签、PromQL 和 HTTP 拉取模式承载，但指标边界必须在写代码前定下来。否则后面会出现三个常见结果：总延迟正常但用户觉得慢，错误率很低但流式请求大量中断，或者 Token 看起来在增长，实际上把输入、输出和总量重复累计了。

建议先采用下面这组低基数指标：

指标	类型	建议标签	记录内容
llm_api_requests_total	Counter	provider、model、endpoint、stream、status	每个最终请求只增加一次
llm_api_request_duration_seconds	Histogram	provider、model、endpoint、stream	FastAPI 收到请求到响应体完成的总耗时
llm_api_upstream_duration_seconds	Histogram	provider、model、endpoint	应用等待模型 Provider 的时间
llm_api_tokens_total	Counter	provider、model、direction、source	输入与输出 Token 分开累计
llm_api_ttft_seconds	Histogram	provider、model、endpoint	流式请求的首 Token 时间
llm_api_inter_token_latency_seconds	Histogram	provider、model、endpoint	能确认一事件对应一 Token 时记录
llm_api_chunk_interval_seconds	Histogram	provider、model、endpoint	无法按 Token 对齐时记录数据块间隔
llm_api_errors_total	Counter	provider、model、endpoint、error_type	最终失败原因，重试过程不重复计入
llm_api_token_usage_missing_total	Counter	provider、model、endpoint	Provider 未返回完整 Token usage 的次数

这里的 endpoint 应使用 /v1/chat/completions 这样的路由模板，不能直接使用原始 URL。provider 和 model 也要来自有限配置，例如模型白名单；不能把请求 ID、用户 ID、完整 Prompt、错误文本或未经归一化的模型名称放进标签，否则时间序列数量会迅速膨胀。

status 与 error_type 是两个不同口径。status 记录最终 HTTP 状态，流式响应在中断前已经返回 200 时仍然是 200；error_type 才记录 timeout、rate_limited、stream_interrupted、client_disconnect 等根因。一个最终失败只应进入一次错误计数，内部重试过程可以另做尝试级指标，但不要混入这张最终请求表。

如果只需要通用 HTTP 请求数、请求与响应大小、耗时直方图，prometheus-fastapi-instrumentator 可以自动提供这些指标；Token、TTFT、Provider 错误和流式中断仍然需要自定义 Collector。

在 FastAPI 与模型适配层分别埋点

安装包名是 prometheus-client，代码里的导入名是 prometheus_client。生产环境应让安装命令与项目已有的锁定文件保持一致，而不是每次部署都临时解析最新版本：

pip install fastapi uvicorn prometheus-client

FastAPI 中间件适合记录请求级总耗时和最终状态，模型适配层则适合记录 Provider 调用时间、Token usage 和具体错误。两层分开后，排查时才能区分“FastAPI 自身变慢”和“上游模型变慢”。

先新建一个只负责定义指标的模块。它只会被导入一次，不要把 Counter 或 Histogram 放进路由函数、create_app() 或启动回调里反复创建：

# monitoring.py
from fastapi import Request
from prometheus_client import Counter, Histogram

from llm_adapter import Usage

REQUEST_BUCKETS = (0.5, 1, 2.5, 5, 10, 30, 60, 120, 300)
TTFT_BUCKETS = (0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30)
INTERVAL_BUCKETS = (0.005, 0.01, 0.02, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5)

REQUESTS = Counter(
    "llm_api_requests_total",
    "Final LLM API requests",
    ("provider", "model", "endpoint", "stream", "status"),
)

REQUEST_DURATION = Histogram(
    "llm_api_request_duration_seconds",
    "End-to-end LLM API request duration",
    ("provider", "model", "endpoint", "stream"),
    buckets=REQUEST_BUCKETS,
)

UPSTREAM_DURATION = Histogram(
    "llm_api_upstream_duration_seconds",
    "Time spent waiting for the model provider",
    ("provider", "model", "endpoint"),
    buckets=REQUEST_BUCKETS,
)

TOKENS = Counter(
    "llm_api_tokens_total",
    "LLM tokens by direction and source",
    ("provider", "model", "direction", "source"),
)

TTFT = Histogram(
    "llm_api_ttft_seconds",
    "Time to first content token",
    ("provider", "model", "endpoint"),
    buckets=TTFT_BUCKETS,
)

INTER_TOKEN_LATENCY = Histogram(
    "llm_api_inter_token_latency_seconds",
    "Latency between confirmed single-token events",
    ("provider", "model", "endpoint"),
    buckets=INTERVAL_BUCKETS,
)

CHUNK_INTERVAL = Histogram(
    "llm_api_chunk_interval_seconds",
    "Latency between provider stream chunks",
    ("provider", "model", "endpoint"),
    buckets=INTERVAL_BUCKETS,
)

ERRORS = Counter(
    "llm_api_errors_total",
    "Final LLM API errors",
    ("provider", "model", "endpoint", "error_type"),
)

TOKEN_USAGE_MISSING = Counter(
    "llm_api_token_usage_missing_total",
    "Responses without complete token usage",
    ("provider", "model", "endpoint"),
)


def labels_for(request: Request) -> dict[str, str]:
    route = request.scope.get("route")
    return {
        "provider": getattr(request.state, "llm_provider", "unknown"),
        "model": getattr(request.state, "llm_model", "unknown"),
        "endpoint": getattr(route, "path", request.url.path),
    }


def stream_label(request: Request) -> str:
    return "true" if getattr(request.state, "llm_stream", False) else "false"


def record_error(request: Request, error_type: str) -> None:
    if getattr(request.state, "llm_error_recorded", False):
        return

    request.state.llm_error_recorded = True
    ERRORS.labels(**labels_for(request), error_type=error_type).inc()


def record_usage(request: Request, usage: Usage | None) -> None:
    if usage is None:
        TOKEN_USAGE_MISSING.labels(**labels_for(request)).inc()
        return

    for direction, value in (
        ("input", usage.input_tokens),
        ("output", usage.output_tokens),
    ):
        if value is not None:
            TOKENS.labels(
                **labels_for(request),
                direction=direction,
                source=usage.source,
            ).inc(value)

    if usage.input_tokens is None or usage.output_tokens is None:
        TOKEN_USAGE_MISSING.labels(**labels_for(request)).inc()

上面的桶边界只是便于首次验证的起点，不是通用配置。总延迟要覆盖实际生成耗时，TTFT 和 Token 间隔则需要更细的分辨率；最终应按历史分布和 SLO 调整。Histogram 会导出 _bucket、_sum 和 _count 三类时间序列，后续 p95 查询依赖其中的 _bucket。

不同 Provider 的响应格式不要直接散落在监控代码里。先在适配层归一成下面几类数据，Provider 特有问题也在这里转换成有限的异常类型：

# llm_adapter.py
from collections.abc import AsyncIterator
from dataclasses import dataclass
from typing import Literal, Protocol


@dataclass(frozen=True)
class Usage:
    input_tokens: int | None = None
    output_tokens: int | None = None
    source: Literal["provider", "estimated"] = "provider"


@dataclass
class Completion:
    output: str
    usage: Usage | None = None


@dataclass
class StreamEvent:
    content: str | None = None
    token_count: int | None = None
    payload: bytes | None = None
    usage: Usage | None = None


class ProviderTimeout(Exception):
    pass


class ProviderRateLimited(Exception):
    pass


class ProviderAuthenticationError(Exception):
    pass


class ProviderInvalidResponse(Exception):
    pass


class ModelAdapter(Protocol):
    async def complete(
        self,
        model_id: str,
        messages: list[dict[str, object]],
    ) -> Completion:
        ...

    def stream(
        self,
        model_id: str,
        messages: list[dict[str, object]],
    ) -> AsyncIterator[StreamEvent]:
        ...

token_count=1 只允许在适配层能够确认“一个内容事件正好对应一个 Token”时设置。否则网络分片、多个 Token 合并块或纯元数据事件都会被误算成 Token 间隔。

请求计数最好放在统一出口，而不是 Provider 成功分支。下面的中间件使用单调时钟计时，并包装响应体迭代器，使流式请求的耗时覆盖到最后一个数据块，而不只是返回响应头的时间：

# app.py
import asyncio
import os
import time
from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from llm_adapter import (
    ModelAdapter,
    ProviderAuthenticationError,
    ProviderInvalidResponse,
    ProviderRateLimited,
    ProviderTimeout,
)
from monitoring import (
    REQUESTS,
    REQUEST_DURATION,
    UPSTREAM_DURATION,
    labels_for,
    record_error,
    record_usage,
    stream_label,
)

app = FastAPI()

PROVIDER = os.getenv("LLM_PROVIDER", "unknown")
ALLOWED_MODELS = {
    item.strip()
    for item in os.getenv("LLM_ALLOWED_MODELS", "").split(",")
    if item.strip()
}
MONITORED_ENDPOINTS = {"/v1/chat/completions"}


class ChatRequest(BaseModel):
    model_id: str = Field(alias="model")
    messages: list[dict[str, object]]
    stream: bool = False


def get_adapter(request: Request) -> ModelAdapter:
    # 在现有应用装配代码中设置一次 app.state.adapter。
    return request.app.state.adapter


Adapter = Annotated[ModelAdapter, Depends(get_adapter)]


def error_type_for_status(status_code: int) -> str:
    if status_code == 401:
        return "authentication_failed"
    if status_code == 403:
        return "permission_denied"
    if status_code == 404:
        return "endpoint_not_found"
    if status_code in (408, 504):
        return "timeout"
    if status_code == 429:
        return "rate_limited"
    if 400 <= status_code < 500:
        return "request_rejected"
    return "http_5xx"


def observe_request(request: Request, status: str, duration: float) -> None:
    request_labels = {
        **labels_for(request),
        "stream": stream_label(request),
        "status": status,
    }
    REQUESTS.labels(**request_labels).inc()

    duration_labels = {
        **labels_for(request),
        "stream": stream_label(request),
    }
    REQUEST_DURATION.labels(**duration_labels).observe(duration)


@app.middleware("http")
async def monitor_llm_request(request: Request, call_next):
    if request.url.path.rstrip("/") not in MONITORED_ENDPOINTS:
        return await call_next(request)

    started = time.perf_counter()

    try:
        response = await call_next(request)
    except asyncio.CancelledError:
        record_error(request, "client_disconnect")
        observe_request(
            request,
            "cancelled",
            time.perf_counter() - started,
        )
        raise
    except Exception:
        record_error(request, "internal_error")
        observe_request(request, "500", time.perf_counter() - started)
        raise

    status = str(response.status_code)
    if response.status_code >= 400:
        record_error(request, error_type_for_status(response.status_code))

    original_body = response.body_iterator

    async def observed_body():
        try:
            async for chunk in original_body:
                yield chunk
        except (GeneratorExit, asyncio.CancelledError):
            record_error(request, "client_disconnect")
            raise
        except Exception:
            record_error(request, "stream_interrupted")
            raise
        finally:
            observe_request(
                request,
                status,
                time.perf_counter() - started,
            )

    response.body_iterator = observed_body()
    return response


async def run_complete(
    request: Request,
    adapter: ModelAdapter,
    model_id: str,
    messages: list[dict[str, object]],
) -> dict[str, str]:
    upstream_started = time.perf_counter()

    try:
        result = await adapter.complete(model_id, messages)
    except ProviderTimeout:
        record_error(request, "timeout")
        raise HTTPException(status_code=504, detail="upstream timeout")
    except ProviderRateLimited:
        record_error(request, "rate_limited")
        raise HTTPException(status_code=429, detail="upstream rate limited")
    except ProviderAuthenticationError:
        record_error(request, "upstream_authentication_failed")
        raise HTTPException(status_code=502, detail="upstream authentication failed")
    except ProviderInvalidResponse:
        record_error(request, "invalid_upstream_response")
        raise HTTPException(status_code=502, detail="invalid upstream response")
    except Exception:
        record_error(request, "upstream_error")
        raise HTTPException(status_code=502, detail="upstream error")
    finally:
        UPSTREAM_DURATION.labels(**labels_for(request)).observe(
            time.perf_counter() - upstream_started
        )

    record_usage(request, result.usage)
    return {"content": result.output}

这里没有在成功或失败分支里分别增加请求总数，因为那很容易漏掉其中一条路径。record_error() 带有去重标记，同一个最终失败即使先被适配层记录，随后又经过 HTTP 异常处理，也不会重复增加 llm_api_errors_total。

多 Provider 接入时，应由配置映射决定 provider，再把模型限制在白名单中。不要直接信任请求体里未经校验的 Provider 或模型名称，否则一个非法字符串就能制造一组新的时间序列。

流式请求单独记录首 Token

流式分支从发起 Provider 请求时保存起点，收到第一个“内容事件”时记录 TTFT。角色声明、usage 事件、心跳或其他元数据不能算首 Token；否则测到的只是首字节或首条 SSE 事件时间，不是用户看到第一个内容的时间。

后续事件只有在能够确认对应单个 Token 时才进入 llm_api_inter_token_latency_seconds，否则进入 llm_api_chunk_interval_seconds。流式连接中断、解析失败和客户端断开都要显式记录错误，即使 HTTP 状态已经是 200。

下面的包装器同时处理 Token 间隔、数据块间隔、usage 缺失和流式异常：

# 继续放入 app.py
from monitoring import (
    CHUNK_INTERVAL,
    INTER_TOKEN_LATENCY,
    TTFT,
)


async def run_stream(
    request: Request,
    adapter: ModelAdapter,
    model_id: str,
    messages: list[dict[str, object]],
) -> AsyncIterator[bytes]:
    upstream_started = time.perf_counter()
    first_content_seen = False
    last_content_at: float | None = None
    usage_seen = False

    try:
        async for event in adapter.stream(model_id, messages):
            now = time.perf_counter()

            if event.content is not None:
                if not first_content_seen:
                    TTFT.labels(**labels_for(request)).observe(
                        now - upstream_started
                    )
                    first_content_seen = True
                elif event.token_count == 1 and last_content_at is not None:
                    INTER_TOKEN_LATENCY.labels(**labels_for(request)).observe(
                        now - last_content_at
                    )
                elif last_content_at is not None:
                    CHUNK_INTERVAL.labels(**labels_for(request)).observe(
                        now - last_content_at
                    )

                last_content_at = now

            if event.usage is not None:
                record_usage(request, event.usage)
                usage_seen = True

            if event.payload is not None:
                yield event.payload

    except ProviderTimeout:
        record_error(request, "stream_timeout")
        raise
    except ProviderRateLimited:
        record_error(request, "rate_limited")
        raise
    except ProviderAuthenticationError:
        record_error(request, "upstream_authentication_failed")
        raise
    except ProviderInvalidResponse:
        record_error(request, "invalid_upstream_response")
        raise
    except (GeneratorExit, asyncio.CancelledError):
        record_error(request, "client_disconnect")
        raise
    except Exception:
        record_error(request, "stream_interrupted")
        raise
    finally:
        if not usage_seen:
            record_usage(request, None)

        UPSTREAM_DURATION.labels(**labels_for(request)).observe(
            time.perf_counter() - upstream_started
        )

路由入口只需要区分流式与非流式。model_id 必须先通过白名单校验，再写入指标标签，避免非法模型名进入 Prometheus：

# 继续放入 app.py
@app.post("/v1/chat/completions")
async def chat_completions(
    payload: ChatRequest,
    request: Request,
    adapter: Adapter,
):
    request.state.llm_provider = PROVIDER
    request.state.llm_stream = payload.stream

    if payload.model_id not in ALLOWED_MODELS:
        raise HTTPException(status_code=400, detail="model not allowed")

    request.state.llm_model = payload.model_id

    if payload.stream:
        return StreamingResponse(
            run_stream(
                request,
                adapter,
                payload.model_id,
                payload.messages,
            ),
            media_type="text/event-stream",
        )

    return await run_complete(
        request,
        adapter,
        payload.model_id,
        payload.messages,
    )

要进一步区分 DNS、TLS、连接等待和首字节时间，prometheus_client 不会自动生成这些阶段，需要单独采集并与应用日志对齐：

curl -w "dns:%{time_namelookup} connect:%{time_connect} tls:%{time_appconnect} ttfb:%{time_starttransfer} total:%{time_total}" \
  -o /dev/null -s "https://<provider-host>/v1/models"

这里的 ttfb 是首字节时间，不是首 Token 时间；即使没有鉴权、接口返回 401，网络阶段计时仍然可用于定位连接问题。Provider 内部的 prefill、decode、KV Cache 或 GPU 指标也需要对应层主动暴露，应用无法从总耗时反推出来。

把不同 Provider 的 Token 归到同一口径

适配层应把不同厂商的 usage 字段统一映射为 input 和 output。总 Token 通过 PromQL 对两者求和，不要同时累计 input、output 和 total，否则同一个请求会被重复计算。

source 只保留有限值：provider 表示 Provider 返回的真实 usage，estimated 表示本地 Tokenizer 估算。Provider 没返回 usage 时，应增加 llm_api_token_usage_missing_total，不能把“未知”当作零 Token。流式响应是否在末尾返回完整 usage 取决于具体 Provider 和调用方式，需要检查实际响应，而不是默认所有模型都会返回。

如果业务接受本地估算，估算值只能用于趋势观察，不能直接替代 Provider 账单。尤其是不同模型的分词规则不一致时，同一段文本换模型后估算值可能变化；因此 source="estimated" 必须与 source="provider" 分开查询。

把 /metrics 暴露出来

在应用级路由暴露指标，并返回 Prometheus 要求的文本格式与内容类型：

# 继续放入 app.py
from fastapi.responses import Response
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest


@app.get("/metrics", include_in_schema=False)
async def metrics():
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST,
    )

这个路由需要和指标定义处于同一个默认 Registry。若使用 prometheus-fastapi-instrumentator 的 expose(app)，不要再注册第二个 /metrics 路由，选择其中一种暴露方式即可。

include_in_schema=False 只会隐藏 OpenAPI 文档，不会限制访问。生产环境应通过内网、网关或鉴权规则限制 /metrics，否则模型名称、Provider、错误类型和内部端点都可能暴露。请求 ID、用户 ID、Prompt 和完整错误文本则应该留在日志系统，不要塞进标签。

让 Prometheus 真正拉到数据

最小抓取配置如下：

scrape_configs:
  - job_name: fastapi-llm
    static_configs:
      - targets:
          - 127.0.0.1:8000

Prometheus 与 FastAPI 同处一台主机时，回环地址可以使用；两套服务都运行在 Docker 中时，127.0.0.1 指的是 Prometheus 容器自己，应改成同一个 Docker 网络中的服务名，例如 fastapi:8000。应用还需要监听容器内可达的地址：

uvicorn app:app --host 0.0.0.0 --port 8000

如果请求经过反向代理，Prometheus 的 metrics_path 必须与代理后的真实路径一致，例如外部路径是 /internal/metrics，代理再转发到应用的 /metrics：

scrape_configs:
  - job_name: fastapi-llm
    metrics_path: /internal/metrics
    static_configs:
      - targets:
          - gateway.example.internal

抓取成功后，先在 Prometheus Targets 页面确认目标为 UP，再确认存在 llm_api_* 序列。Prometheus 会按配置定期通过 HTTP 拉取目标指标；Target 显示 UP 只说明抓取成功，不代表业务请求一定正常。

用 PromQL 找到慢请求

总延迟 p95 可以这样查询：

histogram_quantile(
  0.95,
  sum by (le, model) (
    rate(llm_api_request_duration_seconds_bucket[5m])
  )
)

把 0.95 改成 0.50 或 0.99，分别得到 p50 和 p99。若不同 Provider 使用相同的模型名称，应把 provider 一并加入 sum by (...)。

流式首 Token p95 使用对应的 Histogram：

histogram_quantile(
  0.95,
  sum by (le, model) (
    rate(llm_api_ttft_seconds_bucket[5m])
  )
)

[5m] 只是查询窗口示例。请求量较低时应适当拉长窗口，否则 rate() 中可能没有足够样本。桶边界也要覆盖实际延迟分布，不能直接沿用普通短 HTTP 接口的默认桶。

先判断“有样本但慢”，还是“根本没有样本”，可以使用计数序列：

sum by (model) (
  increase(llm_api_request_duration_seconds_count[5m])
)

如果 _count 有数据而 p95 为空，再检查是否使用了 _bucket、聚合时是否保留了 le，以及桶边界是否合理。总延迟 p95 与上游耗时 p95 还可以对照查看，前者包含 FastAPI、序列化和网络开销，后者更接近 Provider 等待时间：

histogram_quantile(
  0.95,
  sum by (le, model) (
    rate(llm_api_upstream_duration_seconds_bucket[5m])
  )
)

用 PromQL 查错误和 Token

最终错误率使用错误计数除以最终请求计数：

sum by (model) (
  rate(llm_api_errors_total[5m])
)
/
sum by (model) (
  rate(llm_api_requests_total[5m])
)

如果只看上游根因，可以筛选错误类型：

sum by (model) (
  rate(llm_api_errors_total{error_type=~"timeout|rate_limited|upstream_error|upstream_authentication_failed|invalid_upstream_response|stream_timeout|stream_interrupted"}[5m])
)
/
sum by (model) (
  rate(llm_api_requests_total[5m])
)

HTTP 状态错误率则单独计算：

sum by (model) (
  rate(llm_api_requests_total{status=~"5.."}[5m])
)
/
sum by (model) (
  rate(llm_api_requests_total[5m])
)

这两个错误率可能指向同一批请求，不能直接相加：前者按最终根因归类，后者按 HTTP 状态归类。分子和分母必须使用相同的模型筛选、时间窗口和统计范围。

流式中断率也不能用 HTTP 5xx 代替，因为中断发生时状态码可能已经是 200：

sum by (model) (
  rate(llm_api_errors_total{error_type="stream_interrupted"}[5m])
)
/
sum by (model) (
  rate(llm_api_requests_total{stream="true"}[5m])
)

按模型、输入/输出方向和来源查看 Token 增速：

sum by (model, direction, source) (
  rate(llm_api_tokens_total[5m])
)

这里得到的是每秒 Token 增速。要看一段时间内累计消耗，可以使用：

sum by (model, direction, source) (
  increase(llm_api_tokens_total[1h])
)

Provider 未返回 usage 的比例可以这样观察：

sum by (model) (
  rate(llm_api_token_usage_missing_total[5m])
)
/
sum by (model) (
  rate(llm_api_requests_total[5m])
)

按报错症状查采集链路

先从与 Prometheus 相同的网络直接检查应用，而不是先改 PromQL：

curl -i http://127.0.0.1:8000/metrics

如果应用在 Docker 网络中，这一步也要在对应容器或网络内执行。随后按症状定位：

搜索到的症状	优先检查	处理方向
/metrics 返回 404	路由是否注册、代理是否改写路径	先直连应用，再检查代理的路径转发规则
Prometheus Target 显示 up=0	监听地址、端口、防火墙、Docker 服务名	确认 Prometheus 所在网络能够访问目标
Target 为 UP 但查询无数据	指标名称、Job 与实例标签、是否已有请求	查询完整导出名称，例如 llm_api_requests_total
抓取返回 HTML 或内容类型错误	登录页、404 页、代理首页是否覆盖了 /metrics	返回 generate_latest() 和 CONTENT_TYPE_LATEST
出现 Duplicated timeseries in CollectorRegistry	Collector 是否被多次创建、模块是否以不同路径重复导入	指标定义放到单一模块，不要在启动回调中重复创建
错误率始终为零	异常是否绕过适配层、流式中断是否发生在 200 之后	在统一异常出口和流式异常分支增加错误计数
Token 指标为零或缺失	usage 字段映射、流式末尾事件、缺失计数器	区分真实零 Token、未知 usage 与本地估算
p95 查询为空	是否使用 _bucket、是否有样本、桶是否合适	先查 _count，再检查 Histogram 分桶
时间序列数量暴涨	标签是否含请求 ID、原始 URL、Prompt 或错误文本	删除高基数字段，把明细留在日志中
多 worker 下数据不完整	当前客户端版本与多进程采集方式	按锁定版本选择对应的多进程方案

多 worker 场景需要单独处理。Prometheus 每次抓取同一个 IP 和端口时，通常只会命中其中一个 worker；如果每个 worker 都在自己的进程内维护独立计数，抓取结果就会缺少其他 worker 的数据。先确认项目实际使用的版本：

python -m pip show prometheus-client

再按该版本的多进程文档配置共享目录和 MultiProcessCollector，不能把不同版本的目录结构、环境变量和 Collector 写法混用。验证阶段可以先使用单 worker，确认指标口径正确后再切回多 worker。

把关键查询变成告警

告警阈值应来自实际 SLO、历史分布和容量测试。下面给出规则形态，尖括号中的阈值必须替换成项目自己的数值，不能原样加载；for: 10m 只是持续时间示例，实际值取决于业务能容忍多久：

groups:
  - name: fastapi-llm
    rules:
      - alert: LlmTotalP95High
        expr: |
          histogram_quantile(
            0.95,
            sum by (le, provider, model) (
              rate(llm_api_request_duration_seconds_bucket[5m])
            )
          ) > <TOTAL_P95_SECONDS>
        for: 10m

      - alert: LlmTtftP95High
        expr: |
          histogram_quantile(
            0.95,
            sum by (le, provider, model) (
              rate(llm_api_ttft_seconds_bucket[5m])
            )
          ) > <TTFT_P95_SECONDS>
        for: 10m

      - alert: LlmFinalErrorRateHigh
        expr: |
          (
            sum by (provider, model) (
              rate(llm_api_errors_total[5m])
            )
            /
            sum by (provider, model) (
              rate(llm_api_requests_total[5m])
            )
          ) > <FINAL_ERROR_RATIO>
        for: 10m

      - alert: LlmStreamInterruptedRateHigh
        expr: |
          (
            sum by (provider, model) (
              rate(llm_api_errors_total{error_type="stream_interrupted"}[5m])
            )
            /
            sum by (provider, model) (
              rate(llm_api_requests_total{stream="true"}[5m])
            )
          ) > <STREAM_INTERRUPTION_RATIO>
        for: 10m

      - alert: LlmTokenUsageMissingHigh
        expr: |
          (
            sum by (provider, model) (
              rate(llm_api_token_usage_missing_total[5m])
            )
            /
            sum by (provider, model) (
              rate(llm_api_requests_total[5m])
            )
          ) > <MISSING_USAGE_RATIO>
        for: 10m

      - alert: LlmTokenCounterStopped
        expr: |
          (
            sum by (provider, model) (
              rate(llm_api_tokens_total[10m])
            ) == 0
          )
          and on(provider, model)
          (
            sum by (provider, model) (
              rate(llm_api_requests_total[10m])
            ) > 0
          )
        for: 10m

Token 突然归零既可能是业务流量下降，也可能是 usage 字段或适配器发生回归，因此必须和请求量一起判断。PromQL 稳定后再接入 Grafana 仪表盘；OpenTelemetry、Prometheus、Grafana 的全链路追踪属于另一层扩展，不必为了画图先替换掉这组基础指标。

最后做一次可控验证

先确认应用直接暴露了本业务指标：

curl -sS http://127.0.0.1:8000/metrics | grep '^llm_api_'

然后在测试环境发送一个非流式请求：

curl -sS -X POST http://127.0.0.1:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"<allowlisted-model-id>","messages":[{"role":"user","content":"ping"}],"stream":false}'

再发送一个确定不在白名单中的模型名，验证 4xx 路径：

curl -i -X POST http://127.0.0.1:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"<not-in-allowlist>","messages":[{"role":"user","content":"ping"}],"stream":false}'

最后发送一个可以观察到首 Token 的流式请求，并关闭客户端缓冲：

curl -sS -N -X POST http://127.0.0.1:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"<allowlisted-model-id>","messages":[{"role":"user","content":"ping"}],"stream":true}'

如果 Provider 返回 usage，检查 Token 计数是否增长；如果没有返回，应看到 llm_api_token_usage_missing_total 增长，而不是把未知值伪装成零。随后在 Prometheus 查询：

increase(llm_api_requests_total[5m])

increase(llm_api_errors_total[5m])

increase(llm_api_tokens_total[5m])

验证完成的标志不是终端里“看起来有数据”，而是 Targets 页面显示 UP，并且请求、错误、耗时直方图和 Token 指标都能在 PromQL 中查到刚才那次请求带来的增量。