from pathlib import Path from time import perf_counter from dotenv import load_dotenv import typer from rich.console import Console from rich.progress import ( BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn, TimeElapsedColumn, TimeRemainingColumn, ) from rich.table import Table from data_agent_baseline.benchmark.dataset import DABenchPublicDataset from data_agent_baseline.config import load_app_config from data_agent_baseline.run.runner import TaskRunArtifacts, create_run_output_dir, run_benchmark, run_single_task from data_agent_baseline.tools.filesystem import list_context_tree from data_agent_baseline import logger from data_agent_baseline.visualization import generate_executive_report from data_agent_baseline.application.run_execution_service import RunExecutionService from data_agent_baseline.domain.run_models import RunSpec from data_agent_baseline.repositories.filesystem_run_repository import FilesystemRunRepository PROJECT_ROOT = Path(__file__).resolve().parents[2] CONFIGS_DIR = PROJECT_ROOT / "configs" DATA_DIR = Path("kdd-benchmark") # PROJECT_ROOT / "data" ARTIFACTS_DIR = Path("kdd-benchmark/artifacts") # PROJECT_ROOT / "artifacts" ARTIFACT_RUNS_DIR = ARTIFACTS_DIR / "runs" app = typer.Typer(add_completion=False, no_args_is_help=False) console = Console() def _status_value(path: Path) -> str: return "present" if path.exists() else "missing" def _format_compact_rate(completed_count: int, elapsed_seconds: float) -> str: if completed_count <= 0 or elapsed_seconds <= 0: return "rate=0.0 task/min" return f"rate={(completed_count / elapsed_seconds) * 60:.1f} task/min" def _format_last_task(artifact: TaskRunArtifacts | None) -> str: if artifact is None: return "last=-" status = "ok" if artifact.succeeded else "fail" return f"last={artifact.task_id} ({status})" def _build_compact_progress_fields( *, completed_count: int, succeeded_count: int, failed_count: int, task_total: int, max_workers: int, elapsed_seconds: float, last_artifact: TaskRunArtifacts | None, ) -> dict[str, str]: remaining_count = max(task_total - completed_count, 0) running_count = min(max_workers, remaining_count) queued_count = max(remaining_count - running_count, 0) return { "ok": str(succeeded_count), "fail": str(failed_count), "run": str(running_count), "queue": str(queued_count), "speed": _format_compact_rate(completed_count, elapsed_seconds), "last": _format_last_task(last_artifact), } @app.callback() def cli() -> None: """Utilities for working with the local DABench baseline project.""" @app.command() def status( config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), ) -> None: """Show the local project layout and public dataset presence.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) config_path = config.resolve() public_dataset = DABenchPublicDataset(app_config.dataset.root_path) table = Table(title="DABench Baseline Status") table.add_column("Item") table.add_column("Path") table.add_column("State") table.add_row("project_root", str(PROJECT_ROOT), "ready") table.add_row("data_dir", str(DATA_DIR), _status_value(DATA_DIR)) table.add_row("configs_dir", str(CONFIGS_DIR), _status_value(CONFIGS_DIR)) table.add_row("artifacts_dir", str(ARTIFACTS_DIR), _status_value(ARTIFACTS_DIR)) table.add_row("runs_dir", str(ARTIFACT_RUNS_DIR), _status_value(ARTIFACT_RUNS_DIR)) table.add_row("dataset_root", str(app_config.dataset.root_path), _status_value(app_config.dataset.root_path)) table.add_row("config_path", str(config_path), _status_value(config_path)) console.print(table) if public_dataset.exists: console.print(f"Public tasks: {len(public_dataset.list_task_ids())}") counts = public_dataset.task_counts() if counts: rendered_counts = ", ".join( f"{difficulty}={count}" for difficulty, count in sorted(counts.items()) ) console.print(f"Public task counts: {rendered_counts}") @app.command("inspect-task") def inspect_task( task_id: str, config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), ) -> None: """Show task metadata and available context files.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) dataset = DABenchPublicDataset(app_config.dataset.root_path) task = dataset.get_task(task_id) console.print(f"Task: {task.task_id}") console.print(f"Difficulty: {task.difficulty}") console.print(f"Question: {task.question}") context_listing = list_context_tree(task) table = Table(title=f"Context Files for {task.task_id}") table.add_column("Path") table.add_column("Kind") table.add_column("Size") for entry in context_listing["entries"]: table.add_row(str(entry["path"]), str(entry["kind"]), str(entry["size"] or "")) console.print(table) @app.command("search-tasks") def search_tasks( pattern: str = typer.Argument(..., help="File pattern to search for (e.g., '*.db', 'db/', '*.sqlite', '*.json')"), config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), difficulty: str = typer.Option(None, help="Filter by difficulty (easy, medium, hard, extreme)"), show_files: bool = typer.Option(False, "--show-files", help="Show matching files for each task"), ) -> None: """Search for tasks containing files matching a pattern. Examples: uv run dabench search-tasks "*.db" --config configs/react_baseline.azure.yaml uv run dabench search-tasks "db/" --config configs/react_baseline.azure.yaml --show-files uv run dabench search-tasks "*.sqlite" --config configs/react_baseline.azure.yaml --difficulty medium """ import fnmatch app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) dataset = DABenchPublicDataset(app_config.dataset.root_path) # Get tasks filtered by difficulty if specified if difficulty: tasks = dataset.iter_tasks(difficulty=difficulty) else: tasks = dataset.iter_tasks() matching_tasks = [] for task in tasks: context_listing = list_context_tree(task) matching_files = [] for entry in context_listing["entries"]: path_str = str(entry["path"]) # Check if pattern matches if pattern.endswith("/"): # Directory pattern (e.g., "db/") if path_str.startswith(pattern) or f"/{pattern}" in path_str: matching_files.append(path_str) elif "*" in pattern: # Wildcard pattern (e.g., "*.db", "*.sqlite") if fnmatch.fnmatch(path_str, pattern) or fnmatch.fnmatch(path_str.split("/")[-1], pattern): matching_files.append(path_str) else: # Exact match or substring if pattern in path_str: matching_files.append(path_str) if matching_files: matching_tasks.append((task, matching_files)) # Display results if not matching_tasks: console.print(f"[yellow]No tasks found matching pattern: {pattern}[/yellow]") return console.print(f"\n[green]Found {len(matching_tasks)} task(s) matching pattern: {pattern}[/green]\n") table = Table(title=f"Tasks with files matching '{pattern}'") table.add_column("Task ID", style="cyan") table.add_column("Difficulty", style="magenta") table.add_column("Match Count", justify="right", style="green") if show_files: table.add_column("Matching Files", style="yellow") for task, matching_files in matching_tasks: if show_files: files_str = "\n".join(matching_files[:10]) # Show first 10 files if len(matching_files) > 10: files_str += f"\n... and {len(matching_files) - 10} more" table.add_row(task.task_id, task.difficulty, str(len(matching_files)), files_str) else: table.add_row(task.task_id, task.difficulty, str(len(matching_files))) console.print(table) # Summary by difficulty difficulty_counts = {} for task, _ in matching_tasks: difficulty_counts[task.difficulty] = difficulty_counts.get(task.difficulty, 0) + 1 console.print("\n[bold]Summary by difficulty:[/bold]") for diff, count in sorted(difficulty_counts.items()): console.print(f" {diff}: {count} task(s)") @app.command("run-task") def run_task_command( task_id: str, config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), ) -> None: """Run the ReAct baseline on one task.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) try: _, run_output_dir = create_run_output_dir(app_config.run.output_dir, run_id=app_config.run.run_id) except (ValueError, FileExistsError) as exc: raise typer.BadParameter(str(exc), param_hint="run.run_id") from exc artifacts = run_single_task(task_id=task_id, config=app_config, run_output_dir=run_output_dir) console.print(f"Run output: {run_output_dir}") console.print(f"Task output: {artifacts.task_output_dir}") if artifacts.prediction_csv_path is not None: console.print(f"Prediction CSV: {artifacts.prediction_csv_path}") else: console.print("Prediction CSV: not generated") if artifacts.failure_reason is not None: console.print(f"Failure: {artifacts.failure_reason}") @app.command("run-benchmark") def run_benchmark_command( config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), limit: int | None = typer.Option(None, min=1, help="Maximum number of tasks to run."), ) -> None: """Run the ReAct baseline on multiple tasks from the config selection.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) dataset = DABenchPublicDataset(app_config.dataset.root_path) task_total = len(dataset.iter_tasks()) if limit is not None: task_total = min(task_total, limit) effective_workers = app_config.run.max_workers progress_columns = [ SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(), TextColumn("[dim]|[/dim]"), TextColumn("[green]ok={task.fields[ok]}[/green]"), TextColumn("[red]fail={task.fields[fail]}[/red]"), TextColumn("[cyan]run={task.fields[run]}[/cyan]"), TextColumn("[yellow]queue={task.fields[queue]}[/yellow]"), TextColumn("[dim]|[/dim]"), TextColumn("{task.fields[speed]}"), TextColumn("[dim]| elapsed[/dim]"), TimeElapsedColumn(), TextColumn("[dim]| eta[/dim]"), TimeRemainingColumn(), TextColumn("[dim]|[/dim]"), TextColumn("{task.fields[last]}"), ] with Progress(*progress_columns, console=console) as progress: progress_task_id = progress.add_task( "Benchmark", total=task_total, completed=0, **_build_compact_progress_fields( completed_count=0, succeeded_count=0, failed_count=0, task_total=task_total, max_workers=effective_workers, elapsed_seconds=0.0, last_artifact=None, ), ) completion_count = 0 succeeded_count = 0 failed_count = 0 start_time = perf_counter() def on_task_complete(artifact) -> None: nonlocal completion_count, succeeded_count, failed_count completion_count += 1 if artifact.succeeded: succeeded_count += 1 else: failed_count += 1 progress.update( progress_task_id, completed=completion_count, description="Benchmark", refresh=True, **_build_compact_progress_fields( completed_count=completion_count, succeeded_count=succeeded_count, failed_count=failed_count, task_total=task_total, max_workers=effective_workers, elapsed_seconds=perf_counter() - start_time, last_artifact=artifact, ), ) try: run_output_dir, artifacts = run_benchmark( config=app_config, limit=limit, progress_callback=on_task_complete, ) except (ValueError, FileExistsError) as exc: raise typer.BadParameter(str(exc), param_hint="run.run_id") from exc progress.update( progress_task_id, completed=task_total, description="Benchmark", refresh=True, **_build_compact_progress_fields( completed_count=task_total, succeeded_count=succeeded_count, failed_count=failed_count, task_total=task_total, max_workers=effective_workers, elapsed_seconds=perf_counter() - start_time, last_artifact=artifacts[-1] if artifacts else None, ), ) console.print(f"Run output: {run_output_dir}") console.print(f"Tasks attempted: {len(artifacts)}") console.print(f"Succeeded tasks: {sum(1 for item in artifacts if item.succeeded)}") @app.command("run-lang-task") def run_lang_task_command( task_ids: list[str] = typer.Argument(..., help="One or more task IDs to run (e.g. task_418 task_330)."), config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), display_mode: str = typer.Option("technical", help="Display mode: 'technical' (default) or 'executive'."), ) -> None: """Run the LangGraph multi-agent workflow on one or more tasks.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) spec = RunSpec( run_id=app_config.run.run_id, task_ids=list(task_ids), execution_mode="autonomous", evaluation_mode="standard", config_path=config.resolve(), max_workers=app_config.run.max_workers, ) repo = FilesystemRunRepository() service = RunExecutionService(app_config, repo) # Per-task Rich rendering callback, preserving existing display behavior. task_idx_tracker = [0] n_tasks = len(spec.task_ids) def _on_task_done(artifact: TaskRunArtifacts) -> None: task_idx_tracker[0] += 1 idx = task_idx_tracker[0] console.print(f"\n[bold cyan]{'='*60}[/bold cyan]") console.print(f"[bold]Completed {artifact.task_id} ({idx}/{n_tasks})[/bold]") console.print(f"[bold cyan]{'='*60}[/bold cyan]") if display_mode == "executive": trace_path = artifact.task_output_dir / "trace.json" if trace_path.exists(): generate_executive_report(trace_path, artifact.task_output_dir, console) else: console.print("[yellow]Trace file not yet available for executive report[/yellow]") _status = "[green]succeeded[/green]" if artifact.succeeded else "[red]failed[/red]" console.print(f" Status: {_status}") else: _status = "[green]succeeded[/green]" if artifact.succeeded else "[red]failed[/red]" console.print(f" Status: {_status}") if artifact.prediction_csv_path is not None: console.print(f" Prediction CSV: {artifact.prediction_csv_path}") if artifact.failure_reason is not None: console.print(f" Failure: {artifact.failure_reason}") try: result = service.execute_selected_tasks(spec, progress_callback=_on_task_done) except (ValueError, FileExistsError) as exc: raise typer.BadParameter(str(exc), param_hint="run.run_id") from exc console.print(f"\n[bold]{'='*60}[/bold]") console.print(f"[bold]Run output:[/bold] {result.run_output_dir}") console.print(f"[bold]Results:[/bold] {result.succeeded_count}/{len(result.task_results)} succeeded") for task_result in result.task_results: icon = "✅" if task_result.succeeded else "❌" console.print(f" {icon} {task_result.task_id}") @app.command("run-lang-benchmark") def run_lang_benchmark_command( config: Path = typer.Option(..., exists=True, dir_okay=False, help="YAML config path."), limit: int | None = typer.Option(None, min=1, help="Maximum number of tasks to run."), ) -> None: """Run the LangGraph multi-agent workflow across the public dataset.""" app_config = load_app_config(config) logger.initialize_logger(log_debug=app_config.logging.log_debug) dataset = DABenchPublicDataset(app_config.dataset.root_path) task_total = len(dataset.iter_tasks()) if limit is not None: task_total = min(task_total, limit) effective_workers = app_config.run.max_workers progress_columns = [ SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(), TextColumn("[dim]|[/dim]"), TextColumn("[green]ok={task.fields[ok]}[/green]"), TextColumn("[red]fail={task.fields[fail]}[/red]"), TextColumn("[cyan]run={task.fields[run]}[/cyan]"), TextColumn("[yellow]queue={task.fields[queue]}[/yellow]"), TextColumn("[dim]|[/dim]"), TextColumn("{task.fields[speed]}"), TextColumn("[dim]| elapsed[/dim]"), TimeElapsedColumn(), TextColumn("[dim]| eta[/dim]"), TimeRemainingColumn(), TextColumn("[dim]|[/dim]"), TextColumn("{task.fields[last]}"), ] completion_count = 0 succeeded_count = 0 failed_count = 0 start_time = perf_counter() task_finish_times: list[tuple[str, float, bool]] = [] last_completed_artifact: TaskRunArtifacts | None = None with Progress(*progress_columns, console=console) as progress: progress_task_id = progress.add_task( "LangGraph Benchmark", total=task_total, completed=0, **_build_compact_progress_fields( completed_count=0, succeeded_count=0, failed_count=0, task_total=task_total, max_workers=effective_workers, elapsed_seconds=0.0, last_artifact=None, ), ) def on_task_complete(artifact: TaskRunArtifacts) -> None: nonlocal completion_count, succeeded_count, failed_count, last_completed_artifact completion_count += 1 last_completed_artifact = artifact if artifact.succeeded: succeeded_count += 1 else: failed_count += 1 task_finish_times.append((artifact.task_id, perf_counter() - start_time, artifact.succeeded)) progress.update( progress_task_id, completed=completion_count, description="LangGraph Benchmark", refresh=True, **_build_compact_progress_fields( completed_count=completion_count, succeeded_count=succeeded_count, failed_count=failed_count, task_total=task_total, max_workers=effective_workers, elapsed_seconds=perf_counter() - start_time, last_artifact=artifact, ), ) spec = RunSpec( run_id=app_config.run.run_id, task_ids=[], execution_mode="autonomous", evaluation_mode="standard", config_path=config.resolve(), max_workers=app_config.run.max_workers, ) repo = FilesystemRunRepository() service = RunExecutionService(app_config, repo) try: result = service.execute_benchmark( spec, limit=limit, progress_callback=on_task_complete, ) except (ValueError, FileExistsError) as exc: raise typer.BadParameter(str(exc), param_hint="run.run_id") from exc progress.update( progress_task_id, completed=task_total, description="LangGraph Benchmark", refresh=True, **_build_compact_progress_fields( completed_count=task_total, succeeded_count=succeeded_count, failed_count=failed_count, task_total=task_total, max_workers=effective_workers, elapsed_seconds=perf_counter() - start_time, last_artifact=last_completed_artifact, ), ) console.print(f"Run output: {result.run_output_dir}") console.print(f"Tasks attempted: {len(result.task_results)}") console.print(f"Succeeded tasks: {result.succeeded_count}") # Timing summary total_elapsed = perf_counter() - start_time console.print(f"\n[bold]Timing:[/bold] total {total_elapsed:.1f}s ({total_elapsed/60:.1f}min)") if task_finish_times: sorted_times = sorted(task_finish_times, key=lambda t: t[1]) console.print("[bold]Per-task completion order:[/bold]") prev = 0.0 for tid, wall_s, ok in sorted_times: delta = wall_s - prev icon = "✅" if ok else "❌" console.print(f" {icon} {tid}: finished at {wall_s:.1f}s (delta {delta:.1f}s)") prev = wall_s @app.command("eval-lang") def eval_lang_command( run_id: str = typer.Argument(..., help="Run ID or full directory path to the prediction run."), gold_root: Path = typer.Option( Path("kdd-benchmark/output"), help="Root directory containing gold.csv files per task.", ), ) -> None: """Evaluate LangGraph predictions against ground truth (gold). Examples: uv run dabench eval-lang 20260507T063629Z uv run dabench eval-lang /data3/dataFAIR/kdd-dev/public/artifacts/runs/20260507T063629Z """ from data_agent_baseline.langgraph_agent.evaluator import evaluate_run # Resolve run directory run_path = Path(run_id) if not run_path.is_absolute() or not run_path.exists(): # Treat as run_id, append to artifacts path run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]Run directory not found: {run_path}[/red]") raise typer.Exit(1) console.print(f"[bold]Evaluating run:[/bold] {run_path}") console.print(f"[bold]Gold root:[/bold] {gold_root}") results_df, summary = evaluate_run(run_path, gold_root, lambda_values=[0.1, 0.2, 0.5, 1.0]) if results_df.empty: console.print("[yellow]No task directories found in run.[/yellow]") raise typer.Exit(1) # Display results table table = Table(title="Evaluation Results") table.add_column("Task ID", style="cyan") table.add_column("Score (λ=0.1)", justify="right", style="green") table.add_column("Score (λ=0.2)", justify="right", style="green") table.add_column("Score (λ=0.5)", justify="right", style="green") table.add_column("Score (λ=1.0)", justify="right", style="green") table.add_column("Recall", justify="right") table.add_column("Matched/Gold", justify="right") table.add_column("Extra", justify="right", style="yellow") table.add_column("Notes", style="dim") table.add_column("Time (s)", justify="right", style="dim") for _, row in results_df.iterrows(): notes = row.get("notes_l0.1", "") elapsed = row.get("elapsed_seconds") elapsed_str = f"{elapsed:.1f}" if elapsed is not None and elapsed == elapsed else "" table.add_row( row["task_id"], f"{row['score_l0.1']:.4f}", f"{row['score_l0.2']:.4f}", f"{row['score_l0.5']:.4f}", f"{row['score_l1.0']:.4f}", f"{row['recall_l0.1']:.4f}", f"{row['matched_l0.1']}/{row['gold_cols_l0.1']}", str(row["extra_l0.1"]), notes, elapsed_str, ) console.print(table) # Summary console.print(f"\n[bold]Summary ({summary['total_tasks']} tasks):[/bold]") console.print(f" Mean score (λ=0.1): {summary.get('mean_score_l0.1', 0):.4f}") console.print(f" Mean score (λ=0.2): {summary.get('mean_score_l0.2', 0):.4f}") console.print(f" Mean score (λ=0.5): {summary.get('mean_score_l0.5', 0):.4f}") console.print(f" Mean score (λ=1.0): {summary.get('mean_score_l1.0', 0):.4f}") console.print(f" Median score (λ=0.1): {summary.get('median_score_l0.1', 0):.4f}") console.print(f" Tasks with score > 0 (λ=0.1): {summary.get('tasks_with_score_gt0_l0.1', 0)}") console.print(f" Tasks with recall > 0 (λ=0.1): {summary.get('tasks_with_recall_gt0_l0.1', 0)}") console.print(f" Tasks with recall = 0 (λ=0.1): {summary.get('tasks_with_recall_eq0_l0.1', 0)}") # Per-difficulty breakdown difficulty_breakdown = summary.get("difficulty_breakdown", []) if difficulty_breakdown: difficulty_order = {"easy": 0, "medium": 1, "hard": 2, "extreme": 3} sorted_breakdown = sorted( difficulty_breakdown, key=lambda e: difficulty_order.get(e["difficulty"], 99), ) diff_table = Table(title="Breakdown by Difficulty") diff_table.add_column("Difficulty", style="magenta") diff_table.add_column("Tasks", justify="right") diff_table.add_column("Mean Score (λ=0.1)", justify="right", style="green") diff_table.add_column("Recall > 0 (λ=0.1)", justify="right", style="cyan") diff_table.add_column("Time min/avg/max (s)", justify="right", style="dim") for entry in sorted_breakdown: count = entry["count"] recall_gt0 = entry.get("tasks_with_recall_gt0_l0.1", 0) recall_pct = (recall_gt0 / count * 100) if count > 0 else 0.0 # Compute time stats for this difficulty diff_times = results_df[results_df["difficulty"] == entry["difficulty"]]["elapsed_seconds"].dropna() if not diff_times.empty: time_str = f"{diff_times.min():.0f}/{diff_times.mean():.0f}/{diff_times.max():.0f}" else: time_str = "-" diff_table.add_row( entry["difficulty"], str(count), f"{entry.get('mean_score_l0.1', 0):.4f}", f"{recall_gt0} ({recall_pct:.1f}%)", time_str, ) # Overall row total_tasks = summary["total_tasks"] overall_recall_gt0 = summary.get("tasks_with_recall_gt0_l0.1", 0) overall_recall_pct = (overall_recall_gt0 / total_tasks * 100) if total_tasks > 0 else 0.0 all_times = results_df["elapsed_seconds"].dropna() overall_time_str = ( f"{all_times.min():.0f}/{all_times.mean():.0f}/{all_times.max():.0f}" if not all_times.empty else "-" ) diff_table.add_row( "[bold]overall[/bold]", f"[bold]{total_tasks}[/bold]", f"[bold]{summary.get('mean_score_l0.1', 0):.4f}[/bold]", f"[bold]{overall_recall_gt0} ({overall_recall_pct:.1f}%)[/bold]", f"[bold]{overall_time_str}[/bold]", ) console.print(diff_table) # Save evaluation CSV eval_csv_path = run_path / "evaluation.csv" results_df.to_csv(eval_csv_path, index=False) console.print(f"\n[green]Evaluation saved to: {eval_csv_path}[/green]") # Show tasks with zero recall for debugging zero_recall = results_df[results_df["recall_l0.1"] == 0] if not zero_recall.empty: console.print(f"\n[bold red]Tasks with recall = 0 ({len(zero_recall)}):[/bold red]") for _, row in zero_recall.iterrows(): notes = row.get("notes_l0.1", "") console.print( f" ❌ {row['task_id']}: {notes}" if notes else f" ❌ {row['task_id']}" ) @app.command("view-exec-report") def view_exec_report_command( task_id: str = typer.Argument(..., help="Task ID to view report for (e.g., task_355)."), run_id: str = typer.Argument(..., help="Run ID or full directory path containing the task."), ) -> None: """View executive summary report for a completed task. This generates a stakeholder-friendly report showing decision reasoning, execution timeline, and outcomes in business language. Examples: uv run dabench view-exec-report task_355 20260601T075638Z uv run dabench view-exec-report task_355 /data3/dataFAIR/kdd-dev/public/artifacts/runs/20260601T075638Z """ # Resolve run directory run_path = Path(run_id) if not run_path.is_absolute() or not run_path.exists(): run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]❌ Run directory not found: {run_path}[/red]") raise typer.Exit(1) # Find task directory task_dir = run_path / task_id if not task_dir.exists(): console.print(f"[red]❌ Task directory not found: {task_dir}[/red]") raise typer.Exit(1) # Find trace file trace_path = task_dir / "trace.json" if not trace_path.exists(): console.print(f"[red]❌ Trace file not found: {trace_path}[/red]") console.print("[yellow]Hint: This command works with completed tasks that have trace.json files.[/yellow]") raise typer.Exit(1) # Generate report generate_executive_report(trace_path, task_dir, console) @app.command("eval-comprehensive") def eval_comprehensive_command( run_id: str = typer.Argument(..., help="Run ID or full directory path to the prediction run."), gold_root: Path = typer.Option( Path("kdd-benchmark/output"), help="Root directory containing gold.csv files per task.", ), lambda_penalty: float = typer.Option(0.1, help="Lambda penalty for extra columns in scoring."), output_csv: Path = typer.Option( None, help="Optional output CSV path. Defaults to /comprehensive_evaluation.csv", ), ) -> None: """Run comprehensive evaluation capturing all metrics for analysis. Captures: task ID, difficulty, execution success, scores, timing, tokens, trajectory length, tool calls/failures, recovery attempts, confidence, ground truth availability, failure buckets, and more. Examples: uv run dabench eval-comprehensive 20260507T063629Z uv run dabench eval-comprehensive /data3/dataFAIR/kdd-dev/public/artifacts/runs/20260507T063629Z """ from data_agent_baseline.langgraph_agent.comprehensive_evaluator import evaluate_run_comprehensive # Resolve run directory run_path = Path(run_id) if not run_path.is_absolute() or not run_path.exists(): run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]Run directory not found: {run_path}[/red]") raise typer.Exit(1) console.print(f"[bold]Comprehensive evaluation of run:[/bold] {run_path}") console.print(f"[bold]Gold root:[/bold] {gold_root}") console.print(f"[bold]Lambda penalty:[/bold] {lambda_penalty}") results_df, summary = evaluate_run_comprehensive(run_path, gold_root, lambda_penalty=lambda_penalty) if results_df.empty: console.print("[yellow]No task directories found in run.[/yellow]") raise typer.Exit(1) # Display comprehensive results table table = Table(title="Comprehensive Evaluation Results") table.add_column("Task ID", style="cyan", width=10) table.add_column("Difficulty", style="magenta", width=8) table.add_column("Exec Success", justify="center", width=7) table.add_column("Score", justify="right", style="green", width=7) table.add_column("Recall", justify="right", width=7) table.add_column("Time(s)", justify="right", width=7) table.add_column("Traj", justify="right", width=5) table.add_column("Tools", justify="right", width=6) table.add_column("LLMs", justify="right", width=6) table.add_column("Fails", justify="right", width=6) table.add_column("Recov", justify="right", width=6) table.add_column("Extra", justify="right", width=6) table.add_column("Confidence", style="dim", width=18) table.add_column("Bucket", style="yellow", width=12) table.add_column("Shape pred→gold", style="dim", width=16) table.add_column("Notes", style="dim") for _, row in results_df.iterrows(): success_icon = "✅" if row.get("execution_success", 0) == 1 else "❌" score = row.get("final_score", 0) recall = row.get("recall", 0) time_val = row.get("execution_time") traj = row.get("trajectory_length", 0) tools = row.get("tool_calls", 0) llms = row.get("llm_calls", 0) fails = row.get("tool_failures", 0) recov = row.get("recovery_attempts", 0) extra = row.get("extra_columns", 0) bucket = row.get("bucket", "other") # Format confidence as score-label conf_score = row.get("confidence_score") conf_label = row.get("confidence_label", "") if conf_score is not None and conf_score == conf_score: # Check for not NaN confidence = f"{conf_score:.2f}-{conf_label}" if conf_label else f"{conf_score:.2f}" elif conf_label: confidence = conf_label else: confidence = "" shape = str(row.get("shape_pred_gold", "") or "") note = str(row.get("notes", "") or "") note = (note[:50] + "…") if len(note) > 50 else note table.add_row( row["task_id"], row.get("difficulty", "Unknown"), success_icon, f"{score:.3f}", f"{recall:.2f}", f"{time_val:.1f}" if time_val is not None else "N/A", str(traj), str(tools), str(llms), str(fails), str(recov), str(extra), confidence, bucket, shape, note, ) console.print(table) # Display summary console.print(f"\n[bold]Summary ({summary['total_tasks']} tasks):[/bold]") console.print(f" Execution success: {summary.get('execution_success_count', 0)}/{summary['total_tasks']} ({summary.get('execution_success_rate', 0):.1%})") console.print(f" Mean score: {summary.get('mean_score', 0):.4f}") console.print(f" Median score: {summary.get('median_score', 0):.4f}") console.print(f" Mean recall: {summary.get('mean_recall', 0):.4f}") console.print(f" Tasks with score > 0: {summary.get('tasks_with_score_gt_0', 0)}") console.print(f" Mean execution time: {summary.get('mean_execution_time', 0):.1f}s") console.print(f" Total execution time: {summary.get('total_execution_time', 0):.1f}s ({summary.get('total_execution_time', 0)/60:.1f} min)") console.print(f" Mean trajectory length: {summary.get('mean_trajectory_length', 0):.1f}") console.print(f" Mean tool calls: {summary.get('mean_tool_calls', 0):.1f}") console.print(f" Mean LLM calls: {summary.get('mean_llm_calls', 0):.1f}") console.print(f" Total LLM calls: {summary.get('total_llm_calls', 0)}") console.print(f" Mean tool failures: {summary.get('mean_tool_failures', 0):.1f}") console.print(f" Mean recovery attempts: {summary.get('mean_recovery_attempts', 0):.1f}") # Bucket distribution bucket_dist = summary.get("bucket_distribution", {}) if bucket_dist: console.print("\n[bold]Failure bucket distribution:[/bold]") for bucket, count in sorted(bucket_dist.items(), key=lambda x: -x[1]): pct = (count / summary['total_tasks'] * 100) if summary['total_tasks'] > 0 else 0 console.print(f" {bucket}: {count} ({pct:.1f}%)") # Per-difficulty breakdown difficulty_breakdown = summary.get("difficulty_breakdown", []) if difficulty_breakdown: diff_table = Table(title="Breakdown by Difficulty") diff_table.add_column("Difficulty", style="magenta") diff_table.add_column("Tasks", justify="right") diff_table.add_column("Exec Success Rate", justify="right", style="cyan") diff_table.add_column("Mean Score", justify="right", style="green") diff_table.add_column("Mean Recall", justify="right") diff_table.add_column("Time min/avg/max (s)", justify="right", style="dim") difficulty_order = {"Easy": 0, "Medium": 1, "Hard": 2, "Extreme": 3} sorted_breakdown = sorted( difficulty_breakdown, key=lambda e: difficulty_order.get(e["difficulty"], 99), ) for entry in sorted_breakdown: t_min = entry.get("min_execution_time", 0) t_avg = entry.get("mean_execution_time", 0) t_max = entry.get("max_execution_time", 0) time_str = f"{t_min:.0f}/{t_avg:.0f}/{t_max:.0f}" diff_table.add_row( entry["difficulty"], str(entry["count"]), f"{entry.get('execution_success_rate', 0):.1%}", f"{entry.get('mean_score', 0):.4f}", f"{entry.get('mean_recall', 0):.4f}", time_str, ) console.print(diff_table) # Save comprehensive evaluation CSV if output_csv is None: output_csv = run_path / "comprehensive_evaluation.csv" # Flatten action_counts dict for CSV if "action_counts" in results_df.columns: results_df = results_df.drop(columns=["action_counts"]) results_df.to_csv(output_csv, index=False) console.print(f"\n[green]✅ Comprehensive evaluation saved to: {output_csv}[/green]") # Per-Phase timing table (matching tag-failures style, includes tokens) phase_table = summary.get("phase_table", {}) if phase_table: ph_table = Table(title="Per-Phase Timing (across all tasks)") ph_table.add_column("Phase", style="cyan") ph_table.add_column("Tasks", justify="right") ph_table.add_column("Total (s)", justify="right", style="green") ph_table.add_column("Mean (s)", justify="right") ph_table.add_column("Median (s)", justify="right") ph_table.add_column("P90 (s)", justify="right", style="yellow") ph_table.add_column("Max (s)", justify="right", style="red") ph_table.add_column("Total Tokens", justify="right", style="dim") ph_table.add_column("Mean Tokens", justify="right", style="dim") ph_table.add_column("Tool Calls", justify="right", style="dim") ph_table.add_column("LLM Calls", justify="right", style="dim") for ph, vals in sorted(phase_table.items(), key=lambda kv: -kv[1].get("total_s", 0)): ph_table.add_row( ph, str(int(vals.get("tasks", 0))), f"{vals.get('total_s', 0):.1f}", f"{vals.get('mean_s', 0):.2f}", f"{vals.get('median_s', 0):.2f}", f"{vals.get('p90_s', 0):.2f}", f"{vals.get('max_s', 0):.2f}", str(int(vals.get("total_tokens", 0))), f"{vals.get('mean_tokens', 0):.0f}", str(int(vals.get("total_calls", 0))), str(int(vals.get("total_llm_calls", 0))), ) console.print(ph_table) # Show example tasks for each bucket console.print("\n[bold]Example tasks by bucket:[/bold]") for bucket in ["crash", "timeout", "no_prediction", "wrong_column_count", "low_recall", "perfect"]: bucket_tasks = results_df[results_df["bucket"] == bucket] if not bucket_tasks.empty: examples = bucket_tasks["task_id"].head(3).tolist() console.print(f" {bucket}: {', '.join(examples)}") @app.command("tag-failures") def tag_failures_command( run_id: str = typer.Argument(..., help="Run ID or full directory path to the prediction run."), gold_root: Path = typer.Option( Path("kdd-benchmark/output"), help="Root directory containing gold.csv files per task.", ), eval_csv: Path = typer.Option( None, help="Optional evaluation.csv to join scores from. Defaults to /evaluation.csv if present.", ), ) -> None: """Classify task outcomes into failure-mode buckets and show per-phase timing. Examples: uv run dabench tag-failures 20260507T063629Z uv run dabench tag-failures /data3/dataFAIR/kdd-dev/public/artifacts/runs/20260507T063629Z """ from data_agent_baseline.langgraph_agent.failure_tagger import tag_run run_path = Path(run_id) if not run_path.is_absolute() or not run_path.exists(): run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]Run directory not found: {run_path}[/red]") raise typer.Exit(1) if eval_csv is None: default_eval = run_path / "evaluation.csv" eval_csv = default_eval if default_eval.exists() else None console.print(f"[bold]Tagging run:[/bold] {run_path}") console.print(f"[bold]Gold root:[/bold] {gold_root}") if eval_csv: console.print(f"[bold]Eval CSV:[/bold] {eval_csv}") df, summary = tag_run(run_path, gold_root, eval_csv=eval_csv) if df.empty: console.print("[yellow]No tasks found in run.[/yellow]") raise typer.Exit(1) # --- Per-task tags table (one row per task) --- _BUCKET_STYLE = { "perfect": "green", "near_miss": "yellow", "low_recall": "yellow", "value_mismatch": "red", "wrong_column_count": "red", "wrong_row_count": "red", "empty_prediction": "red", "no_prediction": "red", "timeout": "magenta", "api_error": "magenta", "crash": "bright_red", "no_gold": "dim", "other": "dim", } tag_table = Table(title="Per-Task Failure Tags") tag_table.add_column("Task ID", style="cyan") tag_table.add_column("Difficulty", style="magenta") tag_table.add_column("Bucket") tag_table.add_column("Score", justify="right") tag_table.add_column("Recall", justify="right") tag_table.add_column("Shape pred→gold", justify="right", style="dim") tag_table.add_column("Elapsed (s)", justify="right", style="dim") tag_table.add_column("Notes", style="dim") # Sort by bucket (worst first) then by task_id for easy scanning _BUCKET_ORDER = {b: i for i, b in enumerate([ "crash", "api_error", "timeout", "no_prediction", "empty_prediction", "wrong_column_count", "wrong_row_count", "value_mismatch", "low_recall", "other", "no_gold", "near_miss", "perfect", ])} sorted_df = df.assign(_ord=df["bucket"].map(lambda b: _BUCKET_ORDER.get(b, 99))).sort_values( ["_ord", "task_id"] ) for _, r in sorted_df.iterrows(): bucket = r["bucket"] style = _BUCKET_STYLE.get(bucket, "white") score = r["score"] recall = r["recall"] score_str = f"{score:.3f}" if score == score else "-" recall_str = f"{recall:.2f}" if recall == recall else "-" shape_str = f"{r['pred_rows']}x{r['pred_cols']} → {r['gold_rows']}x{r['gold_cols']}" elapsed = r["elapsed_seconds"] or 0 note = r["notes"] or r["failure_reason"] or "" note = (str(note)[:80] + "…") if len(str(note)) > 80 else str(note) tag_table.add_row( r["task_id"], r.get("difficulty", "") or "", f"[{style}]{bucket}[/{style}]", score_str, recall_str, shape_str, f"{float(elapsed):.0f}", note, ) console.print(tag_table) # Bucket summary table bucket_table = Table(title="Failure-Mode Buckets") bucket_table.add_column("Bucket", style="cyan") bucket_table.add_column("Count", justify="right", style="green") bucket_table.add_column("%", justify="right") bucket_table.add_column("Mean Score", justify="right") bucket_table.add_column("Mean Elapsed (s)", justify="right", style="dim") bucket_order = [ "perfect", "near_miss", "low_recall", "value_mismatch", "wrong_column_count", "wrong_row_count", "empty_prediction", "no_prediction", "timeout", "api_error", "crash", "no_gold", "other", ] seen_buckets = [b for b in bucket_order if b in summary and b != "__phases__"] seen_buckets += [b for b in summary if b not in bucket_order and b != "__phases__"] for b in seen_buckets: s = summary[b] mean_score = s.get("mean_score", float("nan")) score_str = f"{mean_score:.3f}" if mean_score == mean_score else "-" bucket_table.add_row( b, str(int(s["count"])), f"{s['pct']:.1f}", score_str, f"{s['mean_elapsed_s']:.1f}", ) console.print(bucket_table) # Per-phase timing summary phase_summary = summary.get("__phases__", {}) or {} if phase_summary: ph_table = Table(title="Per-Phase Timing (across all tasks)") ph_table.add_column("Phase", style="cyan") ph_table.add_column("Tasks", justify="right") ph_table.add_column("Total (s)", justify="right", style="green") ph_table.add_column("Mean (s)", justify="right") ph_table.add_column("Median (s)", justify="right") ph_table.add_column("P90 (s)", justify="right", style="yellow") ph_table.add_column("Max (s)", justify="right", style="red") # Sort by total time desc for ph, vals in sorted(phase_summary.items(), key=lambda kv: -kv[1].get("total_s", 0)): ph_table.add_row( ph, str(int(vals["tasks"])), f"{vals['total_s']:.1f}", f"{vals['mean_s']:.2f}", f"{vals['median_s']:.2f}", f"{vals['p90_s']:.2f}", f"{vals['max_s']:.2f}", ) console.print(ph_table) # Per-bucket task lists (for non-perfect buckets only) console.print("\n[bold]Tasks by bucket (non-perfect):[/bold]") for b in seen_buckets: if b == "perfect": continue sub = df[df["bucket"] == b] if sub.empty: continue console.print(f"\n [bold]{b}[/bold] ({len(sub)}):") for _, r in sub.iterrows(): note = r["notes"] or r["failure_reason"] note = (note[:120] + "…") if len(str(note)) > 120 else note console.print(f" • {r['task_id']} [dim]{note}[/dim]") # Save CSV out_csv = run_path / "failure_tags.csv" df.to_csv(out_csv, index=False) console.print(f"\n[green]Failure tags saved to: {out_csv}[/green]") @app.command("eval-v2") def eval_v2_command( run_id: str = typer.Argument(..., help="Run ID or full directory path to the prediction run."), gold_root: Path = typer.Option( Path("kdd-benchmark/output"), help="Root directory containing gold.csv files per task.", ), task_root: Path = typer.Option( None, help="Root directory containing task.json files (auto-detected if not provided).", ), lambda_penalty: float = typer.Option(0.1, help="Lambda penalty for extra columns in scoring."), mode: str = typer.Option( "standard", help="Display mode: 'standard', 'verbose', or 'research'.", ), ) -> None: """Evaluate run with V2 comprehensive metrics (KDD Creative Track). Produces three CSV files: - task_metrics.csv: Per-task comprehensive metrics - trajectory.csv: Per-step trajectory trace - tool_calls.csv: Per-tool-call analysis Also generates comprehensive_evaluation.csv for backward compatibility. """ from data_agent_baseline.application.evaluation_service import EvaluationService from data_agent_baseline.domain.evaluation_models import EvaluationOptions from data_agent_baseline.repositories.filesystem_evaluation_repository import ( FilesystemEvaluationRepository, ) from data_agent_baseline.langgraph_agent.eval_v2_viz import render_evaluation_report options = EvaluationOptions( gold_root=gold_root, lambda_penalty=lambda_penalty, task_root=task_root, mode=mode, ) run_candidate = Path(run_id) rendered_run_path = run_candidate if run_candidate.is_absolute() else ARTIFACT_RUNS_DIR / run_id console.print(f"[cyan]Evaluating run: {rendered_run_path}[/cyan]") console.print(f"[cyan]Mode: {mode}[/cyan]\n") repository = FilesystemEvaluationRepository() service = EvaluationService(artifact_runs_dir=ARTIFACT_RUNS_DIR, repository=repository) def _progress(stage: str) -> None: if stage == "evaluation_start": console.print(f"[cyan]{'='*80}[/cyan]") console.print("[cyan bold]EVALUATION HARDENING SUITE[/cyan bold]") console.print(f"[cyan]{'='*80}[/cyan]\n") elif stage == "artifact_reconciliation": console.print("[cyan]STEP 1: ARTIFACT RECONCILIATION[/cyan]") elif stage == "engineering_health_report": console.print("[cyan]STEP 2: ENGINEERING HEALTH REPORT[/cyan]") elif stage == "replay_artifact_generation": console.print("[cyan]STEP 3: GENERATE REPLAY ARTIFACTS[/cyan]") elif stage == "consistency_validation": console.print("[cyan]STEP 4: CONSISTENCY VALIDATION[/cyan]") elif stage == "report_mode_validation": console.print("[cyan]STEP 5: REPORT-MODE VALIDATION[/cyan]") elif stage == "evaluation_complete": console.print(f"\n[cyan]{'='*80}[/cyan]\n") try: bundle = service.evaluate_run(run_id, options, progress_callback=_progress) except FileNotFoundError as exc: console.print(f"[red]Error: {exc}[/red]") raise typer.Exit(1) if bundle.validation_issues: console.print( f"[yellow]Consistency validator found {len(bundle.validation_issues)} issue(s).[/yellow]" ) for issue in bundle.validation_issues[:20]: sev = str(issue.get("severity", "error")).upper() console.print( f" • [{sev}] {issue.get('check_id')}: {issue.get('task_id')} - {issue.get('detail')}" ) if len(bundle.validation_issues) > 20: console.print(f" • ... and {len(bundle.validation_issues) - 20} more") render_evaluation_report( bundle.task_metrics, bundle.summary, console, mode=options.mode, validation_issues=bundle.validation_issues, ) console.print("\n[green]Evaluation complete![/green]") console.print(f"[green]Harness health:[/green] {bundle.harness_health_status}") console.print(f"[green]Run quality:[/green] {bundle.run_quality_status}") console.print(f"[green]Replay artifacts:[/green] {bundle.replay_artifact_count}") console.print(f"[green]Failed tasks (attribution):[/green] {bundle.failed_task_count}") console.print("\n[green]Results saved to:[/green]") console.print(f" • {bundle.artifact_paths['task_metrics_csv']}") console.print(f" • {bundle.artifact_paths['trajectory_csv']}") console.print(f" • {bundle.artifact_paths['tool_calls_csv']}") console.print(f" • {bundle.artifact_paths['comprehensive_evaluation_csv']} (backward compatibility)") console.print(f" • {bundle.artifact_paths['validation_report_md']}") console.print(f" • {bundle.artifact_paths['auditor_report_md']}") console.print("\n[green]Hardening artifacts:[/green]") console.print(f" • {bundle.artifact_paths['reconciliation_report_txt']}") console.print(f" • {bundle.artifact_paths['health_report_txt']}") console.print(f" • {bundle.run_path / '*/task_replay.json'} ({bundle.replay_artifact_count} files)") if bundle.status == "invalid": console.print("[red]Evaluation produced inconsistent metrics; failing eval-v2.[/red]") raise typer.Exit(1) if bundle.status == "warning": warning_count = sum( 1 for issue in bundle.validation_issues if str(issue.get("severity", "")).lower() == "warning" ) console.print( f"[yellow]Evaluation completed with {warning_count} validator warning(s).[/yellow]" ) @app.command("view-task-v2") def view_task_v2_command( task_id: str = typer.Argument(..., help="Task ID to view (e.g., task_355)."), run_id: str = typer.Argument(..., help="Run ID or full directory path containing the task."), ) -> None: """View detailed V2 metrics for a specific task (verbose mode).""" from data_agent_baseline.langgraph_agent.eval_v2_viz import render_verbose_task_detail import pandas as pd # Resolve run directory run_path = Path(run_id) if not run_path.is_absolute(): run_path = ARTIFACT_RUNS_DIR / run_id task_dir = run_path / task_id if not task_dir.exists(): console.print(f"[red]Error: Task directory not found: {task_dir}[/red]") raise typer.Exit(1) # Load metrics from task_metrics.csv metrics_path = run_path / "task_metrics.csv" if not metrics_path.exists(): console.print(f"[red]Error: task_metrics.csv not found. Run 'eval-v2' first.[/red]") raise typer.Exit(1) metrics_df = pd.read_csv(metrics_path) task_metrics = metrics_df[metrics_df["task_id"] == task_id] if task_metrics.empty: console.print(f"[red]Error: Task {task_id} not found in metrics.[/red]") raise typer.Exit(1) metrics_dict = task_metrics.iloc[0].to_dict() # Load trajectory trajectory_path = run_path / "trajectory.csv" trajectory_list = [] if trajectory_path.exists(): trajectory_df = pd.read_csv(trajectory_path) task_trajectory = trajectory_df[trajectory_df["task_id"] == task_id] trajectory_list = task_trajectory.to_dict("records") # Load tool calls tool_calls_path = run_path / "tool_calls.csv" tool_calls_list = [] if tool_calls_path.exists(): tool_calls_df = pd.read_csv(tool_calls_path) task_tool_calls = tool_calls_df[tool_calls_df["task_id"] == task_id] tool_calls_list = task_tool_calls.to_dict("records") # Render detailed view render_verbose_task_detail(task_id, metrics_dict, trajectory_list, tool_calls_list, console) @app.command("eval-baseline") def eval_baseline_command( run_id: str = typer.Argument(..., help="Run ID (directory name under artifacts/runs/)"), task_root: Path = typer.Option( DATA_DIR / "input_full", exists=True, file_okay=False, dir_okay=True, help="Root directory containing task metadata (task.json files)", ), gold_root: Path = typer.Option( Path("kdd-benchmark/output"), exists=True, file_okay=False, dir_okay=True, help="Root directory containing gold answer files (output/task_*/gold.csv)", ), output_dir: Path | None = typer.Option( None, help="Optional output directory. Defaults to /baseline_evaluation/", ), ) -> None: """Run Phase 1 evaluation for baseline ReAct agent. Converts baseline traces to canonical schema and generates evaluation reports compatible with the existing evaluation harness. Example: dabench eval-baseline 20260613T114457Z """ from data_agent_baseline.evaluation.baseline_adapter import BaselineTraceAdapter from data_agent_baseline.evaluation.phase1_evaluator import Phase1Evaluator from data_agent_baseline.evaluation.report_generator import Phase1ReportGenerator console.print(f"[bold]Phase 1 Baseline Evaluation[/bold]") console.print(f"Run ID: {run_id}") console.print() # Resolve paths run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]Error: Run directory not found: {run_path}[/red]") raise typer.Exit(1) if output_dir is None: output_dir = run_path / "baseline_evaluation" output_dir.mkdir(parents=True, exist_ok=True) console.print(f"Run path: {run_path}") console.print(f"Output directory: {output_dir}") console.print() # Step 1: Normalize traces console.print("[cyan]Step 1: Normalizing baseline traces...[/cyan]") adapter = BaselineTraceAdapter(task_root=task_root) try: canonical_traces = adapter.normalize_run(run_path, run_id) console.print(f" ✓ Normalized {len(canonical_traces)} traces") except Exception as e: console.print(f"[red]Error normalizing traces: {e}[/red]") raise typer.Exit(1) if len(canonical_traces) == 0: console.print("[yellow]Warning: No traces found in run directory[/yellow]") raise typer.Exit(1) console.print() # Step 2: Save normalized traces console.print("[cyan]Step 2: Saving normalized traces...[/cyan]") from data_agent_baseline.evaluation.normalized_trace_manager import NormalizedTraceManager trace_manager = NormalizedTraceManager(output_dir=output_dir) try: saved_paths = trace_manager.save_normalized_traces(canonical_traces) console.print(f" ✓ Saved {len(saved_paths)} normalized traces") console.print(f" → {output_dir / 'normalized_traces'}") except Exception as e: console.print(f"[red]Error saving normalized traces: {e}[/red]") raise typer.Exit(1) console.print() # Step 3: Validate normalized traces console.print("[cyan]Step 3: Validating normalized traces...[/cyan]") try: validation_results = trace_manager.validate_all_traces() invalid_count = sum(1 for is_valid, _ in validation_results.values() if not is_valid) if invalid_count > 0: console.print(f" [yellow]⚠ {invalid_count} traces have validation errors[/yellow]") for task_id, (is_valid, errors) in validation_results.items(): if not is_valid: console.print(f" {task_id}: {', '.join(errors[:3])}") else: console.print(f" ✓ All {len(validation_results)} traces validated") except Exception as e: console.print(f"[yellow]Warning: Validation error: {e}[/yellow]") console.print() # Step 4: Evaluate tasks console.print("[cyan]Step 4: Computing Phase 1 metrics...[/cyan]") evaluator = Phase1Evaluator(gold_root=gold_root) try: results = evaluator.evaluate_run(canonical_traces) console.print(f" ✓ Evaluated {len(results)} tasks") except Exception as e: console.print(f"[red]Error evaluating tasks: {e}[/red]") raise typer.Exit(1) console.print() # Step 5: Generate reports console.print("[cyan]Step 5: Generating evaluation reports...[/cyan]") generator = Phase1ReportGenerator(output_dir=output_dir) try: outputs = generator.generate_all_reports(results, run_id) for name, path in outputs.items(): console.print(f" ✓ {name}: {path.relative_to(run_path)}") except Exception as e: console.print(f"[red]Error generating reports: {e}[/red]") raise typer.Exit(1) console.print() # Step 4: Display summary console.print("[bold green]✓ Evaluation Complete[/bold green]") console.print() # Load and display summary summary_path = output_dir / "summary_metrics.json" if summary_path.exists(): import json with summary_path.open("r") as f: summary = json.load(f) overall = summary.get("overall", {}) table = Table(title="Evaluation Summary") table.add_column("Metric", style="cyan") table.add_column("Value", style="green") table.add_row("Total Tasks", str(overall.get("total_tasks", 0))) table.add_row("Success Rate", f"{overall.get('success_rate', 0.0) * 100:.1f}%") table.add_row("Perfect Score Rate", f"{overall.get('perfect_rate', 0.0) * 100:.1f}%") table.add_row("Average Score", f"{overall.get('average_score', 0.0):.3f}") table.add_row("Average Steps", f"{overall.get('average_trajectory_length', 0.0):.1f}") table.add_row("Average Runtime", f"{overall.get('average_execution_time', 0.0):.1f}s") console.print(table) console.print() console.print(f"📊 View full report: {(output_dir / 'evaluation_report.md').relative_to(run_path)}") @app.command("view-normalized-trace") def view_normalized_trace_command( run_id: str = typer.Argument(..., help="Run ID (directory name under artifacts/runs/)"), task_id: str = typer.Argument(..., help="Task ID to view"), show_steps: bool = typer.Option( True, "--steps/--no-steps", help="Show detailed step breakdown", ), show_metrics: bool = typer.Option( True, "--metrics/--no-metrics", help="Show derived metrics", ), validate: bool = typer.Option( True, "--validate/--no-validate", help="Validate trace schema", ), ) -> None: """View a normalized trace with detailed breakdown. Example: dabench view-normalized-trace 20260613T114457Z task_22 """ from data_agent_baseline.evaluation.normalized_trace_manager import NormalizedTraceManager run_path = ARTIFACT_RUNS_DIR / run_id if not run_path.exists(): console.print(f"[red]Error: Run directory not found: {run_path}[/red]") raise typer.Exit(1) output_dir = run_path / "baseline_evaluation" if not output_dir.exists(): console.print(f"[red]Error: Evaluation not found. Run 'eval-baseline' first.[/red]") raise typer.Exit(1) manager = NormalizedTraceManager(output_dir=output_dir) # Load trace try: trace = manager.load_normalized_trace(task_id) except FileNotFoundError: console.print(f"[red]Error: Normalized trace not found for {task_id}[/red]") console.print(f"Available traces: {', '.join(manager.list_normalized_traces())}") raise typer.Exit(1) # Display header console.print(f"[bold]Normalized Trace: {task_id}[/bold]") console.print(f"Run ID: {trace['run_id']}") console.print(f"Agent Type: {trace['agent_type']}") console.print() # Task info table = Table(title="Task Information") table.add_column("Field", style="cyan") table.add_column("Value") table.add_row("Task ID", trace["task_id"]) table.add_row("Question", trace["question"]) table.add_row("Difficulty", trace.get("difficulty", "Unknown")) table.add_row("Success", "✓" if trace["success"] else "✗") table.add_row("Duration", f"{trace.get('duration_seconds', 0):.2f}s") if trace.get("failure_reason"): table.add_row("Failure Reason", trace["failure_reason"]) console.print(table) console.print() # Validation if validate: is_valid, errors = manager.validate_normalized_trace(trace) if is_valid: console.print("[green]✓ Trace validation passed[/green]") else: console.print("[red]✗ Trace validation failed:[/red]") for error in errors: console.print(f" - {error}") console.print() # Metrics if show_metrics: metrics = manager.get_trace_metrics(trace) metrics_table = Table(title="Derived Metrics") metrics_table.add_column("Metric", style="cyan") metrics_table.add_column("Value", style="green") metrics_table.add_row("Total Steps", str(metrics["num_steps"])) metrics_table.add_row("Tool Calls", str(metrics["num_tool_calls"])) metrics_table.add_row("Failed Tools", str(metrics["num_failed_tools"])) metrics_table.add_row("Unique Tools", str(metrics["unique_tools"])) console.print(metrics_table) console.print() # Agent breakdown if metrics["agent_steps"]: agent_table = Table(title="Agent Breakdown") agent_table.add_column("Agent", style="cyan") agent_table.add_column("Steps", style="green") for agent, count in metrics["agent_steps"].items(): agent_table.add_row(agent, str(count)) console.print(agent_table) console.print() # Tool breakdown if metrics["tool_counts"]: tool_table = Table(title="Tool Usage") tool_table.add_column("Tool", style="cyan") tool_table.add_column("Count", style="green") sorted_tools = sorted(metrics["tool_counts"].items(), key=lambda x: x[1], reverse=True) for tool, count in sorted_tools: tool_table.add_row(tool, str(count)) console.print(tool_table) console.print() # Steps if show_steps: steps_table = Table(title=f"Execution Steps ({len(trace['steps'])} total)") steps_table.add_column("Step", style="cyan", width=4) steps_table.add_column("Agent", style="magenta", width=15) steps_table.add_column("Role", style="blue", width=10) steps_table.add_column("Action", style="green", width=15) steps_table.add_column("Success", width=7) steps_table.add_column("Thought", width=50) for step in trace["steps"]: success_icon = "✓" if step["tool_success"] else "✗" thought_preview = step["thought"][:47] + "..." if len(step["thought"]) > 50 else step["thought"] steps_table.add_row( str(step["step_id"]), step["agent"], step["agent_role"], step["action"], success_icon, thought_preview, ) console.print(steps_table) console.print() # Final answer if trace.get("final_answer"): answer = trace["final_answer"] console.print("[bold]Final Answer:[/bold]") console.print(f"Columns: {', '.join(answer['columns'])}") console.print(f"Rows: {len(answer['rows'])}") console.print() def main() -> None: print("Starting DABench CLI...") load_dotenv() app() # added for debugging in VSCode, since it doesn't seem to recognize the app() call above as the entry point if __name__ == "__main__": main()