9.4Measuring LLM Metrics

Non-Streaming Measurement

For non-streaming endpoints, you can only measure total generation time directly. TTFT must be estimated:

# llm_metrics_collector.py
import time
import json
from dataclasses import dataclass

@dataclass
class LLMMetrics:
    ttft_ms: float
    total_generation_ms: float
    tokens_per_second: float
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    model: str

def measure_non_streaming(client, prompt: str, model: str = "gpt-5.5") -> LLMMetrics:
    """Measure LLM performance for a non-streaming request."""
    start = time.perf_counter()

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=256,
        stream=False,
    )

    total_ms = (time.perf_counter() - start) * 1000
    completion_tokens = response.usage.completion_tokens
    tps = completion_tokens / (total_ms / 1000) if total_ms > 0 else 0

    return LLMMetrics(
        ttft_ms=total_ms * 0.15,  # heuristic: ~15% of total time is prefill
        total_generation_ms=total_ms,
        tokens_per_second=tps,
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=response.usage.total_tokens,
        model=model,
    )

Streaming Measurement (Accurate TTFT)

For accurate TTFT, you must use the streaming API:

import time

def measure_streaming(client, prompt: str, model: str = "gpt-5.5") -> LLMMetrics:
    """Measure LLM performance for a streaming request with accurate TTFT."""
    start = time.perf_counter()
    ttft = None
    token_count = 0
    full_response = ""

    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=256,
        stream=True,
        stream_options={"include_usage": True},
    )

    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            if ttft is None:
                ttft = (time.perf_counter() - start) * 1000
            full_response += chunk.choices[0].delta.content
            token_count += 1

        # Usage is in the final chunk when stream_options is set
        if hasattr(chunk, 'usage') and chunk.usage:
            prompt_tokens = chunk.usage.prompt_tokens
            completion_tokens = chunk.usage.completion_tokens

    total_ms = (time.perf_counter() - start) * 1000
    generation_time = total_ms - (ttft or 0)
    tps = completion_tokens / (generation_time / 1000) if generation_time > 0 else 0

    return LLMMetrics(
        ttft_ms=ttft or total_ms,
        total_generation_ms=total_ms,
        tokens_per_second=tps,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
        model=model,
    )