mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-08-28 05:25:25 +00:00
470a3377b3
- Fix imports from eval.agents/eval.environments to relative (agents/environments) - Add hatch wheel config for correct package discovery - Extract git diff patch from container for SWE-bench submission - Use sys.executable instead of hardcoded "python" for venv compat - Upgrade claude-haiku config to 4.5 - Add minimax-m2.1 model config Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
515 lines
19 KiB
Python
515 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GitNexus SWE-bench Evaluation Runner
|
|
|
|
Main entry point for running SWE-bench evaluations with and without GitNexus.
|
|
Supports running a single configuration or a full matrix of models x modes.
|
|
|
|
Usage:
|
|
# Single run (default: native_augment mode — GitNexus tools + grep enrichment)
|
|
python run_eval.py single -m claude-sonnet --subset lite --slice 0:5
|
|
|
|
# Baseline comparison (no GitNexus)
|
|
python run_eval.py single -m claude-sonnet --mode baseline --subset lite --slice 0:5
|
|
|
|
# Matrix run (all models x all modes)
|
|
python run_eval.py matrix --subset lite --slice 0:50 --workers 4
|
|
|
|
# Single instance for debugging
|
|
python run_eval.py debug -m claude-haiku -i django__django-16527
|
|
"""
|
|
|
|
import concurrent.futures
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from itertools import product
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import typer
|
|
import yaml
|
|
from rich.console import Console
|
|
from rich.live import Live
|
|
from rich.table import Table
|
|
|
|
# Load .env file from eval/ directory
|
|
_env_file = Path(__file__).parent / ".env"
|
|
if _env_file.exists():
|
|
for line in _env_file.read_text().splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if "=" in line:
|
|
key, _, value = line.partition("=")
|
|
key, value = key.strip(), value.strip()
|
|
if value and key not in os.environ: # Don't override existing env vars
|
|
os.environ[key] = value
|
|
|
|
logger = logging.getLogger("gitnexus_eval")
|
|
console = Console()
|
|
app = typer.Typer(rich_markup_mode="rich", add_completion=False)
|
|
|
|
# Directory paths
|
|
EVAL_DIR = Path(__file__).parent
|
|
CONFIGS_DIR = EVAL_DIR / "configs"
|
|
MODELS_DIR = CONFIGS_DIR / "models"
|
|
MODES_DIR = CONFIGS_DIR / "modes"
|
|
DEFAULT_OUTPUT_DIR = EVAL_DIR / "results"
|
|
|
|
# Available models and modes (discovered from config files)
|
|
AVAILABLE_MODELS = sorted([p.stem for p in MODELS_DIR.glob("*.yaml")])
|
|
AVAILABLE_MODES = sorted([p.stem for p in MODES_DIR.glob("*.yaml")])
|
|
|
|
# SWE-bench dataset mapping (same as mini-swe-agent)
|
|
DATASET_MAPPING = {
|
|
"full": "princeton-nlp/SWE-Bench",
|
|
"verified": "princeton-nlp/SWE-Bench_Verified",
|
|
"lite": "princeton-nlp/SWE-Bench_Lite",
|
|
}
|
|
|
|
_output_lock = threading.Lock()
|
|
|
|
|
|
def load_yaml_config(path: Path) -> dict:
|
|
"""Load a YAML config file."""
|
|
with open(path) as f:
|
|
return yaml.safe_load(f) or {}
|
|
|
|
|
|
def merge_configs(*configs: dict) -> dict:
|
|
"""Recursively merge multiple config dicts (later values win)."""
|
|
result = {}
|
|
for config in configs:
|
|
for key, value in config.items():
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = merge_configs(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def build_config(model_name: str, mode_name: str) -> dict:
|
|
"""Build a complete config from model + mode YAML files."""
|
|
model_file = MODELS_DIR / f"{model_name}.yaml"
|
|
mode_file = MODES_DIR / f"{mode_name}.yaml"
|
|
|
|
if not model_file.exists():
|
|
raise FileNotFoundError(f"Model config not found: {model_file}")
|
|
if not mode_file.exists():
|
|
raise FileNotFoundError(f"Mode config not found: {mode_file}")
|
|
|
|
model_config = load_yaml_config(model_file)
|
|
mode_config = load_yaml_config(mode_file)
|
|
|
|
return merge_configs(mode_config, model_config)
|
|
|
|
|
|
def load_instances(subset: str, split: str, slice_spec: str = "", filter_spec: str = "") -> list[dict]:
|
|
"""Load SWE-bench instances."""
|
|
from datasets import load_dataset
|
|
import re
|
|
|
|
dataset_path = DATASET_MAPPING.get(subset, subset)
|
|
logger.info(f"Loading dataset: {dataset_path}, split: {split}")
|
|
instances = list(load_dataset(dataset_path, split=split))
|
|
|
|
if filter_spec:
|
|
instances = [i for i in instances if re.match(filter_spec, i["instance_id"])]
|
|
|
|
if slice_spec:
|
|
values = [int(x) if x else None for x in slice_spec.split(":")]
|
|
instances = instances[slice(*values)]
|
|
|
|
logger.info(f"Loaded {len(instances)} instances")
|
|
return instances
|
|
|
|
|
|
def get_swebench_docker_image(instance: dict) -> str:
|
|
"""Get Docker image name for a SWE-bench instance."""
|
|
image_name = instance.get("image_name")
|
|
if image_name is None:
|
|
iid = instance["instance_id"]
|
|
id_docker = iid.replace("__", "_1776_")
|
|
image_name = f"docker.io/swebench/sweb.eval.x86_64.{id_docker}:latest".lower()
|
|
return image_name
|
|
|
|
|
|
def process_instance(
|
|
instance: dict,
|
|
config: dict,
|
|
output_dir: Path,
|
|
model_name: str,
|
|
mode_name: str,
|
|
) -> dict:
|
|
"""
|
|
Process a single SWE-bench instance with the given config.
|
|
Returns result dict with instance_id, exit_status, submission, metrics.
|
|
"""
|
|
from minisweagent.models import get_model
|
|
|
|
instance_id = instance["instance_id"]
|
|
run_id = f"{model_name}_{mode_name}"
|
|
instance_dir = output_dir / run_id / instance_id
|
|
instance_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
result = {
|
|
"instance_id": instance_id,
|
|
"model": model_name,
|
|
"mode": mode_name,
|
|
"exit_status": None,
|
|
"submission": "",
|
|
"cost": 0.0,
|
|
"n_calls": 0,
|
|
"gitnexus_metrics": {},
|
|
}
|
|
|
|
agent = None
|
|
|
|
try:
|
|
# Build model
|
|
model = get_model(config=config.get("model", {}))
|
|
|
|
# Build environment
|
|
env_config = dict(config.get("environment", {}))
|
|
env_class_name = env_config.pop("environment_class", "docker")
|
|
|
|
if env_class_name == "eval.environments.gitnexus_docker.GitNexusDockerEnvironment":
|
|
from environments.gitnexus_docker import GitNexusDockerEnvironment
|
|
env_config["image"] = get_swebench_docker_image(instance)
|
|
env = GitNexusDockerEnvironment(**env_config)
|
|
else:
|
|
from minisweagent.environments.docker import DockerEnvironment
|
|
env = DockerEnvironment(image=get_swebench_docker_image(instance), **env_config)
|
|
|
|
# Build agent
|
|
agent_config = dict(config.get("agent", {}))
|
|
agent_class_name = agent_config.pop("agent_class", "eval.agents.gitnexus_agent.GitNexusAgent")
|
|
|
|
from agents.gitnexus_agent import GitNexusAgent
|
|
traj_path = instance_dir / f"{instance_id}.traj.json"
|
|
agent_config["output_path"] = traj_path
|
|
agent = GitNexusAgent(model, env, **agent_config)
|
|
|
|
# Run
|
|
logger.info(f"[{run_id}] Starting {instance_id}")
|
|
info = agent.run(instance["problem_statement"])
|
|
|
|
result["exit_status"] = info.get("exit_status")
|
|
result["cost"] = agent.cost
|
|
result["n_calls"] = agent.n_calls
|
|
result["gitnexus_metrics"] = agent.gitnexus_metrics.to_dict()
|
|
|
|
# Extract git diff patch from the container (SWE-bench needs the model_patch)
|
|
try:
|
|
patch_output = env.execute({"command": "cd /testbed && git diff"})
|
|
result["submission"] = patch_output.get("output", "").strip()
|
|
except Exception as patch_err:
|
|
logger.warning(f"[{run_id}] Failed to extract patch: {patch_err}")
|
|
result["submission"] = info.get("submission", "")
|
|
|
|
except Exception as e:
|
|
logger.error(f"[{run_id}] Error on {instance_id}: {e}")
|
|
result["exit_status"] = type(e).__name__
|
|
result["error"] = str(e)
|
|
result["traceback"] = traceback.format_exc()
|
|
|
|
finally:
|
|
if agent:
|
|
agent.save(
|
|
instance_dir / f"{instance_id}.traj.json",
|
|
{"instance_id": instance_id, "run_id": run_id},
|
|
)
|
|
|
|
# Update predictions file
|
|
_update_preds(output_dir / run_id / "preds.json", instance_id, model_name, result)
|
|
|
|
return result
|
|
|
|
|
|
def _update_preds(preds_path: Path, instance_id: str, model_name: str, result: dict):
|
|
"""Thread-safe update of predictions file."""
|
|
with _output_lock:
|
|
preds_path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = {}
|
|
if preds_path.exists():
|
|
data = json.loads(preds_path.read_text())
|
|
data[instance_id] = {
|
|
"model_name_or_path": model_name,
|
|
"instance_id": instance_id,
|
|
"model_patch": result.get("submission", ""),
|
|
}
|
|
preds_path.write_text(json.dumps(data, indent=2))
|
|
|
|
|
|
def run_configuration(
|
|
model_name: str,
|
|
mode_name: str,
|
|
instances: list[dict],
|
|
output_dir: Path,
|
|
workers: int = 1,
|
|
redo_existing: bool = False,
|
|
) -> list[dict]:
|
|
"""Run a single (model, mode) configuration across all instances."""
|
|
config = build_config(model_name, mode_name)
|
|
run_id = f"{model_name}_{mode_name}"
|
|
run_dir = output_dir / run_id
|
|
|
|
# Skip existing instances
|
|
if not redo_existing and (run_dir / "preds.json").exists():
|
|
existing = set(json.loads((run_dir / "preds.json").read_text()).keys())
|
|
instances = [i for i in instances if i["instance_id"] not in existing]
|
|
if not instances:
|
|
logger.info(f"[{run_id}] All instances already completed, skipping")
|
|
return []
|
|
|
|
console.print(f" [bold]{run_id}[/bold]: {len(instances)} instances, {workers} workers")
|
|
|
|
results = []
|
|
|
|
if workers <= 1:
|
|
for instance in instances:
|
|
result = process_instance(instance, config, output_dir, model_name, mode_name)
|
|
results.append(result)
|
|
else:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
|
|
futures = {
|
|
executor.submit(
|
|
process_instance, instance, config, output_dir, model_name, mode_name
|
|
): instance["instance_id"]
|
|
for instance in instances
|
|
}
|
|
for future in concurrent.futures.as_completed(futures):
|
|
try:
|
|
results.append(future.result())
|
|
except Exception as e:
|
|
iid = futures[future]
|
|
logger.error(f"[{run_id}] Uncaught error for {iid}: {e}")
|
|
|
|
# Save run summary
|
|
summary = {
|
|
"run_id": run_id,
|
|
"model": model_name,
|
|
"mode": mode_name,
|
|
"config": config,
|
|
"total_instances": len(results),
|
|
"completed": sum(1 for r in results if r["exit_status"] not in [None, "error"]),
|
|
"total_cost": sum(r.get("cost", 0) for r in results),
|
|
"total_api_calls": sum(r.get("n_calls", 0) for r in results),
|
|
"results": results,
|
|
}
|
|
(run_dir / "summary.json").mkdir(parents=True, exist_ok=True) if not run_dir.exists() else None
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "summary.json").write_text(json.dumps(summary, indent=2, default=str))
|
|
|
|
return results
|
|
|
|
|
|
# ─── CLI Commands ───────────────────────────────────────────────────────────
|
|
|
|
|
|
@app.command()
|
|
def single(
|
|
model: str = typer.Option(..., "-m", "--model", help=f"Model config name. Available: {', '.join(AVAILABLE_MODELS)}"),
|
|
mode: str = typer.Option("native_augment", "--mode", help=f"Evaluation mode. Available: {', '.join(AVAILABLE_MODES)}"),
|
|
subset: str = typer.Option("lite", "--subset", help="SWE-bench subset: lite, verified, full"),
|
|
split: str = typer.Option("dev", "--split", help="Dataset split"),
|
|
slice_spec: str = typer.Option("", "--slice", help="Slice spec (e.g., '0:5')"),
|
|
filter_spec: str = typer.Option("", "--filter", help="Filter instance IDs by regex"),
|
|
workers: int = typer.Option(1, "-w", "--workers", help="Parallel workers"),
|
|
output: str = typer.Option(str(DEFAULT_OUTPUT_DIR), "-o", "--output", help="Output directory"),
|
|
redo: bool = typer.Option(False, "--redo", help="Redo existing instances"),
|
|
):
|
|
"""Run a single (model, mode) configuration on SWE-bench."""
|
|
output_dir = Path(output)
|
|
instances = load_instances(subset, split, slice_spec, filter_spec)
|
|
|
|
console.print(f"\n[bold]Running evaluation:[/bold] {model} + {mode}")
|
|
console.print(f" Instances: {len(instances)}")
|
|
console.print(f" Output: {output_dir}\n")
|
|
|
|
results = run_configuration(model, mode, instances, output_dir, workers, redo)
|
|
|
|
# Print summary
|
|
_print_summary(results, model, mode)
|
|
|
|
|
|
@app.command()
|
|
def matrix(
|
|
models: list[str] = typer.Option(AVAILABLE_MODELS, "-m", "--models", help="Models to evaluate (comma-separated or repeated)"),
|
|
modes: list[str] = typer.Option(AVAILABLE_MODES, "--modes", help="Modes to evaluate"),
|
|
subset: str = typer.Option("lite", "--subset", help="SWE-bench subset"),
|
|
split: str = typer.Option("dev", "--split", help="Dataset split"),
|
|
slice_spec: str = typer.Option("", "--slice", help="Slice spec"),
|
|
filter_spec: str = typer.Option("", "--filter", help="Filter instances by regex"),
|
|
workers: int = typer.Option(1, "-w", "--workers", help="Parallel workers per config"),
|
|
output: str = typer.Option(str(DEFAULT_OUTPUT_DIR), "-o", "--output", help="Output directory"),
|
|
redo: bool = typer.Option(False, "--redo", help="Redo existing instances"),
|
|
):
|
|
"""Run the full evaluation matrix: all models x all modes."""
|
|
output_dir = Path(output)
|
|
instances = load_instances(subset, split, slice_spec, filter_spec)
|
|
|
|
combos = list(product(models, modes))
|
|
console.print(f"\n[bold]Matrix evaluation:[/bold] {len(models)} models x {len(modes)} modes = {len(combos)} configs")
|
|
console.print(f" Models: {', '.join(models)}")
|
|
console.print(f" Modes: {', '.join(modes)}")
|
|
console.print(f" Instances per config: {len(instances)}")
|
|
console.print(f" Total runs: {len(combos) * len(instances)}")
|
|
console.print(f" Output: {output_dir}\n")
|
|
|
|
all_results = {}
|
|
for model_name, mode_name in combos:
|
|
run_id = f"{model_name}_{mode_name}"
|
|
console.print(f"\n[bold cyan]━━━ {run_id} ━━━[/bold cyan]")
|
|
results = run_configuration(model_name, mode_name, instances, output_dir, workers, redo)
|
|
all_results[run_id] = results
|
|
|
|
# Print comparative summary
|
|
_print_matrix_summary(all_results)
|
|
|
|
# Save master summary
|
|
master = {
|
|
"timestamp": time.time(),
|
|
"models": models,
|
|
"modes": modes,
|
|
"subset": subset,
|
|
"n_instances": len(instances),
|
|
"runs": {
|
|
run_id: {
|
|
"total": len(results),
|
|
"cost": sum(r.get("cost", 0) for r in results),
|
|
"api_calls": sum(r.get("n_calls", 0) for r in results),
|
|
}
|
|
for run_id, results in all_results.items()
|
|
},
|
|
}
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
(output_dir / "matrix_summary.json").write_text(json.dumps(master, indent=2, default=str))
|
|
console.print(f"\n[green]Results saved to {output_dir}[/green]")
|
|
|
|
|
|
@app.command()
|
|
def debug(
|
|
model: str = typer.Option("claude-haiku", "-m", "--model", help="Model config name"),
|
|
mode: str = typer.Option("native_augment", "--mode", help="Evaluation mode"),
|
|
instance_id: str = typer.Option(..., "-i", "--instance", help="SWE-bench instance ID"),
|
|
subset: str = typer.Option("lite", "--subset", help="SWE-bench subset"),
|
|
split: str = typer.Option("dev", "--split"),
|
|
output: str = typer.Option(str(DEFAULT_OUTPUT_DIR / "debug"), "-o", "--output"),
|
|
):
|
|
"""Debug a single SWE-bench instance."""
|
|
from datasets import load_dataset
|
|
|
|
dataset_path = DATASET_MAPPING.get(subset, subset)
|
|
instances = {inst["instance_id"]: inst for inst in load_dataset(dataset_path, split=split)}
|
|
|
|
if instance_id not in instances:
|
|
console.print(f"[red]Instance '{instance_id}' not found in {subset}/{split}[/red]")
|
|
raise typer.Exit(1)
|
|
|
|
instance = instances[instance_id]
|
|
config = build_config(model, mode)
|
|
output_dir = Path(output)
|
|
|
|
console.print(f"\n[bold]Debug run:[/bold] {model} + {mode}")
|
|
console.print(f" Instance: {instance_id}")
|
|
console.print(f" Problem: {instance['problem_statement'][:200]}...\n")
|
|
|
|
result = process_instance(instance, config, output_dir, model, mode)
|
|
_print_summary([result], model, mode)
|
|
|
|
|
|
@app.command()
|
|
def list_configs():
|
|
"""List available model and mode configurations."""
|
|
console.print("\n[bold]Available Models:[/bold]")
|
|
for name in AVAILABLE_MODELS:
|
|
config = load_yaml_config(MODELS_DIR / f"{name}.yaml")
|
|
model_name = config.get("model", {}).get("model_name", "unknown")
|
|
console.print(f" {name:<20} {model_name}")
|
|
|
|
console.print("\n[bold]Available Modes:[/bold]")
|
|
for name in AVAILABLE_MODES:
|
|
config = load_yaml_config(MODES_DIR / f"{name}.yaml")
|
|
gn_mode = config.get("agent", {}).get("gitnexus_mode", "baseline")
|
|
console.print(f" {name:<20} gitnexus_mode={gn_mode}")
|
|
|
|
console.print(f"\n[bold]Matrix:[/bold] {len(AVAILABLE_MODELS)} models x {len(AVAILABLE_MODES)} modes = {len(AVAILABLE_MODELS) * len(AVAILABLE_MODES)} configurations")
|
|
|
|
|
|
# ─── Summary Output ────────────────────────────────────────────────────────
|
|
|
|
|
|
def _print_summary(results: list[dict], model: str, mode: str):
|
|
"""Print a summary table for a single run."""
|
|
if not results:
|
|
console.print("[yellow]No results to display[/yellow]")
|
|
return
|
|
|
|
table = Table(title=f"{model} + {mode}")
|
|
table.add_column("Metric", style="bold")
|
|
table.add_column("Value")
|
|
|
|
total = len(results)
|
|
completed = sum(1 for r in results if r.get("submission"))
|
|
total_cost = sum(r.get("cost", 0) for r in results)
|
|
total_calls = sum(r.get("n_calls", 0) for r in results)
|
|
|
|
table.add_row("Instances", str(total))
|
|
table.add_row("Completed", f"{completed}/{total}")
|
|
table.add_row("Total Cost", f"${total_cost:.4f}")
|
|
table.add_row("Total API Calls", str(total_calls))
|
|
table.add_row("Avg Cost/Instance", f"${total_cost / max(total, 1):.4f}")
|
|
table.add_row("Avg Calls/Instance", f"{total_calls / max(total, 1):.1f}")
|
|
|
|
# GitNexus-specific metrics
|
|
gn_tool_calls = sum(
|
|
r.get("gitnexus_metrics", {}).get("total_tool_calls", 0) for r in results
|
|
)
|
|
gn_augment_hits = sum(
|
|
r.get("gitnexus_metrics", {}).get("augmentation_hits", 0) for r in results
|
|
)
|
|
if gn_tool_calls > 0:
|
|
table.add_row("GitNexus Tool Calls", str(gn_tool_calls))
|
|
if gn_augment_hits > 0:
|
|
table.add_row("Augmentation Hits", str(gn_augment_hits))
|
|
|
|
console.print(table)
|
|
|
|
|
|
def _print_matrix_summary(all_results: dict[str, list[dict]]):
|
|
"""Print a comparative matrix summary."""
|
|
table = Table(title="Evaluation Matrix Summary")
|
|
table.add_column("Configuration", style="bold")
|
|
table.add_column("Instances")
|
|
table.add_column("Completed")
|
|
table.add_column("Cost")
|
|
table.add_column("API Calls")
|
|
table.add_column("GN Tools")
|
|
|
|
for run_id, results in sorted(all_results.items()):
|
|
total = len(results)
|
|
completed = sum(1 for r in results if r.get("submission"))
|
|
cost = sum(r.get("cost", 0) for r in results)
|
|
calls = sum(r.get("n_calls", 0) for r in results)
|
|
gn_calls = sum(r.get("gitnexus_metrics", {}).get("total_tool_calls", 0) for r in results)
|
|
|
|
table.add_row(
|
|
run_id,
|
|
str(total),
|
|
f"{completed}/{total}",
|
|
f"${cost:.2f}",
|
|
str(calls),
|
|
str(gn_calls) if gn_calls > 0 else "-",
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
|
|
app()
|