Index

bench

Benchmarking framework for Diapason inference engines.

Classes

BaseBenchmark

Bases: ABC

Base class for all benchmark implementations.

Subclasses must be registered via @BenchmarkRegistry.register("name") to become discoverable.

Attributes

name abstractmethod property

name: str

Short identifier for this benchmark.

description abstractmethod property

description: str

Human-readable description of what this benchmark measures.

Methods:

run abstractmethod

run(
    engine: InferenceEngine,
    model: str,
    *,
    num_samples: int = 10,
    **kwargs: Any,
) -> BenchmarkResult

Execute the benchmark and return results.

Source code in src/diapason/bench/_stubs.py

@abstractmethod
def run(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    num_samples: int = 10,
    **kwargs: Any,
) -> BenchmarkResult:
    """Execute the benchmark and return results."""

BenchmarkResult dataclass

BenchmarkResult(
    benchmark_name: str,
    model: str,
    engine: str,
    metrics: Dict[str, float] = dict(),
    metadata: Dict[str, Any] = dict(),
    samples: int = 0,
    errors: int = 0,
    warmup_samples: int = 0,
    steady_state_samples: int = 0,
    steady_state_reached: bool = False,
    total_energy_joules: float = 0.0,
    energy_per_token_joules: float = 0.0,
    energy_method: str = "",
)

Result from running a single benchmark.

BenchmarkSuite

BenchmarkSuite(
    benchmarks: Optional[List[BaseBenchmark]] = None,
)

Run a collection of benchmarks and aggregate results.

Source code in src/diapason/bench/_stubs.py

def __init__(self, benchmarks: Optional[List[BaseBenchmark]] = None) -> None:
    self._benchmarks = benchmarks or []

Methods:

run_all

run_all(
    engine: InferenceEngine,
    model: str,
    *,
    num_samples: int = 10,
    **kwargs: Any,
) -> List[BenchmarkResult]

Run all benchmarks and return a list of results.

Source code in src/diapason/bench/_stubs.py

def run_all(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    num_samples: int = 10,
    **kwargs: Any,
) -> List[BenchmarkResult]:
    """Run all benchmarks and return a list of results."""
    results: List[BenchmarkResult] = []
    for bench in self._benchmarks:
        result = bench.run(engine, model, num_samples=num_samples, **kwargs)
        results.append(result)
    return results

to_jsonl

to_jsonl(results: List[BenchmarkResult]) -> str

Serialize results to JSONL format (one JSON object per line).

Source code in src/diapason/bench/_stubs.py

def to_jsonl(self, results: List[BenchmarkResult]) -> str:
    """Serialize results to JSONL format (one JSON object per line)."""
    lines: List[str] = []
    for r in results:
        obj = {
            "benchmark_name": r.benchmark_name,
            "model": r.model,
            "engine": r.engine,
            "metrics": r.metrics,
            "metadata": r.metadata,
            "samples": r.samples,
            "errors": r.errors,
        }
        lines.append(json.dumps(obj))
    return "\n".join(lines)

summary

summary(results: List[BenchmarkResult]) -> Dict[str, Any]

Create a summary dict from benchmark results.

Source code in src/diapason/bench/_stubs.py

def summary(self, results: List[BenchmarkResult]) -> Dict[str, Any]:
    """Create a summary dict from benchmark results."""
    return {
        "benchmark_count": len(results),
        "benchmarks": [
            {
                "name": r.benchmark_name,
                "model": r.model,
                "engine": r.engine,
                "metrics": r.metrics,
                "samples": r.samples,
                "errors": r.errors,
            }
            for r in results
        ],
    }

BenchmarkRegistry

Bases: RegistryBase[Any]

Registry for benchmark implementations.

Functions:

ensure_registered

ensure_registered() -> None

Ensure all benchmark implementations are registered.

Source code in src/diapason/bench/__init__.py

def ensure_registered() -> None:
    """Ensure all benchmark implementations are registered."""
    from diapason.bench.energy import ensure_registered as _reg_energy
    from diapason.bench.latency import ensure_registered as _reg_latency
    from diapason.bench.throughput import ensure_registered as _reg_throughput

    _reg_latency()
    _reg_throughput()
    _reg_energy()