open/ms-ai-architect
1
0
Fork 0
Code Issues Releases 1 Activity

feat(ultraplan-local): v1.6.0 — /ultraresearch-local deep research command

Browse source 
Add /ultraresearch-local for structured research combining local codebase
analysis with external knowledge via parallel agent swarms. Produces research
briefs with triangulation, confidence ratings, and source quality assessment.

New command: /ultraresearch-local with modes --quick, --local, --external, --fg.
New agents: research-orchestrator (opus), docs-researcher, community-researcher,
security-researcher, contrarian-researcher, gemini-bridge (all sonnet).
New template: research-brief-template.md.

Integration: --research flag in /ultraplan-local accepts pre-built research
briefs (up to 3), enriches the interview and exploration phases. Planning
orchestrator cross-references brief findings during synthesis.

Design principle: Context Engineering — right information to right agent at
right time. Research briefs are structured artifacts in the pipeline:
ultraresearch → brief → ultraplan --research → plan → ultraexecute.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-04-08 08:58:35 +02:00
commit baa2d0220b
488 changed files with 213221 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

549
skills/ms-ai-security/references/performance-scalability/performance-benchmarking-frameworks.md Normal file
View file

@ -0,0 +1,549 @@
# Performance Benchmarking Frameworks
**Last updated:** 2026-02
**Status:** GA
**Category:** Performance & Scalability
---
## 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 Azure AI 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 |
| Azure AI Foundry Evaluations | Modellkvalitets-evaluering | Azure AI 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
```python
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
```python
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
```python
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
```python
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](https://github.com/Azure/azure-openai-benchmark) — Offisielt CLI-verktøy
- [Azure Load Testing](https://learn.microsoft.com/azure/load-testing/overview-what-is-azure-load-testing) — Managed lasttesting
- [Performance and latency](https://learn.microsoft.com/azure/ai-foundry/openai/how-to/latency) — Ytelseskonsepter
- [Evaluate generative AI models](https://learn.microsoft.com/azure/ai-foundry/how-to/evaluate-generative-ai-app) — Kvalitetsevaluering
- [Azure Monitor metrics](https://learn.microsoft.com/azure/ai-foundry/openai/how-to/monitor-openai) — 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.