refactor(misc): drop intra-function blank lines (AST-identical)
This commit is contained in:
Teknium
@@ -208,7 +208,6 @@ def _prepare_container_image(
|
||||
except Exception as img_err:
|
||||
if config.get("verbose"):
|
||||
print(f" Prompt {prompt_index}: Docker image check failed: {img_err}", flush=True)
|
||||
|
||||
from tools.terminal_tool import register_task_env_overrides
|
||||
overrides = {
|
||||
"docker_image": container_image,
|
||||
@@ -233,7 +232,6 @@ def _process_single_prompt(
|
||||
"""Run the agent on one prompt; returns trajectory, stats and metadata (or a failure result)."""
|
||||
prompt = prompt_data["prompt"]
|
||||
task_id = f"task_{prompt_index}"
|
||||
|
||||
failure = _prepare_container_image(prompt_index, prompt_data, batch_num, task_id, config)
|
||||
if failure is not None:
|
||||
return failure
|
||||
@@ -243,7 +241,6 @@ def _process_single_prompt(
|
||||
|
||||
if config.get("verbose"):
|
||||
print(f" Prompt {prompt_index}: Using toolsets {selected_toolsets}")
|
||||
|
||||
agent = AIAgent(
|
||||
model=config["model"],
|
||||
max_iterations=config["max_iterations"],
|
||||
@@ -263,7 +260,6 @@ def _process_single_prompt(
|
||||
# Stats before conversion — keep the original evaluation order.
|
||||
tool_stats = _extract_tool_stats(result["messages"])
|
||||
reasoning_stats = _extract_reasoning_stats(result["messages"])
|
||||
|
||||
trajectory = agent._convert_to_trajectory_format(result["messages"], prompt, result["completed"])
|
||||
|
||||
return {
|
||||
@@ -278,7 +274,6 @@ def _process_single_prompt(
|
||||
"toolsets_used": selected_toolsets,
|
||||
"metadata": {"batch_num": batch_num, "timestamp": datetime.now().isoformat(), "model": config["model"]},
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error processing prompt {prompt_index}: {e}")
|
||||
if config.get("verbose"):
|
||||
@@ -300,12 +295,9 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]:
|
||||
``args`` is ``(batch_num, batch_data, output_dir, completed_prompts, config)``.
|
||||
"""
|
||||
batch_num, batch_data, output_dir, completed_prompts_set, config = args
|
||||
|
||||
output_dir = Path(output_dir)
|
||||
print(f"\n🔄 Batch {batch_num}: Starting ({len(batch_data)} prompts)")
|
||||
|
||||
batch_output_file = output_dir / f"batch_{batch_num}.jsonl"
|
||||
|
||||
prompts_to_process = [
|
||||
(idx, data) for idx, data in batch_data
|
||||
if idx not in completed_prompts_set
|
||||
@@ -314,9 +306,7 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]:
|
||||
if not prompts_to_process:
|
||||
print(f"✅ Batch {batch_num}: Already completed (skipping)")
|
||||
return {"batch_num": batch_num, "processed": 0, "skipped": len(batch_data), "tool_stats": {}, "completed_prompts": []}
|
||||
|
||||
print(f" Processing {len(prompts_to_process)} prompts (skipping {len(batch_data) - len(prompts_to_process)} already completed)")
|
||||
|
||||
batch_tool_stats = {}
|
||||
batch_reasoning_stats = dict.fromkeys(_REASONING_KEYS, 0)
|
||||
completed_in_batch = []
|
||||
@@ -348,7 +338,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]:
|
||||
tool_name: stats.get("failure", 0)
|
||||
for tool_name, stats in raw_tool_stats.items()
|
||||
}
|
||||
|
||||
_append_jsonl(batch_output_file, {
|
||||
"prompt_index": prompt_index,
|
||||
"conversations": result["trajectory"],
|
||||
@@ -360,7 +349,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]:
|
||||
"tool_stats": _normalize_tool_stats(raw_tool_stats), # {tool: {count, success, failure}}
|
||||
"tool_error_counts": _normalize_tool_error_counts(raw_error_counts) # {tool: failure_count}
|
||||
})
|
||||
|
||||
_merge_tool_stats(batch_tool_stats, result.get("tool_stats", {}))
|
||||
_merge_reasoning_stats(batch_reasoning_stats, result.get("reasoning_stats", {}))
|
||||
|
||||
@@ -371,7 +359,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]:
|
||||
print(f" {status} Prompt {prompt_index} completed")
|
||||
else:
|
||||
print(f" ❌ Prompt {prompt_index} failed (will retry on resume)")
|
||||
|
||||
print(f"✅ Batch {batch_num}: Completed ({len(prompts_to_process)} prompts processed)")
|
||||
|
||||
return {
|
||||
@@ -457,20 +444,16 @@ class BatchRunner:
|
||||
|
||||
if not validate_distribution(distribution):
|
||||
raise ValueError(f"Unknown distribution: {distribution}. Available: {list(list_distributions().keys())}")
|
||||
|
||||
self.output_dir = Path("data") / run_name
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.checkpoint_file = self.output_dir / "checkpoint.json"
|
||||
self.stats_file = self.output_dir / "statistics.json"
|
||||
|
||||
self.dataset = self._load_dataset()
|
||||
if self.max_samples and self.max_samples < len(self.dataset):
|
||||
full_count = len(self.dataset)
|
||||
self.dataset = self.dataset[:self.max_samples]
|
||||
print(f"✂️ Truncated dataset from {full_count} to {self.max_samples} samples (--max_samples)")
|
||||
|
||||
self.batches = self._create_batches()
|
||||
|
||||
print("📊 Batch Runner Initialized")
|
||||
print(f" Dataset: {self.dataset_file} ({len(self.dataset)} prompts)")
|
||||
print(f" Batch size: {self.batch_size}")
|
||||
@@ -487,7 +470,6 @@ class BatchRunner:
|
||||
"""Load JSONL entries that have a ``prompt`` field; skip blank/invalid lines."""
|
||||
if not self.dataset_file.exists():
|
||||
raise FileNotFoundError(f"Dataset file not found: {self.dataset_file}")
|
||||
|
||||
dataset = []
|
||||
with open(self.dataset_file, 'r', encoding='utf-8') as f:
|
||||
for line_num, line in enumerate(f, 1):
|
||||
@@ -537,7 +519,6 @@ class BatchRunner:
|
||||
def _save_checkpoint(self, checkpoint_data: Dict[str, Any], lock: Optional[Lock] = None):
|
||||
"""Atomically write *checkpoint_data* (stamped ``last_updated``), under *lock* if given."""
|
||||
checkpoint_data["last_updated"] = datetime.now().isoformat()
|
||||
|
||||
from utils import atomic_json_write
|
||||
if lock:
|
||||
with lock:
|
||||
@@ -557,7 +538,6 @@ class BatchRunner:
|
||||
|
||||
if not batch_files:
|
||||
return completed_prompts
|
||||
|
||||
print(f"📂 Scanning {len(batch_files)} batch files for completed prompts...")
|
||||
|
||||
for batch_file in batch_files:
|
||||
@@ -608,15 +588,12 @@ class BatchRunner:
|
||||
if not completed_prompt_texts:
|
||||
return True
|
||||
print(f" Found {len(completed_prompt_texts)} already-completed prompts by content matching")
|
||||
|
||||
filtered_entries, skipped_indices = self._filter_dataset_by_completed(completed_prompt_texts)
|
||||
|
||||
if not filtered_entries:
|
||||
print("\n✅ All prompts have already been processed!")
|
||||
return False
|
||||
|
||||
self.batches = _chunk(filtered_entries, self.batch_size)
|
||||
|
||||
_banner("📊 RESUME SUMMARY")
|
||||
print(f" Original dataset size: {len(self.dataset):,} prompts")
|
||||
print(f" Already completed: {len(skipped_indices):,} prompts")
|
||||
@@ -644,7 +621,6 @@ class BatchRunner:
|
||||
)
|
||||
else:
|
||||
worker_api_key = self.api_key
|
||||
|
||||
config = {key: getattr(self, key) for key in _AGENT_PASSTHROUGH}
|
||||
config["api_key"] = worker_api_key
|
||||
for key in ("distribution", "model", "max_iterations", "verbose", "log_prefix_chars"):
|
||||
@@ -661,7 +637,6 @@ class BatchRunner:
|
||||
(batch_num, batch_data, str(self.output_dir), completed_prompts_set, config)
|
||||
for batch_num, batch_data in enumerate(self.batches)
|
||||
]
|
||||
|
||||
print(f"✅ Created {len(tasks)} batch tasks")
|
||||
print("🚀 Starting parallel batch processing...\n")
|
||||
|
||||
@@ -697,7 +672,6 @@ class BatchRunner:
|
||||
checkpoint_data.setdefault('batch_stats', {})[str(batch_num)] = {
|
||||
key: result.get(key, 0) for key in ('processed', 'skipped', 'discarded_no_reasoning')
|
||||
}
|
||||
|
||||
checkpoint_data['completed_prompts'] = sorted(completed_prompts_set)
|
||||
self._save_checkpoint(checkpoint_data, lock=checkpoint_lock)
|
||||
except Exception as ckpt_err:
|
||||
@@ -725,12 +699,10 @@ class BatchRunner:
|
||||
"""
|
||||
combined_file = self.output_dir / "trajectories.jsonl"
|
||||
print(f"\n📦 Combining ALL batch files into {combined_file.name}...")
|
||||
|
||||
total_entries = 0
|
||||
filtered_entries = 0
|
||||
tombstone_entries = 0
|
||||
batch_files_found = 0
|
||||
|
||||
all_batch_files = sorted(self.output_dir.glob("batch_*.jsonl"))
|
||||
|
||||
with open(combined_file, 'w', encoding='utf-8') as outfile:
|
||||
@@ -747,7 +719,6 @@ class BatchRunner:
|
||||
if data.get("discarded"):
|
||||
tombstone_entries += 1
|
||||
continue
|
||||
|
||||
tool_stats = data.get('tool_stats', {})
|
||||
invalid_tools = [k for k in tool_stats if k not in ALL_POSSIBLE_TOOLS]
|
||||
|
||||
@@ -756,7 +727,6 @@ class BatchRunner:
|
||||
invalid_preview = invalid_tools[0][:50] + "..." if len(invalid_tools[0]) > 50 else invalid_tools[0]
|
||||
print(f" ⚠️ Filtering corrupted entry (batch {batch_num}): invalid tool '{invalid_preview}'")
|
||||
continue
|
||||
|
||||
outfile.write(line)
|
||||
except json.JSONDecodeError:
|
||||
filtered_entries += 1
|
||||
@@ -779,14 +749,12 @@ class BatchRunner:
|
||||
|
||||
if total_tool_stats:
|
||||
sorted_tools = sorted(total_tool_stats.items(), key=lambda x: x[1]["count"], reverse=True)
|
||||
|
||||
print(f"{'Tool Name':<25} {'Count':<10} {'Success':<10} {'Failure':<10} {'Success Rate':<12}")
|
||||
print("-" * 70)
|
||||
for tool_name, stats in sorted_tools:
|
||||
print(f"{tool_name:<25} {stats['count']:<10} {stats['success']:<10} {stats['failure']:<10} {stats['success_rate']:.1f}%")
|
||||
else:
|
||||
print("No tool calls were made during this run.")
|
||||
|
||||
total_discarded = sum(r.get("discarded_no_reasoning", 0) for r in results)
|
||||
|
||||
print("\n🧠 Reasoning Coverage:")
|
||||
@@ -804,7 +772,6 @@ class BatchRunner:
|
||||
print(" No assistant turns recorded.")
|
||||
if total_discarded > 0:
|
||||
print(f" 🚫 Samples discarded (zero reasoning): {total_discarded:,}")
|
||||
|
||||
print(f"\n💾 Results saved to: {self.output_dir}")
|
||||
print(" - Trajectories: trajectories.jsonl (combined)")
|
||||
print(" - Individual batches: batch_*.jsonl (for debugging)")
|
||||
@@ -822,19 +789,15 @@ class BatchRunner:
|
||||
checkpoint_data = self._load_checkpoint()
|
||||
if checkpoint_data.get("run_name") != self.run_name:
|
||||
checkpoint_data = self._empty_checkpoint()
|
||||
|
||||
config = self._worker_config()
|
||||
|
||||
# Index tracking is secondary to content matching (backward compatibility).
|
||||
completed_prompts_set = set(checkpoint_data.get("completed_prompts", []))
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# Checkpoint writes happen in the parent process; keep a lock for safety.
|
||||
checkpoint_lock = Lock()
|
||||
|
||||
results = self._run_pool(config, checkpoint_data, completed_prompts_set, checkpoint_lock)
|
||||
|
||||
total_tool_stats = {}
|
||||
total_reasoning_stats = dict.fromkeys(_REASONING_KEYS, 0)
|
||||
for batch_result in results:
|
||||
@@ -852,9 +815,7 @@ class BatchRunner:
|
||||
total_calls = stats["success"] + stats["failure"]
|
||||
stats["success_rate"] = round(stats["success"] / total_calls * 100, 2) if total_calls > 0 else 0.0
|
||||
stats["failure_rate"] = round(stats["failure"] / total_calls * 100, 2) if total_calls > 0 else 0.0
|
||||
|
||||
kept, batch_files_found = self._combine_batch_files()
|
||||
|
||||
final_stats = {
|
||||
"run_name": self.run_name,
|
||||
"distribution": self.distribution,
|
||||
@@ -871,7 +832,6 @@ class BatchRunner:
|
||||
|
||||
with open(self.stats_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(final_stats, f, indent=2, ensure_ascii=False)
|
||||
|
||||
self._print_summary(results, total_tool_stats, total_reasoning_stats, kept, batch_files_found, start_time)
|
||||
|
||||
|
||||
@@ -956,10 +916,8 @@ def main(
|
||||
"""
|
||||
if list_distributions:
|
||||
from toolset_distributions import print_distribution_info
|
||||
|
||||
print("📊 Available Toolset Distributions")
|
||||
print("=" * 70)
|
||||
|
||||
all_dists = list_distributions()
|
||||
for dist_name in sorted(all_dists.keys()):
|
||||
print_distribution_info(dist_name)
|
||||
@@ -990,7 +948,6 @@ def main(
|
||||
raise SystemExit(1)
|
||||
reasoning_config = {"enabled": True, "effort": reasoning_effort}
|
||||
print(f"🧠 Reasoning effort: {reasoning_effort}")
|
||||
|
||||
prefill_messages = None
|
||||
if prefill_messages_file:
|
||||
try:
|
||||
@@ -1027,9 +984,7 @@ def main(
|
||||
prefill_messages=prefill_messages,
|
||||
max_samples=max_samples,
|
||||
)
|
||||
|
||||
runner.run(resume=resume)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Fatal error: {e}")
|
||||
if verbose:
|
||||
|
||||
Reference in New Issue
Block a user