learningcircuit--local-deep-research/examples/optimization/run_gemini_benchmark.py

#!/usr/bin/env python
"""
Run benchmarks with Gemini Flash via OpenRouter.

This script updates the database LLM configuration and then runs benchmarks
with Gemini Flash via OpenRouter.

Usage:
    # Install dependencies with PDM
    cd /path/to/local-deep-research
    pdm install

    # Run the script with PDM
    pdm run python examples/optimization/run_gemini_benchmark.py --api-key "your-openrouter-api-key" --examples 10
"""

import argparse
import sys
import time
from datetime import datetime, UTC
from pathlib import Path
from typing import Any, Dict, List, Optional

from loguru import logger

# Add the src directory to the Python path
project_root = str(Path(__file__).parent.parent.parent.resolve())
sys.path.insert(0, str(Path(project_root) / "src"))

# Loguru automatically handles logging configuration


def setup_gemini_config(api_key: Optional[str] = None) -> Dict[str, Any]:
    """
    Create a configuration for using Gemini Flash via OpenRouter.

    Args:
        api_key: OpenRouter API key (optional, will try to get from database if not provided)

    Returns:
        Dictionary with Gemini configuration
    """
    # Import database utilities
    from local_deep_research.utilities.db_utils import (
        get_db_setting,
        update_db_setting,
    )

    # Check if API key exists in database
    if not api_key:
        api_key = get_db_setting("llm.openai_endpoint.api_key")
        if not api_key:
            logger.error("No API key found in database and none provided")
            return {}

    # Create configuration
    config = {
        "model_name": "google/gemini-2.0-flash",
        "provider": "openai_endpoint",
        "endpoint_url": "https://openrouter.ai/api/v1",
        "api_key": api_key,
    }

    # Update database with this configuration
    update_db_setting("llm.model", config["model_name"])
    update_db_setting("llm.provider", config["provider"])
    update_db_setting("llm.openai_endpoint.url", config["endpoint_url"])
    update_db_setting("llm.openai_endpoint.api_key", config["api_key"])

    # Log configuration
    logger.info("LLM configuration updated to use Gemini Flash via OpenRouter")
    logger.info(f"Model: {config['model_name']}")
    logger.info(f"Provider: {config['provider']}")

    return config


def run_benchmarks(
    examples: int = 5,
    benchmarks: List[str] = None,
    api_key: Optional[str] = None,
    output_dir: Optional[str] = None,
    search_iterations: int = 2,
    questions_per_iteration: int = 3,
    search_tool: str = "searxng",
) -> Dict[str, Any]:
    """
    Run benchmarks with Gemini Flash via OpenRouter.

    Args:
        examples: Number of examples to evaluate for each benchmark
        benchmarks: List of benchmarks to run (defaults to ["simpleqa", "browsecomp"])
        api_key: OpenRouter API key
        output_dir: Directory to save results
        search_iterations: Number of search iterations per query
        questions_per_iteration: Number of questions per iteration
        search_tool: Search engine to use

    Returns:
        Dictionary with benchmark results
    """
    # Import benchmark functions
    from local_deep_research.benchmarks.benchmark_functions import (
        evaluate_browsecomp,
        evaluate_simpleqa,
    )

    # Set up Gemini configuration
    gemini_config = setup_gemini_config(api_key)
    if not gemini_config:
        return {"error": "Failed to set up Gemini configuration"}

    # Create timestamp for output
    timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
    if not output_dir:
        output_dir = str(
            Path(project_root)
            / "benchmark_results"
            / f"gemini_eval_{timestamp}"
        )

    Path(output_dir).mkdir(parents=True, exist_ok=True)

    # Set benchmark list
    if not benchmarks:
        benchmarks = ["simpleqa", "browsecomp"]

    results = {}

    # Run each benchmark
    for benchmark in benchmarks:
        start_time = time.time()

        try:
            if benchmark.lower() == "simpleqa":
                logger.info(
                    f"Running SimpleQA benchmark with {examples} examples"
                )
                benchmark_results = evaluate_simpleqa(
                    num_examples=examples,
                    search_iterations=search_iterations,
                    questions_per_iteration=questions_per_iteration,
                    search_tool=search_tool,
                    search_model=gemini_config["model_name"],
                    search_provider=gemini_config["provider"],
                    endpoint_url=gemini_config["endpoint_url"],
                    output_dir=str(Path(output_dir) / "simpleqa"),
                )
            elif benchmark.lower() == "browsecomp":
                logger.info(
                    f"Running BrowseComp benchmark with {examples} examples"
                )
                benchmark_results = evaluate_browsecomp(
                    num_examples=examples,
                    search_iterations=search_iterations,
                    questions_per_iteration=questions_per_iteration,
                    search_tool=search_tool,
                    search_model=gemini_config["model_name"],
                    search_provider=gemini_config["provider"],
                    endpoint_url=gemini_config["endpoint_url"],
                    output_dir=str(Path(output_dir) / "browsecomp"),
                )
            else:
                logger.warning(f"Unknown benchmark: {benchmark}")
                continue

            duration = time.time() - start_time

            # Log results
            logger.info(
                f"{benchmark} benchmark completed in {duration:.1f} seconds"
            )
            if isinstance(benchmark_results, dict):
                accuracy = benchmark_results.get("accuracy", "N/A")
                logger.info(f"{benchmark} accuracy: {accuracy}")

            # Add to results
            results[benchmark] = {
                "results": benchmark_results,
                "duration": duration,
            }

        except Exception as e:
            logger.exception(f"Error running {benchmark} benchmark")
            import traceback

            traceback.print_exc()

            results[benchmark] = {
                "error": str(e),
            }

    # Generate summary
    logger.info("=" * 50)
    logger.info("BENCHMARK SUMMARY")
    logger.info("=" * 50)
    logger.info(f"Model: {gemini_config.get('model_name')}")
    logger.info(f"Examples per benchmark: {examples}")

    for benchmark, benchmark_results in results.items():
        if "error" in benchmark_results:
            logger.info(f"{benchmark}: ERROR - {benchmark_results['error']}")
        else:
            accuracy = benchmark_results.get("results", {}).get(
                "accuracy", "N/A"
            )
            duration = benchmark_results.get("duration", 0)
            logger.info(
                f"{benchmark}: Accuracy = {accuracy}, Duration = {duration:.1f}s"
            )

    logger.info(f"Results saved to: {output_dir}")
    logger.info("=" * 50)

    # Save summary to a file
    summary_file = str(Path(output_dir) / "benchmark_summary.json")
    try:
        import json

        with open(summary_file, "w", encoding="utf-8") as f:
            json.dump(
                {
                    "timestamp": timestamp,
                    "model": gemini_config.get("model_name"),
                    "provider": gemini_config.get("provider"),
                    "examples": examples,
                    "benchmarks": [b for b in benchmarks],
                    "results": {
                        b: {
                            "accuracy": (
                                r.get("results", {}).get("accuracy", None)
                                if "error" not in r
                                else None
                            ),
                            "duration": r.get("duration", 0)
                            if "error" not in r
                            else 0,
                            "error": r.get("error", None)
                            if "error" in r
                            else None,
                        }
                        for b, r in results.items()
                    },
                },
                f,
                indent=2,
            )
        logger.info(f"Summary saved to {summary_file}")
    except Exception:
        logger.exception("Error saving summary")

    return {
        "status": "complete",
        "results": results,
        "output_dir": output_dir,
    }


def main():
    """Main function to parse arguments and run benchmarks."""
    parser = argparse.ArgumentParser(
        description="Run benchmarks with Gemini Flash via OpenRouter"
    )

    # Benchmark configuration
    parser.add_argument(
        "--examples",
        type=int,
        default=5,
        help="Number of examples for each benchmark",
    )
    parser.add_argument(
        "--benchmarks",
        nargs="+",
        choices=["simpleqa", "browsecomp"],
        help="Benchmarks to run (default: both)",
    )
    parser.add_argument(
        "--search-iterations",
        type=int,
        default=2,
        help="Number of search iterations",
    )
    parser.add_argument(
        "--questions-per-iteration",
        type=int,
        default=3,
        help="Questions per iteration",
    )
    parser.add_argument(
        "--search-tool", default="searxng", help="Search tool to use"
    )

    # API key
    parser.add_argument(
        "--api-key", help="OpenRouter API key (optional if already in database)"
    )

    # Output directory
    parser.add_argument(
        "--output-dir", help="Directory to save results (optional)"
    )

    args = parser.parse_args()

    # Run benchmarks
    results = run_benchmarks(
        examples=args.examples,
        benchmarks=args.benchmarks,
        api_key=args.api_key,
        output_dir=args.output_dir,
        search_iterations=args.search_iterations,
        questions_per_iteration=args.questions_per_iteration,
        search_tool=args.search_tool,
    )

    return 0 if results.get("status") == "complete" else 1


if __name__ == "__main__":
    sys.exit(main())