Cost, Latency & Token Economics for LLM Apps

from dataclasses import dataclass
from typing import Callable

@dataclass
class TokenBudget:
    """Theo dõi token usage theo task. Emit metric + cảnh báo khi vượt 80%."""
    task_id: str
    max_input: int
    max_output: int
    emit_metric: Callable[[str, float, dict], None]

    used_input: int = 0
    used_output: int = 0

    def record(self, in_tok: int, out_tok: int) -> None:
        self.used_input += in_tok
        self.used_output += out_tok
        # Gửi metric realtime — dashboard \$/task dựng trên metric này.
        self.emit_metric(
            "llm.tokens.used",
            in_tok + out_tok,
            {"task": self.task_id, "kind": "total"},
        )
        self._warn_if_over_budget()

    def _warn_if_over_budget(self) -> None:
        in_pct = self.used_input / max(1, self.max_input)
        out_pct = self.used_output / max(1, self.max_output)
        if max(in_pct, out_pct) >= 0.8:
            self.emit_metric(
                "llm.budget.warning",
                max(in_pct, out_pct),
                {"task": self.task_id},
            )

    def cost_usd(self, p_in_per_m: float, p_out_per_m: float,
                 cache_hit: float = 0.0, discount: float = 0.9) -> float:
        eff_in = (p_in_per_m / 1_000_000) * (1 - cache_hit * discount)
        eff_out = p_out_per_m / 1_000_000
        return self.used_input * eff_in + self.used_output * eff_out

import type { NextRequest, NextResponse } from "next/server";

interface Usage {
  inputTokens: number;
  outputTokens: number;
  cachedInputTokens: number;
  model: string;
  latencyMs: number;
}

// Middleware bọc tầng gọi LLM. Mỗi request emit đúng một record usage.
export async function withTokenMeter<T>(
  req: NextRequest,
  run: () => Promise<{ result: T; usage: Usage }>,
  sink: (u: Usage & { route: string; userId?: string }) => void,
): Promise<T> {
  const started = performance.now();
  const { result, usage } = await run();
  const latencyMs = performance.now() - started;

  // Báo về observability (Datadog, Honeycomb, OTLP...) — một record / request.
  sink({
    ...usage,
    latencyMs,
    route: req.nextUrl.pathname,
    userId: req.headers.get("x-user-id") ?? undefined,
  });

  return result;
}

// Dashboard dùng dữ liệu này để tính $/task realtime:
// SELECT route, SUM(input_tokens * p_in + output_tokens * p_out) / COUNT(DISTINCT task_id)
// FROM llm_usage WHERE ts > now() - interval '1 hour' GROUP BY route;