metrics.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/benchmarks/metrics.py
 
"""
Pure metric extraction + aggregation for the benchmark suite.
 
Purpose
-------
Turn a raw provider ``usage`` dict (the one ``presenter.show_usage``
receives — the only place ``reasoning_tokens`` and DeepSeek's prefix-
cache split survive; ``provider_plugin.UsageStats`` drops the former
and the harness wants both) into a typed :class:`CallMetrics`, then
aggregate calls → turns → run totals. **Hit and miss tokens are kept
distinct and never summed** — cache-hit is ~120x cheaper on DeepSeek
V4-pro, so collapsing them would erase the very economics the suite
measures.
 
All functions are pure: no I/O, no network, no global state. This is
the hermetically-tested core (plan 136 §6 tests 1-2).
 
Responsibilities
----------------
- :class:`CallMetrics` — one API call's token figures.
- :class:`TurnMetrics` — one scripted turn (its calls + TTFT + tool
  names + wall-clock + iteration/tool counts + cold/warm label).
- :func:`extract_call_metrics` — raw usage dict → :class:`CallMetrics`.
- :func:`aggregate_turn` / :func:`aggregate_run` — sum the token
  fields, hit/miss kept separate.
- :func:`derived_full_miss_cost` — the deterministic cold-turn cost
  (``prompt_tokens x miss-rate``), Plan 135's win #2.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``extract_call_metrics`` never raises on a malformed/partial usage
  dict — missing fields yield zeros, not ``KeyError``.
- Cache hit and miss are independent fields; no function sums them.
- Cache extraction delegates to the production
  :func:`oc_chat.core.cache_usage.usage_prompt_cache_hit_miss` so the
  numbers match the runtime exactly (and inherit its
  DeepSeek-then-OpenAI fallback).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.cache_usage (usage_prompt_cache_hit_miss — read-only)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
 
from oc_diagnostics import _dbg
 
from oc_chat.core.cache_usage import usage_prompt_cache_hit_miss
 
_dbg("OC_CHAT.MAIN", "module", "metrics loaded", 4)
 
 
@dataclass(frozen=True)
class CallMetrics:
    """Per-API-call token figures, normalised from the raw usage dict."""
 
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    reasoning_tokens: int = 0
    cache_hit_tokens: int = 0
    cache_miss_tokens: int = 0
 
 
@dataclass
class TurnMetrics:
    """One scripted turn's measurements.
 
    ``ttft_ms`` is the first iteration's time-to-first-token (``None``
    when an iteration emitted only tool calls — never fabricated).
    ``cold_label`` is one of ``"cold_unprewarmed"`` / ``"prewarmed"``
    for turn 1, or ``"warmed"`` for turns 2+ (warmup-aware, plan §3.3).
    """
 
    index: int = 0
    cold_label: str = "warmed"
    calls: list[CallMetrics] = field(default_factory=list)
    ttft_ms: float | None = None
    wall_clock_ms: float = 0.0
    iterations: int = 0
    main_loop_tool_names: list[str] = field(default_factory=list)
    sub_loop_count: int = 0
    sub_loop_tool_names_available: bool = False
 
 
def _as_int(value: object) -> int:
    """Coerce a usage value to int, treating ``None``/garbage as 0."""
    try:
        return int(value or 0)
    except (TypeError, ValueError):
        return 0
 
 
def _reasoning_tokens(raw_usage: dict) -> int:
    """Extract reasoning tokens (DeepSeek nests them under details).
 
    Primary location: ``completion_tokens_details.reasoning_tokens``.
    Fallback: a top-level ``reasoning_tokens`` if a provider emits one.
    """
    details = raw_usage.get("completion_tokens_details")
    if isinstance(details, dict) and details.get("reasoning_tokens") is not None:
        return _as_int(details.get("reasoning_tokens"))
    return _as_int(raw_usage.get("reasoning_tokens"))
 
 
def extract_call_metrics(raw_usage: dict | None) -> CallMetrics:
    """Normalise a single raw provider ``usage`` dict into :class:`CallMetrics`.
 
    Never raises: a ``None`` or partial dict yields zero-filled metrics.
    Cache hit/miss come from the production helper so they match the
    runtime; reasoning tokens come from the raw nested field that
    ``UsageStats`` does not carry.
    """
    if not isinstance(raw_usage, dict):
        return CallMetrics()
    hit, miss = usage_prompt_cache_hit_miss(raw_usage)
    return CallMetrics(
        prompt_tokens=_as_int(raw_usage.get("prompt_tokens")),
        completion_tokens=_as_int(raw_usage.get("completion_tokens")),
        total_tokens=_as_int(raw_usage.get("total_tokens")),
        reasoning_tokens=_reasoning_tokens(raw_usage),
        cache_hit_tokens=int(hit),
        cache_miss_tokens=int(miss),
    )
 
 
def aggregate_calls(calls: list[CallMetrics]) -> CallMetrics:
    """Sum a list of :class:`CallMetrics`. Hit and miss stay separate."""
    return CallMetrics(
        prompt_tokens=sum(c.prompt_tokens for c in calls),
        completion_tokens=sum(c.completion_tokens for c in calls),
        total_tokens=sum(c.total_tokens for c in calls),
        reasoning_tokens=sum(c.reasoning_tokens for c in calls),
        cache_hit_tokens=sum(c.cache_hit_tokens for c in calls),
        cache_miss_tokens=sum(c.cache_miss_tokens for c in calls),
    )
 
 
def aggregate_turn(turn: TurnMetrics) -> CallMetrics:
    """Sum one turn's calls into a single :class:`CallMetrics`."""
    return aggregate_calls(turn.calls)
 
 
def aggregate_run(turns: list[TurnMetrics]) -> CallMetrics:
    """Sum every call across every turn into the run total."""
    all_calls: list[CallMetrics] = []
    for turn in turns:
        all_calls.extend(turn.calls)
    return aggregate_calls(all_calls)
 
 
def derived_full_miss_cost(prompt_tokens: int, miss_rate_per_mtok: float) -> float:
    """The deterministic cold-turn cost: a full cache-miss of the prompt.
 
    ``prompt_tokens x miss-rate`` (rate is $/Mtok). This follows purely
    from prompt size, so it proves Plan 135's win #2 (a slim prompt is
    cheaper on the guaranteed-cold first turn) regardless of whatever
    server cache state the run actually hit.
    """
    return (max(0, int(prompt_tokens)) / 1_000_000.0) * float(miss_rate_per_mtok)
 
 
__all__ = [
    "CallMetrics",
    "TurnMetrics",
    "extract_call_metrics",
    "aggregate_calls",
    "aggregate_turn",
    "aggregate_run",
    "derived_full_miss_cost",
]