Spaces:

vxa8502
/

Sage

Running

Sage / scripts /load_test.py
Update and add evaluation, query-bank, and pipeline scripts
986def4 4 months ago
29.6 kB
#!/usr/bin/env python3
"""
Load test script for Sage API.
Runs a short warm-up phase, then measures steady-state latency and API quality.
The headline metric is ``steady_state_server_p95_ms`` when the API exposes the
``X-Response-Time-Ms`` header, with a client-side fallback when server timing is
unavailable.
Usage:
    # Start the API first:
    python -m sage.api.run
    # Then run the load test:
    python scripts/load_test.py --requests 100 --url http://localhost:8000
    # Test without explanations (faster):
    python scripts/load_test.py --no-explain
    # Run concurrent requests (stress test):
    python scripts/load_test.py --concurrent 10 --requests 100
    # Save results to JSON (for reproducibility):
    python scripts/load_test.py --save
"""

from __future__ import annotations

import argparse
import asyncio
import statistics
import sys
import time
from collections import Counter
from datetime import datetime
from typing import Any

import httpx
import numpy as np

from sage.config import RESULTS_DIR, save_results
from sage.services.faithfulness import is_valid_non_recommendation

# Headline latency target. Historically the project tracked a single
# ``< 500ms`` target; the load test now applies it to steady-state p95.
DEFAULT_TARGET_MS = 500.0
HEADLINE_METRIC_NAME = "steady_state_server_p95_ms"
LEGACY_FALLBACK_METRIC_NAME = "steady_state_client_p95_ms"

# Test queries covering different scenarios
QUERIES = [
    "wireless headphones for working out",
    "laptop for video editing under $1500",
    "best phone case for iPhone",
    "comfortable running shoes",
    "noise canceling earbuds",
    "gaming keyboard mechanical",
    "portable charger high capacity",
    "bluetooth speaker waterproof",
    "monitor for programming",
    "ergonomic office chair",
]


def percentile(data: list[float], p: float) -> float:
    """Calculate the p-th percentile of data."""
    if not data:
        return 0.0
    if p < 0 or p > 100:
        raise ValueError("Percentile must be between 0 and 100")
    return float(np.percentile(data, p))


def _round(value: float | None, digits: int = 1) -> float | None:
    """Round a float if present."""
    return None if value is None else round(value, digits)


def _build_payload(query: str, explain: bool) -> dict[str, Any]:
    """Build a request payload for the load test."""
    return {
        "query": query,
        "k": 3,
        "explain": explain,
    }


def _parse_ms_header(resp: httpx.Response, header_name: str) -> float | None:
    """Parse a millisecond timing header from a response."""
    raw = resp.headers.get(header_name)
    if raw is None:
        return None
    try:
        return float(raw)
    except ValueError:
        return None


def _has_server_timing_header(sample: dict[str, Any]) -> bool:
    """Return True when the response exposed X-Response-Time-Ms."""
    explicit = sample.get("server_timing_header_present")
    if isinstance(explicit, bool):
        return explicit
    return sample.get("server_ms") is not None


def _has_cache_result_header(sample: dict[str, Any]) -> bool:
    """Return True when the response exposed X-Cache-Result."""
    explicit = sample.get("cache_result_header_present")
    if isinstance(explicit, bool):
        return explicit
    return sample.get("cache_result") not in (None, "unknown")


def _is_grounded_recommendation(rec: dict[str, Any]) -> bool:
    """Return True when a recommendation is explicitly grounded."""
    explanation = rec.get("explanation")
    confidence = rec.get("confidence")
    return (
        isinstance(explanation, str)
        and bool(explanation.strip())
        and rec.get("citations_verified") is True
        and isinstance(confidence, dict)
        and confidence.get("is_grounded") is True
    )


def _is_refusal_aware_recommendation(rec: dict[str, Any]) -> bool:
    """Treat grounded recommendations and explicit non-recommendations as success."""
    explanation = rec.get("explanation")
    if isinstance(explanation, str) and is_valid_non_recommendation(explanation):
        return True
    return _is_grounded_recommendation(rec)


def _evaluate_response_quality(
    payload: dict[str, Any],
    explain: bool,
) -> dict[str, bool | None]:
    """Classify request-level API quality from a successful JSON response."""
    if not explain:
        return {
            "grounded_success": None,
            "refusal_aware_success": None,
        }

    policy_decision = payload.get("policy_decision")
    if isinstance(policy_decision, dict):
        observed_behavior = policy_decision.get("observed_behavior")
        if observed_behavior in {"refuse", "clarify", "hedge"}:
            return {
                "grounded_success": False,
                "refusal_aware_success": True,
            }

    recommendations = payload.get("recommendations")
    if not isinstance(recommendations, list) or not recommendations:
        return {
            "grounded_success": False,
            "refusal_aware_success": False,
        }

    return {
        "grounded_success": all(
            _is_grounded_recommendation(rec)
            for rec in recommendations
            if isinstance(rec, dict)
        ),
        "refusal_aware_success": all(
            _is_refusal_aware_recommendation(rec)
            for rec in recommendations
            if isinstance(rec, dict)
        ),
    }


def _build_sample(
    *,
    phase: str,
    request_index: int,
    query: str,
    explain: bool,
    resp: httpx.Response,
    client_ms: float,
) -> dict[str, Any]:
    """Build a per-request sample from a completed response."""
    body: dict[str, Any] = {}
    if resp.status_code == 200:
        try:
            body = resp.json()
        except ValueError:
            body = {}

    quality = _evaluate_response_quality(body, explain)
    response_time_header_present = resp.headers.get("X-Response-Time-Ms") is not None
    server_ms = _parse_ms_header(resp, "X-Response-Time-Ms")
    network_overhead_ms = (
        max(client_ms - server_ms, 0.0) if server_ms is not None else None
    )
    cache_result = resp.headers.get("X-Cache-Result")

    return {
        "phase": phase,
        "request_index": request_index,
        "query": query,
        "status": resp.status_code,
        "client_ms": _round(client_ms),
        "server_ms": _round(server_ms),
        "network_overhead_ms": _round(network_overhead_ms),
        "server_timing_header_present": response_time_header_present,
        "cache_result_header_present": cache_result is not None,
        "cache_result": cache_result if cache_result is not None else "unknown",
        "request_id": resp.headers.get("X-Request-Id"),
        "returned_count": body.get("returned_count"),
        "grounded_success": quality["grounded_success"],
        "refusal_aware_success": quality["refusal_aware_success"],
    }


def _build_exception_sample(
    *,
    phase: str,
    request_index: int,
    query: str,
    error: str,
) -> dict[str, Any]:
    """Build a per-request sample for a request-level exception."""
    return {
        "phase": phase,
        "request_index": request_index,
        "query": query,
        "status": "exception",
        "error": error,
        "client_ms": None,
        "server_ms": None,
        "network_overhead_ms": None,
        "server_timing_header_present": False,
        "cache_result_header_present": False,
        "cache_result": "unknown",
        "request_id": None,
        "returned_count": None,
        "grounded_success": None,
        "refusal_aware_success": None,
    }


def _summarize_values(values: list[float]) -> dict[str, float] | None:
    """Summarize a list of millisecond values."""
    if not values:
        return None

    return {
        "min": round(min(values), 1),
        "max": round(max(values), 1),
        "mean": round(statistics.mean(values), 1),
        "p50": round(percentile(values, 50), 1),
        "p95": round(percentile(values, 95), 1),
        "p99": round(percentile(values, 99), 1),
        "stdev": round(statistics.stdev(values), 1) if len(values) > 1 else 0.0,
    }


def _summarize_phase(samples: list[dict[str, Any]], explain: bool) -> dict[str, Any]:
    """Summarize one load-test phase from request samples."""
    successful = [sample for sample in samples if sample.get("status") == 200]
    client_values = [
        float(sample["client_ms"])
        for sample in successful
        if sample.get("client_ms") is not None
    ]
    server_values = [
        float(sample["server_ms"])
        for sample in successful
        if sample.get("server_ms") is not None
    ]
    overhead_values = [
        float(sample["network_overhead_ms"])
        for sample in successful
        if sample.get("network_overhead_ms") is not None
    ]

    cache_results = Counter(
        str(sample.get("cache_result"))
        for sample in successful
        if sample.get("cache_result") not in (None, "unknown")
    )
    total_successful = len(successful)
    response_time_header_present = sum(
        1 for sample in successful if _has_server_timing_header(sample)
    )
    cache_result_header_present = sum(
        1 for sample in successful if _has_cache_result_header(sample)
    )
    cache_hits = cache_results.get("exact", 0) + cache_results.get("semantic", 0)
    cache_known = sum(cache_results.values())
    cache_hit_rate = cache_hits / cache_known if cache_known > 0 else None

    evaluated = (
        [sample for sample in successful if sample.get("grounded_success") is not None]
        if explain
        else []
    )
    grounded_successes = sum(
        1 for sample in evaluated if sample.get("grounded_success") is True
    )
    refusal_aware_successes = sum(
        1 for sample in evaluated if sample.get("refusal_aware_success") is True
    )
    evaluated_count = len(evaluated)

    return {
        "total_requests": len(samples),
        "successful": total_successful,
        "errors": len(samples) - total_successful,
        "status_counts": dict(Counter(str(sample.get("status")) for sample in samples)),
        "client_latency_ms": _summarize_values(client_values),
        "server_latency_ms": _summarize_values(server_values),
        "network_overhead_ms": _summarize_values(overhead_values),
        "header_presence": {
            "x_response_time_ms": {
                "present": response_time_header_present,
                "missing": total_successful - response_time_header_present,
            },
            "x_cache_result": {
                "present": cache_result_header_present,
                "missing": total_successful - cache_result_header_present,
            },
        },
        "cache_observability": {
            "available": cache_result_header_present > 0,
            "successful_responses": total_successful,
            "header_present_responses": cache_result_header_present,
            "missing_header_responses": total_successful - cache_result_header_present,
            "reason": (
                None
                if cache_result_header_present > 0 or total_successful == 0
                else "X-Cache-Result header absent on successful responses"
            ),
        },
        "cache_results": dict(cache_results),
        "cache_hit_rate": _round(cache_hit_rate, 4),
        "api_quality": {
            "evaluated_requests": evaluated_count,
            "grounded_successes": grounded_successes,
            "grounded_success_rate": _round(
                grounded_successes / evaluated_count if evaluated_count > 0 else None,
                4,
            ),
            "refusal_aware_successes": refusal_aware_successes,
            "refusal_aware_success_rate": _round(
                refusal_aware_successes / evaluated_count
                if evaluated_count > 0
                else None,
                4,
            ),
        },
    }


def _select_headline_metric(
    measured_summary: dict[str, Any],
    target_ms: float,
) -> dict[str, Any]:
    """Select the canonical headline latency metric for the artifact."""
    server = measured_summary.get("server_latency_ms") or {}
    client = measured_summary.get("client_latency_ms") or {}

    if server.get("p95") is not None:
        value_ms = float(server["p95"])
        name = HEADLINE_METRIC_NAME
        source = "server"
    else:
        value_ms = float(client["p95"]) if client.get("p95") is not None else None
        name = LEGACY_FALLBACK_METRIC_NAME
        source = "client_fallback"

    return {
        "name": name,
        "phase": "measured",
        "source": source,
        "value_ms": _round(value_ms),
        "target_ms": round(target_ms, 1),
        "pass": value_ms <= target_ms if value_ms is not None else None,
    }


def _build_results(
    *,
    warmup_samples: list[dict[str, Any]],
    measured_samples: list[dict[str, Any]],
    config: dict[str, Any],
    target_ms: float,
) -> dict[str, Any]:
    """Build the final artifact from warm-up and measured request samples."""
    warmup_summary = _summarize_phase(warmup_samples, explain=config["explain"])
    measured_summary = _summarize_phase(measured_samples, explain=config["explain"])
    headline_metric = _select_headline_metric(measured_summary, target_ms)
    client = measured_summary.get("client_latency_ms") or {}
    server = measured_summary.get("server_latency_ms") or {}
    quality = measured_summary.get("api_quality") or {}
    cache_results = measured_summary.get("cache_results") or {}
    cache_hits = (
        cache_results.get("exact", 0) + cache_results.get("semantic", 0)
        if cache_results
        else None
    )

    return {
        "timestamp": datetime.now().isoformat(),
        "config": config,
        "headline_metric": headline_metric,
        "warmup": warmup_summary,
        "measured": measured_summary,
        "samples": {
            "warmup": warmup_samples,
            "measured": measured_samples,
        },
        # Backward-compatible headline fields (measured client-side latency)
        "total_requests": measured_summary["total_requests"],
        "successful": measured_summary["successful"],
        "errors": measured_summary["errors"],
        "cache_hits": cache_hits,
        "min_ms": client.get("min"),
        "max_ms": client.get("max"),
        "mean_ms": client.get("mean"),
        "p50_ms": client.get("p50"),
        "p95_ms": client.get("p95"),
        "p99_ms": client.get("p99"),
        "stdev_ms": client.get("stdev"),
        "server_p50_ms": server.get("p50"),
        "server_p95_ms": server.get("p95"),
        "server_p99_ms": server.get("p99"),
        "cache_hit_rate": measured_summary.get("cache_hit_rate"),
        "grounded_success_rate": quality.get("grounded_success_rate"),
        "refusal_aware_success_rate": quality.get("refusal_aware_success_rate"),
        "target_ms": round(target_ms, 1),
        "pass": headline_metric.get("pass"),
    }


def _print_progress(phase: str, completed: int, total: int) -> None:
    """Print a short progress update."""
    if completed % 10 == 0 or completed == total:
        print(f"  {phase}: completed {completed}/{total} requests...")


def _run_phase_sync(
    client: httpx.Client,
    endpoint: str,
    *,
    phase: str,
    num_requests: int,
    explain: bool,
) -> list[dict[str, Any]]:
    """Run one sequential phase and return per-request samples."""
    samples: list[dict[str, Any]] = []

    for i in range(num_requests):
        query = QUERIES[i % len(QUERIES)]

        try:
            start = time.perf_counter()
            resp = client.post(endpoint, json=_build_payload(query, explain))
            client_ms = (time.perf_counter() - start) * 1000
            samples.append(
                _build_sample(
                    phase=phase,
                    request_index=i + 1,
                    query=query,
                    explain=explain,
                    resp=resp,
                    client_ms=client_ms,
                )
            )

            if resp.status_code != 200:
                print(
                    f"  [{phase} {i + 1}] Error: {resp.status_code} - {resp.text[:100]}"
                )
        except Exception as exc:
            print(f"  [{phase} {i + 1}] Exception: {exc}")
            samples.append(
                _build_exception_sample(
                    phase=phase,
                    request_index=i + 1,
                    query=query,
                    error=str(exc),
                )
            )

        _print_progress(phase, i + 1, num_requests)

    return samples


async def _run_phase_concurrent(
    client: httpx.AsyncClient,
    endpoint: str,
    *,
    phase: str,
    num_requests: int,
    concurrency: int,
    explain: bool,
) -> list[dict[str, Any]]:
    """Run one concurrent phase and return per-request samples."""
    samples: list[dict[str, Any]] = []
    lock = asyncio.Lock()

    async def fetch_one(query: str, request_index: int) -> None:
        try:
            start = time.perf_counter()
            resp = await client.post(endpoint, json=_build_payload(query, explain))
            client_ms = (time.perf_counter() - start) * 1000
            sample = _build_sample(
                phase=phase,
                request_index=request_index,
                query=query,
                explain=explain,
                resp=resp,
                client_ms=client_ms,
            )
            async with lock:
                samples.append(sample)
                if resp.status_code != 200:
                    print(
                        f"  [{phase} {request_index}] Error: {resp.status_code} - {resp.text[:100]}"
                    )
        except Exception as exc:
            print(f"  [{phase} {request_index}] Exception: {exc}")
            async with lock:
                samples.append(
                    _build_exception_sample(
                        phase=phase,
                        request_index=request_index,
                        query=query,
                        error=str(exc),
                    )
                )

    for batch_start in range(0, num_requests, concurrency):
        batch_end = min(batch_start + concurrency, num_requests)
        tasks = []
        for i in range(batch_start, batch_end):
            query = QUERIES[i % len(QUERIES)]
            tasks.append(fetch_one(query, i + 1))

        await asyncio.gather(*tasks)
        _print_progress(phase, batch_end, num_requests)

    samples.sort(key=lambda sample: int(sample["request_index"]))
    return samples


def run_load_test(
    base_url: str,
    num_requests: int,
    warmup_requests: int,
    explain: bool,
    timeout: float,
    target_ms: float,
) -> dict[str, Any]:
    """Run a sequential load test with warm-up and measured phases."""
    endpoint = f"{base_url}/recommend"

    print(f"\nRunning sequential load test against {endpoint}")
    print(
        f"  warmup={warmup_requests}, measured={num_requests}, explain={explain}, timeout={timeout}s"
    )
    print("-" * 50)

    with httpx.Client(timeout=timeout) as client:
        warmup_samples = _run_phase_sync(
            client,
            endpoint,
            phase="warmup",
            num_requests=warmup_requests,
            explain=explain,
        )
        measured_samples = _run_phase_sync(
            client,
            endpoint,
            phase="measured",
            num_requests=num_requests,
            explain=explain,
        )

    return _build_results(
        warmup_samples=warmup_samples,
        measured_samples=measured_samples,
        config={
            "url": base_url,
            "warmup_requests": warmup_requests,
            "num_requests": num_requests,
            "mode": "sequential",
            "explain": explain,
            "timeout_s": timeout,
        },
        target_ms=target_ms,
    )


async def run_concurrent_load_test(
    base_url: str,
    num_requests: int,
    warmup_requests: int,
    concurrency: int,
    explain: bool,
    timeout: float,
    target_ms: float,
) -> dict[str, Any]:
    """Run a concurrent load test with a sequential warm-up phase."""
    endpoint = f"{base_url}/recommend"

    print(f"\nRunning concurrent load test against {endpoint}")
    print(
        "  "
        f"warmup={warmup_requests}, measured={num_requests}, concurrency={concurrency}, "
        f"explain={explain}, timeout={timeout}s"
    )
    print("-" * 50)

    warmup_transport = httpx.Client(timeout=timeout)
    try:
        warmup_samples = _run_phase_sync(
            warmup_transport,
            endpoint,
            phase="warmup",
            num_requests=warmup_requests,
            explain=explain,
        )
    finally:
        warmup_transport.close()

    async with httpx.AsyncClient(timeout=timeout) as client:
        measured_samples = await _run_phase_concurrent(
            client,
            endpoint,
            phase="measured",
            num_requests=num_requests,
            concurrency=concurrency,
            explain=explain,
        )

    return _build_results(
        warmup_samples=warmup_samples,
        measured_samples=measured_samples,
        config={
            "url": base_url,
            "warmup_requests": warmup_requests,
            "num_requests": num_requests,
            "mode": "concurrent",
            "concurrency": concurrency,
            "explain": explain,
            "timeout_s": timeout,
        },
        target_ms=target_ms,
    )


def _print_latency_block(label: str, stats: dict[str, Any] | None) -> None:
    """Print a formatted latency stats block when available."""
    if not stats:
        return
    print(f"  {label}:")
    print(
        "    "
        f"p50={stats['p50']:.1f}ms  p95={stats['p95']:.1f}ms  p99={stats['p99']:.1f}ms"
    )
    print(
        f"    min={stats['min']:.1f}  max={stats['max']:.1f}  mean={stats['mean']:.1f}"
    )


def print_results(
    results: dict[str, Any], target_ms: float = DEFAULT_TARGET_MS
) -> None:
    """Print formatted load-test results."""
    print("\n" + "=" * 50)
    print("LOAD TEST RESULTS")
    print("=" * 50)

    config = results.get("config", {})
    print(
        f"\nMode: {config.get('mode', 'sequential')}"
        + (
            f" ({config.get('concurrency')} parallel requests)"
            if config.get("concurrency")
            else ""
        )
    )
    print(
        f"Warm-up: {config.get('warmup_requests', 0)} requests | "
        f"Measured: {config.get('num_requests', 0)} requests"
    )

    measured = results.get("measured", {})
    warmup = results.get("warmup", {})
    headline = results.get("headline_metric", {})

    print(
        f"Measured success: {measured.get('successful', 0)}/{measured.get('total_requests', 0)}"
    )
    print(f"Measured errors: {measured.get('errors', 0)}")

    if headline.get("value_ms") is not None:
        status = "PASS" if headline.get("pass") else "FAIL"
        print(
            "\nHeadline metric:"
            f" {headline.get('name')} = {headline['value_ms']:.1f}ms "
            f"(target <= {target_ms:.0f}ms) [{status}]"
        )

    print("\nLatency drill-down (measured phase):")
    _print_latency_block("Client RTT", measured.get("client_latency_ms"))
    _print_latency_block("Server time", measured.get("server_latency_ms"))
    _print_latency_block(
        "Network/platform overhead", measured.get("network_overhead_ms")
    )

    if warmup:
        warmup_client = (warmup.get("client_latency_ms") or {}).get("p95")
        warmup_server = (warmup.get("server_latency_ms") or {}).get("p95")
        if warmup_client is not None or warmup_server is not None:
            print("\nWarm-up reference:")
            if warmup_client is not None:
                print(f"  Warm-up client p95: {warmup_client:.1f}ms")
            if warmup_server is not None:
                print(f"  Warm-up server p95: {warmup_server:.1f}ms")

    cache_results = measured.get("cache_results", {})
    header_presence = measured.get("header_presence", {})
    cache_observability = measured.get("cache_observability", {})
    if cache_results:
        exact = cache_results.get("exact", 0)
        semantic = cache_results.get("semantic", 0)
        miss = cache_results.get("miss", 0)
        hit_rate = measured.get("cache_hit_rate")
        print("\nCache breakdown (measured phase):")
        print(f"  exact={exact}  semantic={semantic}  miss={miss}")
        if hit_rate is not None:
            print(f"  hit rate={hit_rate * 100:.1f}%")
    elif measured.get("successful", 0) > 0:
        missing = cache_observability.get(
            "missing_header_responses", measured.get("successful", 0)
        )
        total = cache_observability.get(
            "successful_responses", measured.get("successful", 0)
        )
        print("\nCache breakdown (measured phase):")
        print(
            "  unavailable "
            f"(X-Cache-Result header absent on {missing}/{total} successful responses)"
        )

    if header_presence:
        response_time = header_presence.get("x_response_time_ms", {})
        cache_header = header_presence.get("x_cache_result", {})
        print("\nObservability headers (measured phase):")
        print(
            "  "
            f"X-Response-Time-Ms present on "
            f"{response_time.get('present', 0)}/{measured.get('successful', 0)} successful responses"
        )
        print(
            "  "
            f"X-Cache-Result present on "
            f"{cache_header.get('present', 0)}/{measured.get('successful', 0)} successful responses"
        )

    quality = measured.get("api_quality", {})
    if quality.get("evaluated_requests", 0) > 0:
        print("\nAPI quality (measured phase, explain=true):")
        print(
            "  "
            f"grounded_success_rate="
            f"{quality['grounded_success_rate'] * 100:.1f}% "
            f"({quality['grounded_successes']}/{quality['evaluated_requests']})"
        )
        print(
            "  "
            f"refusal_aware_success_rate="
            f"{quality['refusal_aware_success_rate'] * 100:.1f}% "
            f"({quality['refusal_aware_successes']}/{quality['evaluated_requests']})"
        )

    print("\n" + "=" * 50)


def main() -> None:
    parser = argparse.ArgumentParser(description="Load test Sage API")
    parser.add_argument(
        "--url",
        default="http://localhost:8000",
        help="Base URL of the API (default: http://localhost:8000)",
    )
    parser.add_argument(
        "--requests",
        type=int,
        default=100,
        help="Number of measured requests to send after warm-up (default: 100)",
    )
    parser.add_argument(
        "--warmup-requests",
        type=int,
        default=len(QUERIES),
        help=f"Number of warm-up requests before measurement (default: {len(QUERIES)})",
    )
    parser.add_argument(
        "--no-explain",
        action="store_true",
        help="Disable explanations (faster, tests retrieval only)",
    )
    parser.add_argument(
        "--concurrent",
        type=int,
        default=0,
        metavar="N",
        help="Run N concurrent measured requests per batch (default: 0 = sequential)",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=30.0,
        help="Request timeout in seconds (default: 30)",
    )
    parser.add_argument(
        "--target-ms",
        type=float,
        default=DEFAULT_TARGET_MS,
        help=f"Headline latency target in ms (default: {DEFAULT_TARGET_MS:.0f})",
    )
    parser.add_argument(
        "--target-p99",
        dest="target_ms",
        type=float,
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--save",
        action="store_true",
        help="Save results to data/eval_results/load_test_*.json",
    )

    args = parser.parse_args()

    # Quick health check
    try:
        resp = httpx.get(f"{args.url}/health", timeout=5.0)
        if resp.status_code != 200:
            print(f"API health check failed: {resp.status_code}")
            sys.exit(1)
        health = resp.json()
        print(f"API Status: {health.get('status', 'unknown')}")
        print(
            f"Qdrant: {'connected' if health.get('qdrant_connected') else 'disconnected'}"
        )
        print(f"LLM: {'available' if health.get('llm_reachable') else 'unavailable'}")
    except Exception as exc:
        print(f"Cannot connect to API at {args.url}: {exc}")
        sys.exit(1)

    if args.concurrent > 0:
        results = asyncio.run(
            run_concurrent_load_test(
                base_url=args.url,
                num_requests=args.requests,
                warmup_requests=args.warmup_requests,
                concurrency=args.concurrent,
                explain=not args.no_explain,
                timeout=args.timeout,
                target_ms=args.target_ms,
            )
        )
    else:
        results = run_load_test(
            base_url=args.url,
            num_requests=args.requests,
            warmup_requests=args.warmup_requests,
            explain=not args.no_explain,
            timeout=args.timeout,
            target_ms=args.target_ms,
        )

    print_results(results, target_ms=args.target_ms)

    if args.save:
        RESULTS_DIR.mkdir(parents=True, exist_ok=True)
        saved_path = save_results(results, "load_test")
        print(f"\nResults saved: {saved_path}")


if __name__ == "__main__":
    main()