ms-ai-architect/skills/ms-ai-security/references/performance-scalability/performance-benchmarking-frameworks.md

Performance Benchmarking Frameworks

Last updated: 2026-06-24 Status: GA Category: Performance & Scalability Type: reference Source: https://learn.microsoft.com/azure/load-testing/overview-what-is-azure-load-testing

---

Innhold

• Introduksjon
• Kjernekomponenter
• Metric Definition Standards
• Baseline Establishment
• Regression Detection
• Comparative Analysis Methods
• Norsk offentlig sektor
• Beslutningsrammeverk
• Referanser
• For Cosmo

Introduksjon

Et performance benchmarking framework for Azure AI Services gir en strukturert tilnærming til å måle, sammenligne og spore ytelse over tid. Uten et rammeverk blir ytelsesmålinger ad hoc, ikke-reproduserbare og vanskelige å sammenligne mellom modellversjoner, deployment-konfigurasjoner eller arkitekturendringer.

Microsoft tilbyr et offisielt benchmarking-verktøy (azure-openai-benchmark) spesifikt for Azure OpenAI, samt Azure Load Testing for bredere lasttesting. I tillegg tilbyr Microsoft Foundry innebygde evalueringsverktøy som kan brukes for å måle modellkvalitet. Et komplett benchmarking framework kombinerer disse verktøyene med egendefinerte metrikker, baseline-etablering og automatisk regresjonsdeteksjon.

For norsk offentlig sektor er et benchmarking framework viktig for å dokumentere ytelseskrav i tjenesteavtaler, verifisere at nye modellversjoner møter kvalitetskrav, og for å sikre at AI-tjenester oppfyller krav til responstid i henhold til digitaliseringsstrategien.

Kjernekomponenter

Komponent	Formål	Teknologi
azure-openai-benchmark	Offisielt Azure OpenAI benchmarking CLI	GitHub/Python
Azure Load Testing	Managed lasttesting med JMeter	Azure Load Testing
Microsoft Foundry Evaluations	Modellkvalitets-evaluering	Microsoft Foundry
Azure Monitor	Metrikk-innsamling og visualisering	Azure Monitor
Application Insights	End-to-end request tracing	App Insights
Custom Benchmark Suite	Prosjektspesifikke ytelsestester	Python/C#

Metric Definition Standards

Kjernemetrikker for AI-ytelse

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

class MetricCategory(Enum):
    LATENCY = "latency"
    THROUGHPUT = "throughput"
    QUALITY = "quality"
    COST = "cost"
    AVAILABILITY = "availability"

@dataclass
class BenchmarkMetric:
    name: str
    category: MetricCategory
    unit: str
    description: str
    target: Optional[float] = None
    warning_threshold: Optional[float] = None
    critical_threshold: Optional[float] = None

# Standard metrikkdefinisjoner for Azure OpenAI
STANDARD_METRICS = [
    BenchmarkMetric(
        name="time_to_first_token",
        category=MetricCategory.LATENCY,
        unit="ms",
        description="Tid fra forespørsel sendt til første token mottatt",
        target=500,
        warning_threshold=1000,
        critical_threshold=3000
    ),
    BenchmarkMetric(
        name="end_to_end_latency_p50",
        category=MetricCategory.LATENCY,
        unit="ms",
        description="P50 total responstid inkl. alle tokens",
        target=2000,
        warning_threshold=5000,
        critical_threshold=15000
    ),
    BenchmarkMetric(
        name="end_to_end_latency_p95",
        category=MetricCategory.LATENCY,
        unit="ms",
        description="P95 total responstid",
        target=5000,
        warning_threshold=10000,
        critical_threshold=30000
    ),
    BenchmarkMetric(
        name="tokens_per_second",
        category=MetricCategory.THROUGHPUT,
        unit="tokens/s",
        description="Output tokens generert per sekund",
        target=40,
        warning_threshold=20,
        critical_threshold=10
    ),
    BenchmarkMetric(
        name="requests_per_second",
        category=MetricCategory.THROUGHPUT,
        unit="req/s",
        description="Vellykkede forespørsler per sekund",
        target=5,
        warning_threshold=2,
        critical_threshold=1
    ),
    BenchmarkMetric(
        name="throttle_rate",
        category=MetricCategory.AVAILABILITY,
        unit="%",
        description="Andel forespørsler som fikk 429",
        target=0,
        warning_threshold=5,
        critical_threshold=20
    ),
    BenchmarkMetric(
        name="error_rate",
        category=MetricCategory.AVAILABILITY,
        unit="%",
        description="Andel feilede forespørsler (ekskl. 429)",
        target=0,
        warning_threshold=1,
        critical_threshold=5
    ),
    BenchmarkMetric(
        name="cost_per_request_nok",
        category=MetricCategory.COST,
        unit="NOK",
        description="Gjennomsnittlig kostnad per forespørsel",
        target=0.50,
        warning_threshold=1.00,
        critical_threshold=5.00
    ),
    BenchmarkMetric(
        name="prompt_cache_hit_rate",
        category=MetricCategory.COST,
        unit="%",
        description="Andel input-tokens som treffer prompt cache",
        target=60,
        warning_threshold=30,
        critical_threshold=10
    )
]

Baseline Establishment

Systematisk baseline-etablering

import json
import asyncio
from datetime import datetime
from dataclasses import asdict

@dataclass
class BenchmarkBaseline:
    model: str
    deployment_type: str
    region: str
    date: str
    workload_shape: dict
    metrics: dict
    environment: dict

class BaselineEstablisher:
    """Establish performance baseline for AI deployments."""

    def __init__(self, client, model: str, deployment_type: str, region: str):
        self.client = client
        self.model = model
        self.deployment_type = deployment_type
        self.region = region

    async def establish_baseline(
        self,
        test_prompts: list[dict],
        num_iterations: int = 100,
        concurrency_levels: list[int] = None
    ) -> BenchmarkBaseline:
        """Run comprehensive baseline benchmark."""
        if concurrency_levels is None:
            concurrency_levels = [1, 5, 10, 20]

        all_results = {}

        for concurrency in concurrency_levels:
            results = await self._run_at_concurrency(
                test_prompts, num_iterations, concurrency)
            all_results[f"concurrency_{concurrency}"] = results

        # Beregn aggregerte metrikker
        baseline_metrics = self._aggregate_metrics(all_results)

        baseline = BenchmarkBaseline(
            model=self.model,
            deployment_type=self.deployment_type,
            region=self.region,
            date=datetime.utcnow().isoformat(),
            workload_shape={
                "num_prompts": len(test_prompts),
                "avg_input_tokens": self._avg_tokens(test_prompts),
                "iterations": num_iterations,
                "concurrency_levels": concurrency_levels
            },
            metrics=baseline_metrics,
            environment={
                "api_version": "2024-10-21",
                "sdk_version": "1.x"
            }
        )

        return baseline

    async def _run_at_concurrency(
        self, prompts, iterations, concurrency
    ) -> dict:
        """Run benchmark at specific concurrency level."""
        import time
        semaphore = asyncio.Semaphore(concurrency)
        latencies = []
        ttfts = []
        token_counts = []
        errors = 0
        throttled = 0

        async def send_one(prompt):
            nonlocal errors, throttled
            async with semaphore:
                start = time.time()
                try:
                    # Streaming for TTFT measurement
                    first_token_time = None
                    total_tokens = 0

                    stream = await self.client.chat.completions.create(
                        model=self.model,
                        messages=prompt["messages"],
                        stream=True,
                        max_tokens=500
                    )

                    async for chunk in stream:
                        if first_token_time is None and \
                           chunk.choices and \
                           chunk.choices[0].delta.content:
                            first_token_time = time.time()
                        if chunk.choices and chunk.choices[0].delta.content:
                            total_tokens += 1

                    end = time.time()
                    latencies.append((end - start) * 1000)
                    if first_token_time:
                        ttfts.append((first_token_time - start) * 1000)
                    token_counts.append(total_tokens)

                except Exception as e:
                    errors += 1
                    if hasattr(e, 'status_code') and e.status_code == 429:
                        throttled += 1

        tasks = []
        for i in range(iterations):
            prompt = prompts[i % len(prompts)]
            tasks.append(send_one(prompt))

        start_time = time.time()
        await asyncio.gather(*tasks)
        total_duration = time.time() - start_time

        return {
            "latency_p50": sorted(latencies)[len(latencies)//2] if latencies else 0,
            "latency_p95": sorted(latencies)[int(len(latencies)*0.95)] if latencies else 0,
            "latency_p99": sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0,
            "ttft_p50": sorted(ttfts)[len(ttfts)//2] if ttfts else 0,
            "ttft_p95": sorted(ttfts)[int(len(ttfts)*0.95)] if ttfts else 0,
            "throughput_rps": round(len(latencies) / total_duration, 2),
            "tps": round(sum(token_counts) / total_duration, 1),
            "error_rate": round(errors / iterations * 100, 2),
            "throttle_rate": round(throttled / iterations * 100, 2)
        }

    def _aggregate_metrics(self, all_results: dict) -> dict:
        """Aggregate results across concurrency levels."""
        return {
            "optimal_concurrency": max(
                all_results.keys(),
                key=lambda k: all_results[k]["throughput_rps"]
            ),
            "by_concurrency": all_results
        }

    def _avg_tokens(self, prompts):
        return round(sum(
            len(str(p).split()) for p in prompts
        ) / len(prompts))

    def save_baseline(self, baseline: BenchmarkBaseline, path: str):
        """Save baseline to JSON file."""
        with open(path, "w") as f:
            json.dump(asdict(baseline), f, indent=2, default=str)

Regression Detection

Automatisk regresjonsdeteksjon

from dataclasses import dataclass

@dataclass
class RegressionResult:
    metric_name: str
    baseline_value: float
    current_value: float
    change_pct: float
    severity: str  # "none", "warning", "critical"
    direction: str  # "improved", "degraded", "stable"

class RegressionDetector:
    """Detect performance regressions against baseline."""

    def __init__(
        self,
        baseline: BenchmarkBaseline,
        warning_threshold_pct: float = 20,
        critical_threshold_pct: float = 50
    ):
        self.baseline = baseline
        self.warning_pct = warning_threshold_pct
        self.critical_pct = critical_threshold_pct

    def compare(self, current_metrics: dict) -> list[RegressionResult]:
        """Compare current metrics against baseline."""
        results = []

        # Definer retning: for noen metrikker er lavere bedre
        lower_is_better = {
            "latency_p50", "latency_p95", "latency_p99",
            "ttft_p50", "ttft_p95",
            "error_rate", "throttle_rate",
            "cost_per_request_nok"
        }

        baseline_data = self.baseline.metrics.get(
            "by_concurrency", {}).get(
                self.baseline.metrics.get("optimal_concurrency", ""),
                {})

        for metric_name, baseline_value in baseline_data.items():
            if metric_name not in current_metrics:
                continue

            current_value = current_metrics[metric_name]
            if baseline_value == 0:
                continue

            change_pct = (
                (current_value - baseline_value) / baseline_value * 100)

            # Bestem om endring er forbedring eller forverring
            is_lower_better = metric_name in lower_is_better
            if is_lower_better:
                degraded = change_pct > 0
            else:
                degraded = change_pct < 0

            abs_change = abs(change_pct)

            if abs_change < 5:
                severity = "none"
                direction = "stable"
            elif degraded:
                severity = (
                    "critical" if abs_change > self.critical_pct
                    else "warning" if abs_change > self.warning_pct
                    else "none")
                direction = "degraded"
            else:
                severity = "none"
                direction = "improved"

            results.append(RegressionResult(
                metric_name=metric_name,
                baseline_value=round(baseline_value, 2),
                current_value=round(current_value, 2),
                change_pct=round(change_pct, 1),
                severity=severity,
                direction=direction
            ))

        return results

    def generate_report(self, results: list[RegressionResult]) -> str:
        """Generate human-readable regression report."""
        lines = [
            "# Performance Regression Report",
            f"Baseline: {self.baseline.date}",
            f"Model: {self.baseline.model}",
            f"Region: {self.baseline.region}",
            ""
        ]

        critical = [r for r in results if r.severity == "critical"]
        warnings = [r for r in results if r.severity == "warning"]
        improvements = [r for r in results if r.direction == "improved"]

        if critical:
            lines.append("## CRITICAL Regressions")
            for r in critical:
                lines.append(
                    f"- **{r.metric_name}**: "
                    f"{r.baseline_value} → {r.current_value} "
                    f"({r.change_pct:+.1f}%)")

        if warnings:
            lines.append("\n## Warnings")
            for r in warnings:
                lines.append(
                    f"- {r.metric_name}: "
                    f"{r.baseline_value} → {r.current_value} "
                    f"({r.change_pct:+.1f}%)")

        if improvements:
            lines.append("\n## Improvements")
            for r in improvements:
                lines.append(
                    f"- {r.metric_name}: "
                    f"{r.baseline_value} → {r.current_value} "
                    f"({r.change_pct:+.1f}%)")

        return "\n".join(lines)

Comparative Analysis Methods

A/B-testing av modeller og konfigurasjoner

class ABBenchmarkComparator:
    """Compare performance between two configurations."""

    def __init__(self):
        self.results_a = None
        self.results_b = None

    async def compare_configs(
        self,
        config_a: dict,
        config_b: dict,
        test_prompts: list[dict],
        iterations: int = 100
    ) -> dict:
        """Run same workload against two configs and compare."""
        # Kjør A
        self.results_a = await self._benchmark(
            config_a, test_prompts, iterations)

        # Kjør B
        self.results_b = await self._benchmark(
            config_b, test_prompts, iterations)

        # Sammenlign
        comparison = {}
        for metric in self.results_a:
            if metric in self.results_b:
                val_a = self.results_a[metric]
                val_b = self.results_b[metric]
                if val_a != 0:
                    change = (val_b - val_a) / val_a * 100
                else:
                    change = 0

                comparison[metric] = {
                    "config_a": round(val_a, 2),
                    "config_b": round(val_b, 2),
                    "change_pct": round(change, 1),
                    "winner": "A" if self._is_better(metric, val_a, val_b)
                             else "B"
                }

        return comparison

    def _is_better(self, metric: str, val_a: float, val_b: float) -> bool:
        """Determine if A is better than B for given metric."""
        lower_better = {"latency", "error", "throttle", "cost", "ttft"}
        is_lower_better = any(k in metric for k in lower_better)
        return (val_a < val_b) if is_lower_better else (val_a > val_b)


# CI/CD integrasjon
async def ci_benchmark_gate(
    baseline_path: str,
    client,
    model: str,
    test_prompts: list[dict],
    max_regression_pct: float = 20
) -> bool:
    """Run benchmark as CI/CD quality gate."""
    with open(baseline_path) as f:
        baseline_data = json.load(f)
    baseline = BenchmarkBaseline(**baseline_data)

    # Kjør benchmark
    establisher = BaselineEstablisher(client, model, "standard", "norwayeast")
    current = await establisher._run_at_concurrency(test_prompts, 50, 10)

    # Sjekk regresjoner
    detector = RegressionDetector(baseline, warning_threshold_pct=max_regression_pct)
    results = detector.compare(current)

    critical = [r for r in results if r.severity == "critical"]
    if critical:
        print("BENCHMARK GATE FAILED:")
        for r in critical:
            print(f"  {r.metric_name}: {r.change_pct:+.1f}% regression")
        return False

    print("BENCHMARK GATE PASSED")
    return True

Norsk offentlig sektor

• Dokumentasjon: Benchmark-resultater bør lagres som del av prosjektdokumentasjonen og refereres i tjenesteavtaler.
• Regelmessighet: Kjør benchmarks månedlig og etter alle modelloppgraderinger, arkitekturendringer eller kvotejusteringer.
• Kvalitetskrav: Definer akseptable ytelsesgrenser i samarbeid med tjenesteeier — bruk STANDARD_METRICS som utgangspunkt.
• Åpenhet: For AI-tjenester som eksponeres mot borgere, dokumenter forventet responstid og tilgjengelighet.
• CI/CD: Integrer benchmark-gate i deployment-pipeline for å fange regresjoner før de når produksjon.

Beslutningsrammeverk

Scenario	Anbefaling	Begrunnelse
Ny deployment	Etabler baseline med full suite	Referansepunkt for fremtidige sammenligninger
Modelloppgradering	A/B sammenligning mot baseline	Verifiser at ny modell er like god eller bedre
Kvoteendring	Kjør throughput-benchmark	Mål faktisk forbedring
Produksjonsalert	Sammenlign mot baseline	Identifiser om det er regresjon
Kvartalsvis review	Full benchmark suite	Fang gradvis degradering

Referanser

• Azure OpenAI Benchmark Tool — Offisielt CLI-verktøy
• Azure Load Testing — Managed lasttesting
• Performance and latency — Ytelseskonsepter
• Evaluate generative AI models — Kvalitetsevaluering
• Azure Monitor metrics — Azure OpenAI monitoring

For Cosmo

• Bruk denne referansen når kunden trenger å etablere ytelsesbaselines, sette opp regelmessig ytelsestesting, eller integrere benchmarks i CI/CD.
• Et benchmark framework er IKKE valgfritt for produksjons-AI — uten baseline kan du ikke oppdage regresjoner eller validere forbedringer.
• Bruk det offisielle azure-openai-benchmark for PTU-dimensjonering, og custom Python-benchmarks for applikasjonsspesifikke metrikker.
• Kjør benchmarks i minimum 10 minutter per scenario for å oppnå steady state — korte tester gir misvisende resultater.
• Integrer ci_benchmark_gate i deployment pipeline — aldri deploy til produksjon uten å verifisere ytelse mot baseline.