AI 应用最小可观测性：没有日志就没有工程化

{
  "ts": "2026-07-18T14:02:11+08:00",
  "request_id": "9f3c...",
  "route": "chat",
  "model": "auto",
  "upstream": "openai/gpt-4o-mini",
  "latency_ms": 842,
  "ttft_ms": 210,
  "prompt_tokens": 312,
  "completion_tokens": 88,
  "cost_usd": 0.00012,
  "status": "ok",
  "cache_hit": false,
  "tool_calls": 0,
  "key_alias": "web-prod"
}

# llm_observe.py
from __future__ import annotations
import json, time, uuid, logging
from dataclasses import asdict, dataclass
from typing import Any
from openai import OpenAI

log = logging.getLogger("llm")
logging.basicConfig(level=logging.INFO, format="%(message)s")

@dataclass
class CallMetrics:
    request_id: str
    model: str
    upstream: str
    latency_ms: int
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    status: str
    error: str | None = None

PRICE_IN = {"gpt-4o-mini": 0.15 / 1_000_000}
PRICE_OUT = {"gpt-4o-mini": 0.60 / 1_000_000}

def estimate(model: str, pt: int, ct: int) -> float:
    return pt * PRICE_IN.get(model, 0) + ct * PRICE_OUT.get(model, 0)

def chat(client: OpenAI, model: str, messages: list[dict[str, Any]], **kw) -> str:
    rid = str(uuid.uuid4())
    t0 = time.perf_counter()
    try:
        r = client.chat.completions.create(model=model, messages=messages, **kw)
        pt = r.usage.prompt_tokens if r.usage else 0
        ct = r.usage.completion_tokens if r.usage else 0
        m = CallMetrics(
            request_id=rid,
            model=model,
            upstream=getattr(r, "model", model) or model,
            latency_ms=int((time.perf_counter() - t0) * 1000),
            prompt_tokens=pt,
            completion_tokens=ct,
            cost_usd=estimate("gpt-4o-mini", pt, ct),
            status="ok",
        )
        log.info(json.dumps(asdict(m), ensure_ascii=False))
        return r.choices[0].message.content or ""
    except Exception as e:  # noqa: BLE001
        m = CallMetrics(
            request_id=rid,
            model=model,
            upstream="unknown",
            latency_ms=int((time.perf_counter() - t0) * 1000),
            prompt_tokens=0,
            completion_tokens=0,
            cost_usd=0.0,
            status="error",
            error=str(e)[:300],
        )
        log.info(json.dumps(asdict(m), ensure_ascii=False))
        raise

# 假设日志一行一个 JSON
# 今日错误率
grep '"ts":"2026-07-18' llm.log | grep -c '"status":"error"'

# 按 upstream 聚合延迟（粗暴版）
grep '"ts":"2026-07-18' llm.log \
  | jq -r 'select(.status=="ok") | [.upstream, .latency_ms] | @tsv' \
  | awk '{s[$1]+=$2; n[$1]++} END{for(k in s) printf "%s avg=%.0fms n=%d\n",k,s[k]/n[k],n[k]}'