Spaces:
Running
Running
File size: 4,069 Bytes
24480a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """
backend/api/benchmarks_hub.py — Gap 3: router per i 3 benchmark standalone
Espone come endpoint HTTP le classi benchmark finora orfane:
POST /api/debug/benchmark/advanced ← AdvancedBenchmark (concorrenza)
POST /api/debug/benchmark/fabric ← FabricBenchmark (token rotation)
POST /api/debug/benchmark/extended ← ExtendedBenchmark (multi-scenario)
Tutti usano mock interni (asyncio.sleep + dict simulati) — sicuri da eseguire
in produzione senza side-effect su provider reali.
"""
import asyncio, logging, time
from fastapi import APIRouter
from fastapi.responses import JSONResponse
router = APIRouter()
_logger = logging.getLogger("api.benchmarks_hub")
@router.post("/api/debug/benchmark/advanced")
async def run_advanced() -> JSONResponse:
"""Stress test concorrenza: 3 scenari × N task simulati."""
t0 = time.monotonic()
try:
from api.advanced_complex_benchmark import AdvancedBenchmark
b = AdvancedBenchmark()
await b.setup()
await asyncio.gather(
b.run_complex_task("complex_reasoning", "reasoning", count=5),
b.run_complex_task("data_heavy_sync", "memory", count=5),
b.run_complex_task("multi_provider_chain","compute", count=5),
)
elapsed = round((time.monotonic() - t0) * 1000)
total = b.metrics["successes"] + b.metrics["failures"]
return JSONResponse({
"ok": True,
"type": "advanced",
"elapsed_ms": elapsed,
"success_rate": round(b.metrics["successes"] / max(1, total) * 100, 1),
"total_calls": total,
"successes": b.metrics["successes"],
"failures": b.metrics["failures"],
"latencies": {
k: {"avg_ms": round(sum(v) / len(v), 1), "max_ms": round(max(v), 1), "n": len(v)}
for k, v in b.metrics.items() if isinstance(v, list) and v
},
})
except Exception as exc:
_logger.exception("[advanced_bench]")
return JSONResponse({"ok": False, "type": "advanced", "error": str(exc)}, status_code=500)
@router.post("/api/debug/benchmark/fabric")
async def run_fabric() -> JSONResponse:
"""Benchmark rotazione token e fallback provider."""
t0 = time.monotonic()
try:
from api.fabric_benchmark import FabricBenchmark
b = FabricBenchmark()
await b.setup_simulated_environment()
await b.run_benchmark()
elapsed = round((time.monotonic() - t0) * 1000)
lats = b.stats.get("latencies", [])
return JSONResponse({
"ok": True,
"type": "fabric",
"elapsed_ms": elapsed,
"rotations": b.stats.get("rotations", 0),
"fallbacks": b.stats.get("fallbacks", 0),
"errors": len(b.stats.get("errors", [])),
"avg_latency_ms": round(sum(lats) / len(lats), 1) if lats else None,
})
except Exception as exc:
_logger.exception("[fabric_bench]")
return JSONResponse({"ok": False, "type": "fabric", "error": str(exc)}, status_code=500)
@router.post("/api/debug/benchmark/extended")
async def run_extended() -> JSONResponse:
"""Multi-scenario: normal load + cascading failure + oracle stress."""
t0 = time.monotonic()
try:
from api.extended_benchmark import ExtendedBenchmark
b = ExtendedBenchmark()
await b.setup()
await asyncio.gather(
b.scenario_normal_load(),
b.scenario_cascading_failure(),
b.scenario_oracle_stress(),
)
elapsed = round((time.monotonic() - t0) * 1000)
return JSONResponse({
"ok": True,
"type": "extended",
"elapsed_ms": elapsed,
"scenarios": b.results,
})
except Exception as exc:
_logger.exception("[extended_bench]")
return JSONResponse({"ok": False, "type": "extended", "error": str(exc)}, status_code=500)
|