""" TD Dataset Finder v3 — HARD DATA EDITION v3 changes (March 2026): - HARD datasets first, easy ones last (model at 95% needs challenge, not basics) - Added: competition math (MATH/AMC/AIME), PhD-level knowledge (GPQA), competitive programming (code_contests, TACO), evolved instructions (WizardLM) - Removed easy-first ordering that caused training loss = 0.0000 - Fixed: MMLU-Pro split, HellaSwag split v2 changes: - Known-good fallback datasets tried FIRST (reliable, parseable) - HuggingFace search only used if fallbacks don't have enough - More fallback datasets added per category (4-5 each) - Retention floor increased to 20% (prevents catastrophic forgetting) - Stall detection in fallback downloads (skip after 500 unparseable items) Takes the weakness report from td_weakness.py and finds training data to fix those weaknesses. Searches multiple sources in priority order. This is Step 3 of the self-improvement loop. Sources (in order of priority): 1. Known-good fallback datasets (curated, tested, parseable) 2. HuggingFace Datasets API search (if fallbacks aren't enough) 3. GitHub — repos with training data, benchmarks, problem sets 4. Kaggle — competitions and datasets 5. (Future) Web scraping — Stack Overflow, math forums, etc. Input: - WeaknessReport from td_weakness.py Output: - Formatted training data in chat template, ready for training - Data allocation follows the weakness report (worse = more data) """ import json import os import random import time from pathlib import Path from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, field # ============================================================ # DATA STRUCTURES # ============================================================ @dataclass class DatasetResult: """A dataset found from searching.""" name: str source: str # "huggingface", "github", "kaggle", "web" url: str description: str size_estimate: int # approximate number of samples relevance_score: float # 0-1, how well it matches the weakness category: str # which weakness category it's for @dataclass class FormattedSample: """A single training sample in chat template format.""" system: str user: str assistant: str category: str source: str @dataclass class TrainingData: """Complete training data package ready for the trainer.""" samples: List[FormattedSample] sources_used: List[DatasetResult] allocation: Dict[str, int] # category -> actual samples collected total_samples: int categories_covered: List[str] categories_missing: List[str] # categories we couldn't find data for # ============================================================ # CHAT TEMPLATE FORMATTER # ============================================================ def format_chat_template( system: str, user: str, assistant: str, ) -> str: """ Format a sample into Qwen3-VL's chat template. This is the <|im_start|> format that selfimprove.py already uses. """ parts = [] if system: parts.append(f"<|im_start|>system\n{system}<|im_end|>") parts.append(f"<|im_start|>user\n{user}<|im_end|>") parts.append(f"<|im_start|>assistant\n{assistant}<|im_end|>") return "\n".join(parts) def format_sample_to_dict(sample: FormattedSample) -> Dict: """Convert a FormattedSample to the dict format selfimprove.py expects.""" return { "text": format_chat_template(sample.system, sample.user, sample.assistant), "category": sample.category, "source": sample.source, } # ============================================================ # HUGGINGFACE SEARCH # ============================================================ def search_huggingface( keywords: List[str], tags: List[str], category: str, max_datasets: int = 5, ) -> List[DatasetResult]: """ Search HuggingFace Datasets API for matching datasets. Uses the huggingface_hub library to search by keyword and tags. Returns metadata about found datasets (doesn't download yet). """ results = [] try: from huggingface_hub import HfApi api = HfApi() # Search by each keyword seen = set() for keyword in keywords[:3]: # Top 3 keywords try: datasets = api.list_datasets( search=keyword, sort="downloads", direction=-1, limit=max_datasets, ) for ds in datasets: if ds.id in seen: continue seen.add(ds.id) # Estimate relevance based on tag overlap ds_tags = set(ds.tags or []) tag_overlap = len(ds_tags.intersection(set(tags))) relevance = min(0.5 + (tag_overlap * 0.15), 1.0) # Boost if keyword appears in name if any(kw.lower() in ds.id.lower() for kw in keyword.split()): relevance = min(relevance + 0.2, 1.0) results.append(DatasetResult( name=ds.id, source="huggingface", url=f"https://huggingface.co/datasets/{ds.id}", description=ds.id, # HF API doesn't always return description size_estimate=ds.downloads or 0, # Use downloads as proxy relevance_score=relevance, category=category, )) except Exception as e: print(f" Warning: HF search failed for '{keyword}': {e}") continue except ImportError: print(" Warning: huggingface_hub not installed. Install with: pip install huggingface_hub") return [] # Sort by relevance results.sort(key=lambda r: r.relevance_score, reverse=True) return results[:max_datasets] def download_huggingface_dataset( dataset_name: str, category: str, num_samples: int, split: str = "train", ) -> List[FormattedSample]: """ Download and format samples from a HuggingFace dataset. Handles common dataset formats: - QA pairs (question/answer columns) - Instruction pairs (instruction/output columns) - Chat format (conversations column) - Text only (text column) — wrap in QA format """ samples = [] try: from datasets import load_dataset print(f" Downloading {dataset_name} (split={split})...") try: ds = load_dataset(dataset_name, split=split, streaming=True) except Exception: # Try without split specification try: ds = load_dataset(dataset_name, streaming=True) # Get first available split if hasattr(ds, 'keys'): split_name = list(ds.keys())[0] ds = ds[split_name] else: return [] except Exception as e: print(f" Failed to load {dataset_name}: {e}") return [] # Collect samples (streaming so we don't download everything) # Timeout: 5 min per dataset to prevent hanging on slow/broken sources import time as _time download_start = _time.time() download_timeout = 300 # 5 minutes max per dataset count = 0 stall_count = 0 for item in ds: if count >= num_samples: break if _time.time() - download_start > download_timeout: print(f" Timeout: {dataset_name} took >{download_timeout}s — moving on with {count} samples") break sample = _format_hf_item(item, category, dataset_name) if sample: samples.append(sample) count += 1 stall_count = 0 else: stall_count += 1 # If 500 consecutive items produce nothing, dataset format is wrong if stall_count > 500: print(f" {dataset_name}: 500 consecutive unparseable items — skipping") break print(f" Got {len(samples)} samples from {dataset_name}") except ImportError: print(" Warning: datasets library not installed. Install with: pip install datasets") except Exception as e: print(f" Error downloading {dataset_name}: {e}") return samples def _format_hf_item(item: Dict, category: str, source: str) -> Optional[FormattedSample]: """ Try to format a HuggingFace dataset item into a chat sample. Handles many common column naming conventions. """ # Guard: item must be a dict if not isinstance(item, dict): return None # Common column name patterns for user/assistant content user_keys = ["question", "instruction", "input", "prompt", "query", "problem", "text"] assistant_keys = ["answer", "output", "response", "solution", "target", "completion"] system_keys = ["system", "system_prompt", "context"] user_text = None assistant_text = None system_text = "" # Try to find user content (guard: value might be non-string like int, list, etc.) for key in user_keys: if key in item and item[key]: val = item[key] if isinstance(val, (str, int, float)): user_text = str(val).strip() if user_text: break # Skip lists, dicts, None-like values # Try to find assistant content for key in assistant_keys: if key in item and item[key]: val = item[key] if isinstance(val, (str, int, float)): assistant_text = str(val).strip() if assistant_text: break # Try to find system content for key in system_keys: if key in item and item[key]: val = item[key] if isinstance(val, (str, int, float)): system_text = str(val).strip() break # Handle conversation format if not user_text and "conversations" in item: convs = item["conversations"] if isinstance(convs, list) and len(convs) >= 2: for c in convs: if not isinstance(c, dict): continue # Skip malformed conversation entries role = str(c.get("role", c.get("from", ""))).strip() content = c.get("content", c.get("value", "")) if content is None: continue content = str(content).strip() if role in ["user", "human"] and not user_text: user_text = content elif role in ["assistant", "gpt", "bot"] and not assistant_text: assistant_text = content elif role == "system" and not system_text: system_text = content # Handle messages format (OpenAI-style) if not user_text and "messages" in item: msgs = item["messages"] if isinstance(msgs, list): for m in msgs: if not isinstance(m, dict): continue # Skip malformed message entries role = str(m.get("role", "")).strip() content = m.get("content", "") if content is None: continue content = str(content).strip() if role == "user" and not user_text: user_text = content elif role == "assistant" and not assistant_text: assistant_text = content elif role == "system" and not system_text: system_text = content # Handle MMLU-style multiple choice: question + choices + answer_index if not assistant_text and "choices" in item and "answer" in item: choices = item["choices"] answer_idx = item["answer"] if isinstance(choices, list) and len(choices) > 0 and len(choices) <= 26: # Format as multiple choice question (cap at 26 = A-Z) if user_text: choice_strs = [f"{chr(65+i)}. {c}" for i, c in enumerate(choices)] user_text = user_text + "\n" + "\n".join(choice_strs) if isinstance(answer_idx, int) and 0 <= answer_idx < len(choices): assistant_text = f"{chr(65+answer_idx)}. {choices[answer_idx]}" elif isinstance(answer_idx, str) and len(answer_idx) == 1 and answer_idx.isalpha(): idx = ord(answer_idx.upper()) - 65 if 0 <= idx < len(choices): assistant_text = f"{answer_idx.upper()}. {choices[idx]}" # Handle GSM8K-style: answer contains "#### final_number" if assistant_text and "####" in assistant_text: # Keep the full chain-of-thought — it's good training data for reasoning pass # Need at least user + assistant if not user_text or not assistant_text: return None # Skip very short or very long samples if len(user_text) < 10 or len(assistant_text) < 5: # Allow short answers (like "42") return None if len(user_text) > 5000 or len(assistant_text) > 10000: return None return FormattedSample( system=system_text, user=user_text, assistant=assistant_text, category=category, source=source, ) # ============================================================ # WELL-KNOWN DATASETS PER CATEGORY # ============================================================ # If the search finds nothing, fall back to these known-good datasets # Verified dataset names on HuggingFace (March 2026) # Each tuple: (dataset_name, config_or_subset) # If a dataset fails to load, it moves to the next one silently # Each tuple: (dataset_name, config_or_subset, split_override) # split_override: if None, uses "train". Set to specific split name if needed. # If a dataset fails to load, it moves to the next one silently. # # IMPORTANT: HARD datasets come FIRST. The model is at 95% — it already knows # basic stuff like GSM8K and alpaca. Training loss = 0 on easy data. # We need competition math, PhD-level knowledge, complex code, etc. FALLBACK_DATASETS = { "math": [ # HARD FIRST — competition math, not grade school ("lighteval/MATH", "all", None), # 12.5K, AMC/AIME competition math ("AI-MO/NuminaMath-CoT", None, None), # 860K chain-of-thought math (diverse difficulty) ("TIGER-Lab/MathInstruct", None, None), # 262K math instruction pairs (mixed difficulty) ("microsoft/orca-math-word-problems-200k", None, None), # 200K word problems (harder than GSM8K) ("openai/gsm8k", "main", None), # 8.8K grade school — LAST (easiest) ], "code": [ # HARD FIRST — competitive programming, real-world code ("deepmind/code_contests", None, None), # Google Code Jam / Codeforces level ("BAAI/TACO", None, None), # Hard algorithmic problems ("m-a-p/CodeFeedback-Filtered-Instruction", None, None), # 157K code instructions ("sahil2801/CodeAlpaca-20k", None, None), # 20K code instruction pairs ("openai/openai_humaneval", None, "test"), # 164 problems — split is "test" ], "reasoning": [ # HARD FIRST — multi-step reasoning, not simple logic ("allenai/ai2_arc", "ARC-Challenge", None), # 2.6K challenge set (hard split) ("Rowan/hellaswag", None, "validation"), # 40K commonsense (use validation) ("allenai/winogrande", "winogrande_xl", None), # 40K commonsense reasoning ("allenai/openbookqa", "main", None), # Science reasoning ("teknium/OpenHermes-2.5", None, None), # Diverse reasoning in conversations ], "creativity": [ # HARD FIRST — evolved/complex instructions, not basic templates ("WizardLMTeam/WizardLM_evol_instruct_V2_196k", None, None), # 196K evolved (HARDER by design) ("teknium/OpenHermes-2.5", None, None), # 1M diverse instruction pairs ("Open-Orca/OpenOrca", None, None), # 4.2M conversations — diverse ("tatsu-lab/alpaca", None, None), # 52K instructions — LAST (easiest) ], "knowledge": [ # HARD FIRST — graduate-level, not basic trivia ("Idavidrein/gpqa", "gpqa_extended", None), # Graduate-level science (PhD-level!) ("TIGER-Lab/MMLU-Pro", None, "test"), # Harder MMLU (10 choices, harder questions) ("cais/mmlu", "all", "test"), # Classic MMLU — split is "test" ("cais/mmlu", "all", "validation"), # Also try validation split ("truthfulqa/truthful_qa", "generation", "validation"), # Truthfulness — split is "validation" ("lukaemon/mmlu", "geography", "test"), # Geography specifically (weak spot!) ("lukaemon/mmlu", "astronomy", "test"), # Science knowledge (weak spot!) ("lukaemon/mmlu", "world_religions", "test"), # More knowledge topics ("lukaemon/mmlu", "global_facts", "test"), # General knowledge ], "instruction_following": [ # HARD FIRST — complex format constraints ("HuggingFaceH4/ultrafeedback_binarized", None, "train_prefs"), # 61K preference pairs ("WizardLMTeam/WizardLM_evol_instruct_V2_196k", None, None), # Evolved = harder instructions ("tatsu-lab/alpaca", None, None), # 52K instructions ("Intel/orca_dpo_pairs", None, None), # 12K DPO pairs ], } # ============================================================ # MAIN DATA FINDER # ============================================================ def find_training_data( search_plan: List[Dict], data_allocation: Dict[str, int], output_dir: str = "td_fuse_outputs/training_data", hf_token: Optional[str] = None, ) -> TrainingData: """ Search for and download training data based on the weakness report. This is the main entry point that td_loop.py calls. Args: search_plan: From WeaknessReport.search_plan data_allocation: From WeaknessReport.data_allocation output_dir: Where to save the formatted data hf_token: HuggingFace token for private datasets Returns: TrainingData with formatted samples ready for training """ print("\n" + "=" * 60) print("TD DATASET FINDER v1") print("=" * 60) if hf_token: os.environ["HF_TOKEN"] = hf_token all_samples = [] all_sources = [] categories_covered = [] categories_missing = [] for plan_item in search_plan: category = plan_item["category"] needed = data_allocation.get(category, 1000) keywords = plan_item["keywords"] hf_tags = plan_item.get("hf_tags", []) print(f"\n--- Searching for: {category} ({needed} samples needed) ---") category_samples = [] # Step 1: Try KNOWN-GOOD fallback datasets FIRST (reliable, parseable) fallbacks = FALLBACK_DATASETS.get(category, []) if fallbacks: print(f" Trying {len(fallbacks)} known-good datasets for {category}...") try: from datasets import load_dataset # Import once outside loop except ImportError: print(" Warning: datasets library not installed — cannot load datasets") fallbacks = [] # Skip the fallback loop for fallback_entry in fallbacks: if len(category_samples) >= needed: break # Unpack: (name, config, split_override) — split_override can be None ds_name = fallback_entry[0] config = fallback_entry[1] split_override = fallback_entry[2] if len(fallback_entry) > 2 else None use_split = split_override or "train" remaining = needed - len(category_samples) try: if config: ds = load_dataset(ds_name, config, streaming=True, split=use_split) else: ds = load_dataset(ds_name, streaming=True, split=use_split) count = 0 stall = 0 for item in ds: if count >= remaining: break sample = _format_hf_item(item, category, ds_name) if sample: category_samples.append(sample) count += 1 stall = 0 else: stall += 1 if stall > 500: break if count > 0: print(f" Got {count} from: {ds_name}") all_sources.append(DatasetResult( name=ds_name, source="huggingface_known", url=f"https://huggingface.co/datasets/{ds_name}", description=f"Known-good dataset for {category}", size_estimate=count, relevance_score=0.8, category=category, )) except Exception as e: print(f" {ds_name} failed: {e}") continue # Step 2: If still not enough, search HuggingFace for more if len(category_samples) < needed: print(f" Got {len(category_samples)}/{needed} from known datasets. Searching HuggingFace...") hf_results = search_huggingface(keywords, hf_tags, category) if hf_results: print(f" Found {len(hf_results)} datasets on HuggingFace:") for r in hf_results[:3]: print(f" - {r.name} (relevance: {r.relevance_score:.2f})") # Download from best matches for result in hf_results: if len(category_samples) >= needed: break # Skip datasets we already used as fallbacks if any(result.name == fb_entry[0] for fb_entry in fallbacks): continue remaining = needed - len(category_samples) samples = download_huggingface_dataset( dataset_name=result.name, category=category, num_samples=remaining, ) category_samples.extend(samples) if samples: all_sources.append(result) # Track what we got if category_samples: all_samples.extend(category_samples) categories_covered.append(category) print(f" Total for {category}: {len(category_samples)} samples") else: categories_missing.append(category) print(f" WARNING: No data found for {category}!") # Shuffle all samples together random.shuffle(all_samples) # Save to disk output_path = Path(output_dir) try: output_path.mkdir(parents=True, exist_ok=True) except OSError as e: print(f"\n ERROR: Cannot create output dir {output_dir}: {e}") # Still return what we collected even if can't save actual_allocation = {} for s in all_samples: actual_allocation[s.category] = actual_allocation.get(s.category, 0) + 1 return TrainingData( samples=all_samples, sources_used=all_sources, allocation=actual_allocation, total_samples=len(all_samples), categories_covered=categories_covered, categories_missing=categories_missing, ) # Save as JSONL (one sample per line) data_file = output_path / "training_data.jsonl" try: with open(data_file, "w") as f: for sample in all_samples: f.write(json.dumps(format_sample_to_dict(sample)) + "\n") print(f"\nSaved {len(all_samples)} samples to {data_file}") except OSError as e: print(f"\n WARNING: Failed to save training data: {e} (disk may be full)") # Save metadata meta_file = output_path / "data_metadata.json" actual_allocation = {} for s in all_samples: actual_allocation[s.category] = actual_allocation.get(s.category, 0) + 1 meta = { "total_samples": len(all_samples), "allocation": actual_allocation, "sources": [ {"name": s.name, "source": s.source, "category": s.category, "url": s.url} for s in all_sources ], "categories_covered": categories_covered, "categories_missing": categories_missing, } try: with open(meta_file, "w") as f: json.dump(meta, f, indent=2) print(f"Saved metadata to {meta_file}") except OSError as e: print(f" WARNING: Failed to save metadata: {e}") return TrainingData( samples=all_samples, sources_used=all_sources, allocation=actual_allocation, total_samples=len(all_samples), categories_covered=categories_covered, categories_missing=categories_missing, ) # ============================================================ # STANDALONE USAGE # ============================================================ if __name__ == "__main__": import sys if len(sys.argv) < 2: print("Usage: python td_data_finder.py [output_dir]") print("\nSearches for training data based on the weakness report.") sys.exit(1) # Load weakness report from td_weakness import load_report report = load_report(sys.argv[1]) output_dir = sys.argv[2] if len(sys.argv) > 2 else "td_fuse_outputs/training_data" # Find data training_data = find_training_data( search_plan=report.search_plan, data_allocation=report.data_allocation, output_dir=output_dir, ) print(f"\nDone! Got {training_data.total_samples} samples across {len(training_data.categories_covered)} categories.") if training_data.categories_missing: print(f"Missing data for: {', '.join(training_data.categories_missing)}")