82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
"""Evaluation Harness Runner for GCP Solution Architecture Agent."""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from app.config import get_settings
|
|
from app.skills.loader import SkillLoader
|
|
from app.workflows.gcp_architecture_graph import create_agent
|
|
from eval.metrics import evaluate_case_run
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EvalHarness:
|
|
"""Offline Evaluation Harness for running benchmark suites against agent workflows."""
|
|
|
|
def __init__(self, dataset_path: Path | None = None) -> None:
|
|
settings = get_settings()
|
|
self.dataset_path = dataset_path or settings.EVAL_DATASET_PATH
|
|
self.skill_loader = SkillLoader(settings.SKILLS_DIR)
|
|
self.skill_loader.load_skills()
|
|
self.agent = create_agent(self.skill_loader)
|
|
|
|
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 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 benchmark case: %s", case.get("id"))
|
|
initial_state = {
|
|
"workflow_request": case.get("workflow_request", ""),
|
|
"target_dir": ".",
|
|
"active_skills": [],
|
|
}
|
|
|
|
final_state = self.agent.invoke(initial_state)
|
|
eval_result = evaluate_case_run(final_state, 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 Benchmark Summary ===")
|
|
print(json.dumps(summary, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|