如何统计大模型 Token 用量、调用时延与接口成本

系列：Python + FastAPI 大模型应用基础（第 7 篇）
目标：使用 Python 记录模型调用次数、Token 用量、端到端时延、首 Token 时延和估算成本，并明确哪些数据可以计算、哪些数据不能凭空猜测。

1. 为什么“接口能用”之后必须做统计

一个大模型接口上线后，至少要回答以下问题：

• 今天调用了多少次？
• 成功率和失败率是多少？
• 平均响应时间是多少？
• 用户等待首个 Token 需要多久？
• 输入和输出分别消耗多少 Token？
• 每个业务场景花费多少？
• 哪个模型更快，哪个模型更贵？
• 重试是否导致上游调用次数和费用放大？

如果没有数据，只能得到“感觉有点慢”“好像有点贵”这种无法验证的结论。

从第一性原理看，模型观测链路是：

业务请求进入
    ↓ 记录开始时间
调用外部模型
    ↓ 记录状态码、首 Token 时间、结束时间
解析模型响应
    ↓ 读取服务端返回的 Token 用量
结合当前价格配置
    ↓ 估算本次调用成本
写入结构化指标

2. 需要统计哪些指标

2.1 调用量指标

• 用户请求数；
• 逻辑模型调用数；
• 实际上游 HTTP 请求数；
• 重试次数；
• 成功数和失败数。

一个用户请求可能因为重试产生多次上游 HTTP 请求，所以这几个数字不能混为一谈。

2.2 时延指标

• latency_ms：从开始调用到取得完整结果的总时延；
• ttft_ms：Time To First Token（首 Token 时延）；
• 连接耗时；
• 模型生成耗时；
• 数据库或其他工具调用耗时。

非流式接口通常只能直接测到总时延。流式接口可以额外记录首 Token 到达时间。

2.3 Token 指标

常见兼容式响应可能包含：

{
  "usage": {
    "prompt_tokens": 120,
    "completion_tokens": 80,
    "total_tokens": 200
  }
}

为了让企业内部字段更稳定，本文统一转换为：

• input_tokens：输入 Token 数；
• output_tokens：输出 Token 数；
• total_tokens：总 Token 数。

不同服务商可能使用不同字段，也可能完全不返回用量。不能只根据接口路径推断字段，必须以实际响应和官方文档为准。

2.4 成本指标

最基础的估算公式：

输入成本 = 输入 Token 数 ÷ 1,000,000 × 每百万输入 Token 单价
输出成本 = 输出 Token 数 ÷ 1,000,000 × 每百万输出 Token 单价
总成本 = 输入成本 + 输出成本

但真实计费可能还区分：

• 缓存输入；
• 未缓存输入；
• 批处理；
• 推理 Token；
• 工具调用；
• 图片、音频或视频；
• 不同上下文长度；
• 不同地区和币种。

因此，本文价格全部通过环境变量配置，不写任何具体厂商价格。实际价格必须从当前官方定价页面获取并记录生效日期。

3. 为什么金额必须使用 Decimal

Python 的 float 使用二进制浮点数，某些十进制小数不能被精确表示：

# 结果可能不是人类直觉中的精确 0.3
print(0.1 + 0.2)

成本计算属于金额计算，应该使用 Decimal：

from decimal import Decimal

cost = Decimal("0.1") + Decimal("0.2")
print(cost)  # 精确得到 Decimal('0.3')

价格字符串也要直接传给 Decimal，不要先转换成 float。

4. 创建项目

项目结构：

llm_observability/
├── app/
│   ├── __init__.py
│   ├── metrics.py
│   ├── llm_client.py
│   └── main.py
├── data/
└── requirements.txt

本文使用 Python 3.10 及以上版本。

requirements.txt：

fastapi>=0.115,<1
uvicorn[standard]>=0.30,<1
httpx>=0.27,<1
pydantic>=2.7,<3

安装依赖：

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements.txt

5. 定义统一用量、价格和指标结构

新建 app/metrics.py：

import asyncio
import json
import logging
from dataclasses import asdict, dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TokenUsage:
    """企业内部统一的 Token 用量。"""

    input_tokens: int
    output_tokens: int
    total_tokens: int

    def __post_init__(self) -> None:
        values = (
            self.input_tokens,
            self.output_tokens,
            self.total_tokens,
        )
        if any(value < 0 for value in values):
            raise ValueError("Token 数不能为负数")


@dataclass(frozen=True)
class ModelPricing:
    """某个模型在某个时间点使用的价格配置。"""

    input_per_million: Decimal
    output_per_million: Decimal
    currency: str
    effective_date: str

    @classmethod
    def from_strings(
        cls,
        input_per_million: str,
        output_per_million: str,
        currency: str,
        effective_date: str,
    ) -> "ModelPricing":
        """从字符串创建价格，避免先经过 float 丢失精度。"""

        try:
            input_price = Decimal(input_per_million)
            output_price = Decimal(output_per_million)
        except InvalidOperation as exc:
            raise ValueError("模型价格不是合法十进制数字") from exc

        if input_price < 0 or output_price < 0:
            raise ValueError("模型价格不能为负数")
        if not currency.strip():
            raise ValueError("currency 不能为空")
        if not effective_date.strip():
            raise ValueError("effective_date 不能为空")

        return cls(
            input_per_million=input_price,
            output_per_million=output_price,
            currency=currency.strip().upper(),
            effective_date=effective_date.strip(),
        )


class CostEstimator:
    """根据已确认的 Token 用量和价格估算成本。"""

    TOKENS_PER_MILLION = Decimal("1000000")

    def __init__(self, pricing: ModelPricing) -> None:
        self.pricing = pricing

    def estimate(self, usage: TokenUsage) -> Decimal:
        """返回未强制截断精度的估算成本。"""

        input_cost = (
            Decimal(usage.input_tokens)
            / self.TOKENS_PER_MILLION
            * self.pricing.input_per_million
        )
        output_cost = (
            Decimal(usage.output_tokens)
            / self.TOKENS_PER_MILLION
            * self.pricing.output_per_million
        )
        return input_cost + output_cost


@dataclass(frozen=True)
class CallMetric:
    """一次逻辑模型调用的结构化指标。"""

    request_id: str
    occurred_at: str
    business_scene: str
    provider: str
    model: str
    status: str
    status_code: int | None
    error_code: str | None
    latency_ms: float
    ttft_ms: float | None
    input_tokens: int | None
    output_tokens: int | None
    total_tokens: int | None
    estimated_cost: str | None
    currency: str | None
    price_effective_date: str | None


class JsonlMetricRecorder:
    """将指标追加到 JSON Lines 文件，仅用于本地学习。"""

    def __init__(self, file_path: Path) -> None:
        self.file_path = file_path
        self._lock = asyncio.Lock()

    def _append_line(self, line: str) -> None:
        """同步文件写入函数，由 asyncio.to_thread 调用。"""

        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        with self.file_path.open("a", encoding="utf-8") as file:
            file.write(line)
            file.write("\n")

    async def record(self, metric: CallMetric) -> None:
        """把一个指标对象安全地追加为一行 JSON。"""

        payload: dict[str, Any] = asdict(metric)
        line = json.dumps(
            payload,
            ensure_ascii=False,
            separators=(",", ":"),
        )

        # 锁只能保护当前 Python 进程内的并发写入
        async with self._lock:
            await asyncio.to_thread(self._append_line, line)

    async def safe_record(self, metric: CallMetric) -> None:
        """指标写入失败时记录错误，但不覆盖原始业务结果。"""

        try:
            await self.record(metric)
        except OSError:
            # 观测系统失败不能静默，但也不应让成功的模型回答变失败
            logger.exception("模型调用指标写入失败")

为什么价格需要记录生效日期

模型价格可能变化。如果只保存一个成本数字，却不知道使用的是哪一版价格，以后无法审计和重新计算。

至少应该保存：

• 模型完整标识；
• 价格币种；
• 价格生效日期或价格版本；
• 输入和输出单价；
• 原始 Token 用量。

6. 归一化不同字段的 Token 用量

不同兼容接口可能返回 prompt_tokens/completion_tokens，也可能返回 input_tokens/output_tokens。下面只处理这两组明确字段，不根据文本长度猜测。

把下面函数继续写入 app/metrics.py：

def _non_negative_int(value: object) -> int | None:
    """只接受真正的非负整数，拒绝字符串和布尔值。"""

    # bool 是 int 的子类，因此必须单独排除
    if isinstance(value, bool) or not isinstance(value, int):
        return None
    if value < 0:
        return None
    return value


def normalize_usage(response_data: dict[str, Any]) -> TokenUsage | None:
    """把已知字段映射为统一 TokenUsage；无法确认时返回 None。"""

    raw_usage = response_data.get("usage")
    if not isinstance(raw_usage, dict):
        return None

    input_tokens = _non_negative_int(raw_usage.get("prompt_tokens"))
    if input_tokens is None:
        input_tokens = _non_negative_int(raw_usage.get("input_tokens"))

    output_tokens = _non_negative_int(
        raw_usage.get("completion_tokens")
    )
    if output_tokens is None:
        output_tokens = _non_negative_int(raw_usage.get("output_tokens"))

    total_tokens = _non_negative_int(raw_usage.get("total_tokens"))

    # 输入或输出缺失时，不能可靠计算基础输入/输出成本
    if input_tokens is None or output_tokens is None:
        return None

    # 如果服务没有返回总量，只做确定性的加法，不猜测其他隐藏用量
    if total_tokens is None:
        total_tokens = input_tokens + output_tokens

    return TokenUsage(
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        total_tokens=total_tokens,
    )

如果服务商存在缓存 Token、推理 Token 或其他计费字段，应在 Adapter（适配器）中建立更完整的用量对象，而不是把它们强行塞进基础输入和输出字段。

7. 实现带指标记录的模型客户端

新建 app/llm_client.py：

import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4

import httpx

from app.metrics import (
    CallMetric,
    CostEstimator,
    JsonlMetricRecorder,
    TokenUsage,
    normalize_usage,
)


class LLMCallError(RuntimeError):
    """模型调用统一异常，包含稳定错误代码。"""

    def __init__(self, message: str, error_code: str) -> None:
        super().__init__(message)
        self.error_code = error_code


@dataclass(frozen=True)
class ChatResult:
    """模型回答及本次请求 ID。"""

    request_id: str
    content: str


class InstrumentedLLMClient:
    """在模型调用边界统一记录时延、用量和成本。"""

    def __init__(
        self,
        http_client: httpx.AsyncClient,
        base_url: str,
        api_key: str,
        provider: str,
        model: str,
        cost_estimator: CostEstimator,
        recorder: JsonlMetricRecorder,
    ) -> None:
        self.http_client = http_client
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.provider = provider
        self.model = model
        self.cost_estimator = cost_estimator
        self.recorder = recorder

    async def chat(
        self,
        user_message: str,
        business_scene: str,
    ) -> ChatResult:
        """调用模型，并在 finally 中记录成功或失败指标。"""

        cleaned_message = user_message.strip()
        if not cleaned_message:
            raise ValueError("用户问题不能为空")

        # request_id 由服务端生成，不能直接信任外部用户提交的值
        request_id = str(uuid4())
        started_at = datetime.now(timezone.utc)
        started_counter = time.perf_counter()

        status = "error"
        status_code: int | None = None
        error_code: str | None = None
        usage: TokenUsage | None = None
        estimated_cost: str | None = None

        try:
            response = await self.http_client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "messages": [
                        {
                            "role": "system",
                            "content": "你是一名严谨的 Python 助手。",
                        },
                        {"role": "user", "content": cleaned_message},
                    ],
                    "temperature": 0.2,
                },
            )
            status_code = response.status_code

            if not 200 <= status_code < 300:
                error_code = f"UPSTREAM_HTTP_{status_code}"
                raise LLMCallError(
                    f"模型服务返回异常状态码：{status_code}",
                    error_code=error_code,
                )

            try:
                response_data: dict[str, Any] = response.json()
                content = response_data["choices"][0]["message"][
                    "content"
                ]
            except (ValueError, KeyError, IndexError, TypeError) as exc:
                error_code = "INVALID_UPSTREAM_RESPONSE"
                raise LLMCallError(
                    "模型响应结构不正确",
                    error_code=error_code,
                ) from exc

            if not isinstance(content, str) or not content.strip():
                error_code = "EMPTY_MODEL_CONTENT"
                raise LLMCallError(
                    "模型返回了空内容",
                    error_code=error_code,
                )

            # 用量缺失不会让正确的模型回答失败，但成本会保持 None
            usage = normalize_usage(response_data)
            if usage is not None:
                cost = self.cost_estimator.estimate(usage)
                estimated_cost = str(cost)

            status = "success"
            return ChatResult(
                request_id=request_id,
                content=content.strip(),
            )
        except httpx.TimeoutException as exc:
            error_code = "UPSTREAM_TIMEOUT"
            raise LLMCallError(
                "等待模型服务响应超时",
                error_code=error_code,
            ) from exc
        except httpx.HTTPError as exc:
            error_code = "UPSTREAM_NETWORK_ERROR"
            raise LLMCallError(
                "模型服务发生网络异常",
                error_code=error_code,
            ) from exc
        except LLMCallError as exc:
            # 保留此前设置的错误代码；防止未来分支遗漏赋值
            error_code = error_code or exc.error_code
            raise
        finally:
            latency_ms = round(
                (time.perf_counter() - started_counter) * 1000,
                2,
            )

            metric = CallMetric(
                request_id=request_id,
                occurred_at=started_at.isoformat(),
                business_scene=business_scene,
                provider=self.provider,
                model=self.model,
                status=status,
                status_code=status_code,
                error_code=error_code,
                latency_ms=latency_ms,
                # 当前方法是非流式调用，因此没有 TTFT
                ttft_ms=None,
                input_tokens=(usage.input_tokens if usage else None),
                output_tokens=(usage.output_tokens if usage else None),
                total_tokens=(usage.total_tokens if usage else None),
                estimated_cost=estimated_cost,
                currency=(
                    self.cost_estimator.pricing.currency
                    if estimated_cost is not None
                    else None
                ),
                price_effective_date=(
                    self.cost_estimator.pricing.effective_date
                    if estimated_cost is not None
                    else None
                ),
            )

            # 指标写入失败不会覆盖模型调用的原始成功或失败结果
            await self.recorder.safe_record(metric)

为什么不记录用户问题和模型全文

指标文件的目的不是保存完整会话。用户问题可能包含客户资料、简历、合同或内部数据。默认只记录必要元数据：

• 请求 ID；
• 模型和业务场景；
• 状态码和错误代码；
• 时延；
• Token 用量；
• 估算成本。

如果必须保存会话内容，应进入专门的数据表，执行用户权限、加密、脱敏和数据保留策略，而不是混入普通监控日志。

8. 使用 FastAPI 组装客户端

新建 app/main.py：

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path

import httpx
from fastapi import FastAPI, Request
from pydantic import BaseModel, Field, field_validator

from app.llm_client import InstrumentedLLMClient
from app.metrics import (
    CostEstimator,
    JsonlMetricRecorder,
    ModelPricing,
)


class ChatRequest(BaseModel):
    """聊天接口请求。"""

    user_message: str = Field(min_length=1, max_length=4000)

    @field_validator("user_message")
    @classmethod
    def message_must_not_be_blank(cls, value: str) -> str:
        cleaned_value = value.strip()
        if not cleaned_value:
            raise ValueError("user_message 不能为空白字符串")
        return cleaned_value


class ChatResponse(BaseModel):
    """对外只返回业务必要数据，不直接返回内部成本。"""

    request_id: str
    content: str


def required_env(name: str) -> str:
    """读取必需环境变量。"""

    value = os.getenv(name, "").strip()
    if not value:
        raise RuntimeError(f"缺少环境变量：{name}")
    return value


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """创建共享连接池和带指标的模型客户端。"""

    pricing = ModelPricing.from_strings(
        input_per_million=required_env("MODEL_INPUT_PRICE_PER_1M"),
        output_per_million=required_env("MODEL_OUTPUT_PRICE_PER_1M"),
        currency=required_env("MODEL_PRICE_CURRENCY"),
        effective_date=required_env("MODEL_PRICE_EFFECTIVE_DATE"),
    )

    timeout = httpx.Timeout(
        connect=5.0,
        read=60.0,
        write=10.0,
        pool=5.0,
    )
    http_client = httpx.AsyncClient(timeout=timeout)

    recorder = JsonlMetricRecorder(
        file_path=Path("data/model_call_metrics.jsonl")
    )
    app.state.llm_client = InstrumentedLLMClient(
        http_client=http_client,
        base_url=required_env("LLM_BASE_URL"),
        api_key=required_env("LLM_API_KEY"),
        provider=required_env("LLM_PROVIDER"),
        model=required_env("LLM_MODEL"),
        cost_estimator=CostEstimator(pricing),
        recorder=recorder,
    )

    yield

    await http_client.aclose()


app = FastAPI(
    title="大模型用量与成本统计",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/api/v1/chat", response_model=ChatResponse)
async def chat(
    chat_request: ChatRequest,
    request: Request,
) -> ChatResponse:
    """调用模型；business_scene 由服务端确定，不能由用户随意伪造。"""

    llm_client: InstrumentedLLMClient = request.app.state.llm_client
    result = await llm_client.chat(
        user_message=chat_request.user_message,
        business_scene="python_tutorial",
    )
    return ChatResponse(
        request_id=result.request_id,
        content=result.content,
    )

设置环境变量：

$env:LLM_BASE_URL = "https://替换为模型服务地址/v1"
$env:LLM_API_KEY = "替换为真实密钥"
$env:LLM_PROVIDER = "替换为服务商内部名称"
$env:LLM_MODEL = "替换为真实模型标识"

# 以下价格必须来自当前官方定价，示例不提供虚构数字
$env:MODEL_INPUT_PRICE_PER_1M = "填入每百万输入Token单价"
$env:MODEL_OUTPUT_PRICE_PER_1M = "填入每百万输出Token单价"
$env:MODEL_PRICE_CURRENCY = "填入币种代码，例如CNY或USD"
$env:MODEL_PRICE_EFFECTIVE_DATE = "填入价格生效日期，例如YYYY-MM-DD"

注意：上面的中文占位符不是合法价格，必须替换后才能启动服务。

9. 如何统计首 Token 时延

流式接口需要记录两个时间点：

开始调用时间 t0
首个非空文本到达时间 t1
流结束时间 t2

TTFT = t1 - t0
总时延 = t2 - t0

核心代码：

import time
from collections.abc import AsyncIterator


async def observe_stream(
    upstream: AsyncIterator[str],
) -> AsyncIterator[str]:
    """包装上游文本流，并测量首 Token 和总时延。"""

    started_counter = time.perf_counter()
    first_token_ms: float | None = None

    try:
        async for text_piece in upstream:
            if text_piece and first_token_ms is None:
                first_token_ms = round(
                    (time.perf_counter() - started_counter) * 1000,
                    2,
                )

            yield text_piece
    finally:
        total_latency_ms = round(
            (time.perf_counter() - started_counter) * 1000,
            2,
        )

        # 真实项目应在这里写入 CallMetric
        # 即使客户端中途断开，finally 也会执行清理和指标逻辑
        print(
            {
                "ttft_ms": first_token_ms,
                "latency_ms": total_latency_ms,
            }
        )

不能把第一个 SSE 事件到达时间直接当作首 Token 时间，因为第一个事件可能只包含角色、请求 ID 或其他元数据。应该以第一个非空可展示文本为准。

10. 编写可重复验证的成本单元测试

新建 test_metrics.py：

from decimal import Decimal

from app.metrics import (
    CostEstimator,
    ModelPricing,
    TokenUsage,
    normalize_usage,
)


def test_cost_estimator_uses_decimal() -> None:
    """验证成本公式使用 Decimal，并得到精确结果。"""

    pricing = ModelPricing.from_strings(
        input_per_million="2.00",
        output_per_million="8.00",
        currency="USD",
        effective_date="2099-01-01",
    )
    usage = TokenUsage(
        input_tokens=500_000,
        output_tokens=250_000,
        total_tokens=750_000,
    )

    cost = CostEstimator(pricing).estimate(usage)

    # 输入成本 1，输出成本 2，总成本 3
    assert cost == Decimal("3.000")


def test_normalize_compatible_usage() -> None:
    """验证常见兼容字段能够被归一化。"""

    usage = normalize_usage(
        {
            "usage": {
                "prompt_tokens": 120,
                "completion_tokens": 80,
                "total_tokens": 200,
            }
        }
    )

    assert usage == TokenUsage(
        input_tokens=120,
        output_tokens=80,
        total_tokens=200,
    )


def test_missing_usage_returns_none() -> None:
    """用量缺失时不能通过字符数编造 Token。"""

    assert normalize_usage({"choices": []}) is None

安装并运行测试：

.\.venv\Scripts\python.exe -m pip install "pytest>=8,<9"
.\.venv\Scripts\python.exe -m pytest -q

测试价格使用的是人为构造的数学数据，不是任何真实模型的价格。

11. JSONL 指标文件适合什么场景

JSON Lines 每行保存一个独立 JSON 对象，适合本地学习和排查：

{"request_id":"...","model":"...","status":"success","latency_ms":1234.5,"input_tokens":120,"output_tokens":80}

但本文实现只使用进程内锁，存在以下限制：

• 多个 FastAPI 进程不会共享同一把锁；
• 多台服务器无法统一聚合；
• 文件会不断增大；
• 查询和统计效率有限；
• 磁盘故障可能丢失指标。

生产环境通常将结构化日志发送到日志平台，或使用 Metrics（指标）系统和数据库集中统计。

12. 成本为什么只能叫“估算成本”

即使公式正确，本地结果仍可能和最终账单不同：

• 服务商按不同类别 Token 分开计费；
• 缓存命中价格不同；
• 价格在统计期间发生变化；
• 账单存在折扣、套餐或税费；
• 服务商用量字段与本地映射不完整；
• 重试产生了额外请求；
• 流式中断后本地没有收到最终 usage。

因此：

• 本地记录用于监控趋势和预算预警；
• 最终财务结算应以服务商正式账单为准；
• 应定期对账，检查本地统计与账单差异。

13. 对抗性审查：哪些做法容易产生错误结论

13.1 使用字符数代替 Token 数

字符和 Token 不是固定比例，不同语言和 Tokenizer（分词器）结果不同。字符数只能用于非常粗略的容量预警，不能直接作为账单依据。

13.2 把 total_tokens 全部按输入价计算

输入和输出价格可能不同，必须分别计算。如果响应只提供总量而没有输入、输出拆分，本文选择不估算基础成本。

13.3 硬编码网上找到的价格

价格可能已经过期，也可能对应不同模型版本、地区或计费类型。价格配置必须来自当前官方信息，并记录生效日期。

13.4 只统计成功请求

失败请求和重试也可能产生费用。指标必须记录失败状态、状态码、错误代码和实际上游尝试次数。

13.5 使用普通系统时间计算耗时

系统时间可能因为同步发生跳变。持续时间应该使用 time.perf_counter() 或 time.monotonic()，业务发生时间再使用 UTC 时钟。

13.6 监控失败被静默吞掉

本文不让指标写入故障覆盖模型结果，但会记录错误日志。生产系统还要监控“指标系统自身是否正常”，避免长期无数据却无人发现。

13.7 指标标签包含用户原文

不能把 Prompt、客户姓名或任意请求 ID 作为高基数 Metrics 标签，否则会产生隐私和性能问题。完整请求 ID适合进入日志，聚合指标标签应保持有限集合。

14. 下一步应该监控什么

完成基础统计后，可以按以下维度聚合：

• 模型；
• 服务商；
• 业务场景；
• 成功或失败；
• 错误类型；
• 用户或租户，但要避免暴露身份；
• 时间窗口。

常见统计：

• P50、P95、P99 时延；
• 首 Token P95；
• 每分钟请求量；
• 限流率和熔断率；
• 每次成功回答平均成本；
• 每个业务场景每日成本；
• 重试放大倍数。

平均值可能掩盖少量极慢请求，因此不能只看平均时延。

15. 本篇总结

本文完成了模型用量与成本统计的基础闭环：

• 使用 time.perf_counter() 测量时延；
• 使用首个非空文本计算 TTFT；
• 把外部 Token 字段归一化为内部 TokenUsage；
• 使用 Decimal 计算金额；
• 价格通过配置提供并记录生效日期；
• 用量缺失时不根据字符数编造账单；
• 成功和失败调用都写入结构化指标；
• 指标默认不包含用户问题和模型全文；
• 本地成本只能作为估算，最终以正式账单为准。

下一篇将使用 FastAPI、SQLAlchemy 和 MySQL 持久化多轮对话，并重点处理用户隔离、短事务和模型调用失败后的数据状态。

16. 练习题

1. 增加 upstream_attempts 字段，统计重试放大倍数；
2. 为流式接口记录首 Token 时延；
3. 分别统计成功和失败请求的 P95 时延；
4. 增加缓存输入 Token 字段，但必须以服务商真实字段为依据；
5. 设计价格版本表，而不是直接覆盖旧价格；
6. 模拟指标文件不可写，确认模型调用的原始结果不被覆盖；
7. 对比本地每日估算与服务商正式账单，并记录差异原因。