74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
"""Evaluation Harness Runner for GCP Solution Architecture Agent.
|
|
|
|
Uses google.adk.evaluation.Evaluator with PostgreSQL database metrics persistence.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from app.adk.evaluation import ADKEvaluator
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EvalHarness:
|
|
"""Offline Evaluation Harness for running benchmark suites against ADK agents."""
|
|
|
|
def __init__(self, dataset_path: Path | None = None) -> None:
|
|
settings = get_settings()
|
|
self.dataset_path = dataset_path or settings.EVAL_DATASET_PATH
|
|
self.evaluator = ADKEvaluator()
|
|
|
|
def load_benchmark_cases(self) -> List[Dict[str, Any]]:
|
|
"""Load benchmark dataset JSON."""
|
|
if not self.dataset_path.is_file():
|
|
logger.error("Benchmark dataset not found at %s", self.dataset_path)
|
|
return []
|
|
|
|
with open(self.dataset_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
def run_eval_suite(self) -> Dict[str, Any]:
|
|
"""Execute all benchmark test cases through ADKEvaluator and compile scoring metrics."""
|
|
cases = self.load_benchmark_cases()
|
|
if not cases:
|
|
return {"status": "error", "message": "No benchmark cases loaded."}
|
|
|
|
results = []
|
|
total_passed = 0
|
|
|
|
for case in cases:
|
|
logger.info("Evaluating ADK benchmark case: %s", case.get("id"))
|
|
eval_result = self.evaluator.evaluate_benchmark_case(case)
|
|
results.append(eval_result)
|
|
|
|
if eval_result.get("passed"):
|
|
total_passed += 1
|
|
|
|
pass_rate = (total_passed / len(cases)) * 100.0 if cases else 0.0
|
|
|
|
summary = {
|
|
"total_cases": len(cases),
|
|
"passed_cases": total_passed,
|
|
"failed_cases": len(cases) - total_passed,
|
|
"pass_rate_percentage": pass_rate,
|
|
"results": results,
|
|
}
|
|
|
|
return summary
|
|
|
|
|
|
def main() -> None:
|
|
"""CLI Runner for Evaluation Harness."""
|
|
harness = EvalHarness()
|
|
summary = harness.run_eval_suite()
|
|
print("=== GCP Solution Architecture Agent ADK Benchmark Summary ===")
|
|
print(json.dumps(summary, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|