run_eval.py

#!/usr/bin/env python3
"""
Gaslight Eval - Main Evaluation Runner

Run evaluations of LLM epistemic robustness under adversarial self-history manipulation.
Uses OpenRouter for unified access to 70+ models including frontier, Chinese, and open source.

Usage:
    python run_eval.py --models claude-3.5-sonnet,gpt-4o --scenarios 5 --output results.json
    python run_eval.py --preset recommended --scenarios 10
    python run_eval.py --category chinese --scenarios 5
    python run_eval.py --list-models
"""

import argparse
import json
import logging
import os
import sys
from datetime import datetime
from typing import List, Optional

# Add the package to path if running directly
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from gaslight_eval import (
    AmbiguityLevel,
    ScenarioType,
    ManipulationDepth,
    EvalScenario,
    EvalResult,
    ScenarioGenerator,
    ConversationManipulator,
    ModelInterface,
    ResponseScorer,
    LLMJudge,
)
from gaslight_eval.models import EvalRun
from gaslight_eval.manipulation import ManipulationVariant
from gaslight_eval.interface import (
    get_recommended_eval_set,
    get_budget_eval_set,
    get_reasoning_eval_set,
    MODEL_CONFIGS,
)


def setup_logging(verbose: bool = False) -> None:
    """Set up logging configuration."""
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )


def parse_models(model_str: str) -> List[str]:
    """Parse comma-separated model list."""
    return [m.strip() for m in model_str.split(",") if m.strip()]


def parse_ambiguity_levels(level_str: str) -> List[AmbiguityLevel]:
    """Parse ambiguity level specification."""
    if level_str.lower() == "all":
        return list(AmbiguityLevel)

    levels = []
    for part in level_str.split(","):
        part = part.strip().upper()
        try:
            if part.isdigit():
                levels.append(AmbiguityLevel(int(part)))
            else:
                levels.append(AmbiguityLevel[part])
        except (KeyError, ValueError):
            logging.warning(f"Unknown ambiguity level: {part}")

    return levels or [AmbiguityLevel.NONE]


def get_models_from_args(args) -> List[str]:
    """Determine which models to use based on CLI arguments."""
    models = []

    # Check for preset
    if args.preset:
        preset_map = {
            "recommended": get_recommended_eval_set,
            "budget": get_budget_eval_set,
            "reasoning": get_reasoning_eval_set,
        }
        if args.preset in preset_map:
            models = preset_map[args.preset]()
        else:
            raise ValueError(f"Unknown preset: {args.preset}. Use: {', '.join(preset_map.keys())}")

    # Check for category
    elif args.category:
        models = ModelInterface.get_models_by_category(args.category)
        if not models:
            raise ValueError(f"No models found for category: {args.category}")

    # Use explicit model list
    elif args.models:
        models = parse_models(args.models)

    return models


def list_models_detailed():
    """List all available models with detailed information."""
    all_models = ModelInterface.list_all_models()

    print("\n" + "=" * 70)
    print("AVAILABLE MODELS (via OpenRouter)")
    print("=" * 70)

    # Frontier models
    print("\n🏆 FRONTIER / SOTA MODELS")
    print("-" * 70)
    for model in all_models.get("frontier", []):
        print(f"  {model['key']:30s} → {model['display_name']}")

    # Chinese models
    print("\n🇨🇳 CHINESE MODELS")
    print("-" * 70)
    for model in all_models.get("chinese", []):
        print(f"  {model['key']:30s} → {model['display_name']}")

    # Open source models
    print("\n🔓 OPEN SOURCE MODELS")
    print("-" * 70)
    for model in all_models.get("opensource", []):
        print(f"  {model['key']:30s} → {model['display_name']}")

    # Presets
    print("\n📦 MODEL PRESETS")
    print("-" * 70)
    print("  --preset recommended  : Top models from each category (14 models)")
    print("  --preset budget       : Cost-effective models (10 models)")
    print("  --preset reasoning    : Strong reasoning models (7 models)")

    # Categories
    print("\n📁 CATEGORIES")
    print("-" * 70)
    print("  --category frontier   : All frontier/SOTA models")
    print("  --category chinese    : All Chinese models")
    print("  --category opensource : All open source models")

    print("\n💡 Set OPENROUTER_API_KEY environment variable to use these models")
    print("=" * 70 + "\n")


def run_evaluation(
    models: List[str],
    scenarios_per_level: int,
    ambiguity_levels: List[AmbiguityLevel],
    output_path: str,
    judge_model: Optional[str] = None,
    include_inverted: bool = True,
    include_impossible: bool = True,
    seed: Optional[int] = None,
    dry_run: bool = False,
    manipulation_depth: ManipulationDepth = ManipulationDepth.MODERATE,
    fabricated_turns: int = 3,
    include_reasoning: bool = True,
) -> EvalRun:
    """
    Run the complete evaluation.

    Args:
        models: List of model names to evaluate
        scenarios_per_level: Number of scenarios per ambiguity level
        ambiguity_levels: List of ambiguity levels to test
        output_path: Path to save results
        judge_model: Model to use as judge (default: first available)
        include_inverted: Whether to include inverted control scenarios
        include_impossible: Whether to include impossible fabrication scenarios
        seed: Random seed for reproducibility
        dry_run: If True, generate scenarios but don't call APIs
        manipulation_depth: Level of manipulation to apply (NONE, MINIMAL, MODERATE, EXTREME)
        fabricated_turns: Number of fabricated turns for EXTREME depth (1-5)
        include_reasoning: Whether to include fake reasoning traces

    Returns:
        EvalRun with all results
    """
    logger = logging.getLogger(__name__)

    # Initialize run
    run = EvalRun.create(
        name=f"gaslight_eval_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
        description=f"Evaluation of {len(models)} models across {len(ambiguity_levels)} ambiguity levels",
        models=models,
        config={
            "scenarios_per_level": scenarios_per_level,
            "ambiguity_levels": [l.name for l in ambiguity_levels],
            "include_inverted": include_inverted,
            "include_impossible": include_impossible,
            "seed": seed,
            "judge_model": judge_model,
            "manipulation_depth": manipulation_depth.name,
            "fabricated_turns": fabricated_turns,
            "include_reasoning": include_reasoning,
        }
    )

    # Generate scenarios
    logger.info("Generating evaluation scenarios...")
    generator = ScenarioGenerator(seed=seed)
    scenarios = generator.generate_scenario_set(
        ambiguity_levels=ambiguity_levels,
        scenarios_per_level=scenarios_per_level,
        include_inverted=include_inverted,
        include_impossible=include_impossible,
    )

    # Apply manipulation to all scenarios
    depth_names = {0: "NONE", 1: "MINIMAL", 2: "MODERATE", 3: "EXTREME"}
    logger.info(f"Generated {len(scenarios)} scenarios, applying {depth_names.get(manipulation_depth.value, 'MODERATE')} manipulation...")
    manipulated_scenarios = ManipulationVariant.create_batch_manipulated_scenarios(
        scenarios,
        include_reasoning_trace=include_reasoning,
        manipulation_depth=manipulation_depth,
        fabricated_turns=fabricated_turns,
    )

    for scenario in manipulated_scenarios:
        run.add_scenario(scenario)

    if dry_run:
        logger.info("Dry run - saving scenarios without API calls")
        run.save(output_path)
        return run

    # Initialize model interfaces
    logger.info("Initializing model interfaces (via OpenRouter)...")
    model_interfaces = {}
    for model_name in models:
        try:
            interface = ModelInterface(model_name)
            model_interfaces[model_name] = interface
            display = interface.display_name if hasattr(interface, 'display_name') else model_name
            logger.info(f"  ✓ {model_name} ({display})")
        except Exception as e:
            logger.error(f"  ✗ {model_name}: {e}")

    if not model_interfaces:
        raise RuntimeError("No models could be initialized. Check your OPENROUTER_API_KEY.")

    # Initialize judge
    judge = None
    if judge_model:
        try:
            judge = LLMJudge(judge_model)
            logger.info(f"Initialized judge: {judge_model}")
        except Exception as e:
            logger.warning(f"Could not initialize judge {judge_model}: {e}")

    if judge is None and model_interfaces:
        # Use a fast, capable model as judge
        judge_candidates = ["claude-3.5-haiku", "gpt-4o-mini", "gemini-1.5-flash"]
        for candidate in judge_candidates:
            if candidate in model_interfaces or candidate in MODEL_CONFIGS:
                try:
                    judge = LLMJudge(candidate)
                    logger.info(f"Using {candidate} as judge")
                    break
                except Exception as e:
                    continue

        if judge is None:
            # Fall back to first available model
            first_model = list(model_interfaces.keys())[0]
            try:
                judge = LLMJudge(first_model)
                logger.info(f"Using {first_model} as judge")
            except Exception as e:
                logger.warning(f"Could not initialize any judge: {e}")

    # Initialize pattern-based scorer as fallback
    pattern_scorer = ResponseScorer()

    # Run evaluations
    total_evals = len(manipulated_scenarios) * len(model_interfaces)
    completed = 0

    logger.info(f"Starting evaluation: {total_evals} total evaluations")

    manipulator = ConversationManipulator()

    for scenario in manipulated_scenarios:
        for model_name, interface in model_interfaces.items():
            try:
                logger.debug(f"Running {model_name} on {scenario.scenario_id}")

                # Get model response
                response, latency_ms, token_count = interface.run_eval(
                    scenario, manipulator
                )

                # Score with judge if available
                if judge:
                    try:
                        judge_result = judge.evaluate(
                            response=response,
                            family_decision=scenario.family_decision,
                            fabricated_action=scenario.fabricated_model_action,
                        )
                    except Exception as e:
                        logger.warning(f"Judge failed, using pattern scoring: {e}")
                        pattern_result = pattern_scorer.score_response(response)
                        from gaslight_eval.models import JudgeResult
                        judge_result = JudgeResult(
                            acceptance_score=pattern_result.acceptance_score,
                            rationalization_score=pattern_result.rationalization_score,
                            coherence_score=pattern_result.coherence_score,
                            category=pattern_result.category,
                            reasoning="Pattern-based scoring (judge unavailable)",
                            confidence=pattern_result.confidence,
                        )
                else:
                    # Use pattern-based scoring
                    pattern_result = pattern_scorer.score_response(response)
                    from gaslight_eval.models import JudgeResult
                    judge_result = JudgeResult(
                        acceptance_score=pattern_result.acceptance_score,
                        rationalization_score=pattern_result.rationalization_score,
                        coherence_score=pattern_result.coherence_score,
                        category=pattern_result.category,
                        reasoning="Pattern-based scoring",
                        confidence=pattern_result.confidence,
                    )

                # Create result
                result = EvalResult.create(
                    scenario_id=scenario.scenario_id,
                    model_name=model_name,
                    model_provider=interface.provider,
                    model_response=response,
                    judge_result=judge_result,
                    latency_ms=latency_ms,
                    token_count=token_count,
                    metadata={
                        "ambiguity_level": scenario.ambiguity_level.name,
                        "scenario_type": scenario.scenario_type.name,
                        "model_category": interface.category if hasattr(interface, 'category') else "",
                        "manipulation_depth": scenario.manipulation_depth.value,
                        "fabricated_turns": scenario.fabricated_turns,
                    }
                )
                result.confidence_markers = pattern_scorer.score_response(response).matched_markers

                run.add_result(result)
                completed += 1

                if completed % 10 == 0:
                    logger.info(f"Progress: {completed}/{total_evals} ({100*completed//total_evals}%)")

                    # Save intermediate results
                    run.save(output_path)

            except Exception as e:
                logger.error(f"Error evaluating {model_name} on {scenario.scenario_id}: {e}")
                completed += 1

    # Finalize and save
    run.complete()
    run.save(output_path)

    logger.info(f"Evaluation complete. Results saved to {output_path}")
    logger.info(f"Total results: {len(run.results)}")

    return run


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Gaslight Eval - Test LLM epistemic robustness via OpenRouter",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Run with specific models
  python run_eval.py --models claude-3.5-sonnet,gpt-4o,deepseek-chat --scenarios 5

  # Use a preset model set
  python run_eval.py --preset recommended --scenarios 5
  python run_eval.py --preset budget --scenarios 10
  python run_eval.py --preset reasoning --scenarios 3

  # Run all models in a category
  python run_eval.py --category chinese --scenarios 5
  python run_eval.py --category frontier --scenarios 3

  # Different manipulation depths
  python run_eval.py --depth none --scenarios 3      # Baseline (no manipulation)
  python run_eval.py --depth minimal --scenarios 3   # Minimal manipulation
  python run_eval.py --depth moderate --scenarios 3  # Default (single fabricated action)
  python run_eval.py --depth extreme --fabricated-turns 3 --scenarios 3  # Multi-turn

  # List all available models
  python run_eval.py --list-models

  # Dry run (generate scenarios without API calls)
  python run_eval.py --dry-run --scenarios 3 --output test_scenarios.json

Environment:
  OPENROUTER_API_KEY    Required for API access
        """,
    )

    # Model selection (mutually exclusive)
    model_group = parser.add_mutually_exclusive_group()
    model_group.add_argument(
        "--models",
        type=str,
        help="Comma-separated list of models to evaluate",
    )
    model_group.add_argument(
        "--preset",
        type=str,
        choices=["recommended", "budget", "reasoning"],
        help="Use a preset model set",
    )
    model_group.add_argument(
        "--category",
        type=str,
        choices=["frontier", "chinese", "opensource"],
        help="Use all models in a category",
    )

    parser.add_argument(
        "--scenarios",
        type=int,
        default=5,
        help="Number of scenarios per ambiguity level (default: 5)",
    )
    parser.add_argument(
        "--ambiguity",
        type=str,
        default="all",
        help="Ambiguity levels to test (all, or comma-separated: 0,1,2,3,4 or NONE,MINIMAL,...)",
    )
    parser.add_argument(
        "--output",
        "-o",
        type=str,
        default="eval_results.json",
        help="Output file path (default: eval_results.json)",
    )
    parser.add_argument(
        "--judge",
        type=str,
        default=None,
        help="Model to use as LLM judge (default: claude-3.5-haiku or first available)",
    )
    parser.add_argument(
        "--no-inverted",
        action="store_true",
        help="Skip inverted control scenarios",
    )
    parser.add_argument(
        "--no-impossible",
        action="store_true",
        help="Skip impossible fabrication scenarios",
    )
    parser.add_argument(
        "--depth",
        type=str,
        choices=["none", "minimal", "moderate", "extreme"],
        default="moderate",
        help="Manipulation depth level (default: moderate)",
    )
    parser.add_argument(
        "--fabricated-turns",
        type=int,
        default=3,
        choices=range(1, 6),
        metavar="{1-5}",
        help="Number of fabricated turns for EXTREME depth (default: 3)",
    )
    parser.add_argument(
        "--no-reasoning",
        action="store_true",
        help="Don't include fake reasoning traces in fabricated responses",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=None,
        help="Random seed for reproducibility",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Generate scenarios without calling APIs",
    )
    parser.add_argument(
        "--list-models",
        action="store_true",
        help="List all available models and presets",
    )
    parser.add_argument(
        "--verbose",
        "-v",
        action="store_true",
        help="Enable verbose logging",
    )

    args = parser.parse_args()

    setup_logging(args.verbose)

    if args.list_models:
        list_models_detailed()
        return

    # Get models from arguments
    try:
        models = get_models_from_args(args)
    except ValueError as e:
        print(f"Error: {e}")
        sys.exit(1)

    # Default to a single model if nothing specified
    if not models:
        models = ["claude-3.5-sonnet"]
        print("No models specified, defaulting to claude-3.5-sonnet")

    # Check for API key
    if not args.dry_run and not os.environ.get("OPENROUTER_API_KEY"):
        print("\n⚠️  Warning: OPENROUTER_API_KEY not set!")
        print("Set it with: export OPENROUTER_API_KEY=your_key_here")
        print("Get a key at: https://openrouter.ai/keys\n")

    ambiguity_levels = parse_ambiguity_levels(args.ambiguity)

    # Parse manipulation depth
    depth_map = {
        "none": ManipulationDepth.NONE,
        "minimal": ManipulationDepth.MINIMAL,
        "moderate": ManipulationDepth.MODERATE,
        "extreme": ManipulationDepth.EXTREME,
    }
    manipulation_depth = depth_map.get(args.depth.lower(), ManipulationDepth.MODERATE)

    print(f"\n{'='*70}")
    print("GASLIGHT EVAL - LLM Epistemic Robustness Testing")
    print(f"{'='*70}")
    print(f"Provider: OpenRouter")
    print(f"Models ({len(models)}): {', '.join(models[:5])}{'...' if len(models) > 5 else ''}")
    print(f"Ambiguity levels: {', '.join(l.name for l in ambiguity_levels)}")
    print(f"Scenarios per level: {args.scenarios}")
    print(f"Manipulation depth: {manipulation_depth.name}")
    if manipulation_depth == ManipulationDepth.EXTREME:
        print(f"Fabricated turns: {args.fabricated_turns}")
    print(f"Output: {args.output}")
    print(f"{'='*70}\n")

    try:
        run = run_evaluation(
            models=models,
            scenarios_per_level=args.scenarios,
            ambiguity_levels=ambiguity_levels,
            output_path=args.output,
            judge_model=args.judge,
            include_inverted=not args.no_inverted,
            include_impossible=not args.no_impossible,
            seed=args.seed,
            dry_run=args.dry_run,
            manipulation_depth=manipulation_depth,
            fabricated_turns=args.fabricated_turns,
            include_reasoning=not args.no_reasoning,
        )

        # Print summary
        print(f"\n{'='*70}")
        print("EVALUATION COMPLETE")
        print(f"{'='*70}")
        print(f"Total scenarios: {len(run.scenarios)}")
        print(f"Total results: {len(run.results)}")
        print(f"Output saved to: {args.output}")
        print(f"\nAnalyze with: python analyze.py --input {args.output} --summary")

    except KeyboardInterrupt:
        print("\n\nEvaluation interrupted by user.")
        sys.exit(1)
    except Exception as e:
        logging.error(f"Evaluation failed: {e}")
        raise


if __name__ == "__main__":
    main()