Upload folder using huggingface_hub

run_transformers_training.py

@@ -123,30 +123,22 @@ def load_env_variables():
         os.environ["HUGGING_FACE_HUB_TOKEN"] = os.environ.get("HF_TOKEN")
 
 def load_configs(base_path):
-    """Load all configuration from a single consolidated file."""
-    configs = {}
-    
+    """Load configuration from transformers_config.json file."""
     # Using a single consolidated config file
-    config_file = "transformers_config.json"
+    config_file = base_path
     
-    file_path = os.path.join(base_path, config_file)
     try:
-        with open(file_path, "r") as f:
+        with open(config_file, "r") as f:
             config = json.load(f)
-            # Extract sections into separate config dictionaries for compatibility
-            configs["transformers"] = config
-            configs["hardware"] = config.get("hardware", {})
-            configs["dataset"] = config.get("dataset", {})
-            logger.info(f"Loaded consolidated configuration from {file_path}")
+            logger.info(f"Loaded configuration from {config_file}")
+            return config
     except Exception as e:
         logger.error(f"Error loading {config_file}: {e}")
         raise
-    
-    return configs
 
 def parse_args():
     parser = argparse.ArgumentParser(description="Fine-tune a language model on a text dataset")
-    parser.add_argument("--config_dir", type=str, default=".", help="Directory containing configuration files")
+    parser.add_argument("--config", type=str, default="transformers_config.json", help="Path to configuration file")
     return parser.parse_args()
 
 def load_model_and_tokenizer(config):
@@ -157,8 +149,8 @@ def load_model_and_tokenizer(config):
             logger.error("Please ensure unsloth is in requirements.txt")
             raise ImportError("Unsloth is required for this training setup")
         
-        # Get model name correctly from nested config structure
-        model_name = config.get("model", {}).get("name") or config.get("model_name_or_path") or config.get("model_name")
+        # Get model name correctly from config
+        model_name = config.get("model_name") or config.get("model", {}).get("name")
         logger.info(f"Loading model: {model_name}")
         
         if not model_name:
@@ -166,14 +158,12 @@ def load_model_and_tokenizer(config):
             
         logger.info("Using Unsloth optimizations with pre-quantized model")
         
-        # Check for flash attention without importing it directly
+        # Check for flash attention 
         use_flash_attention = config.get("use_flash_attention", True)
-        try:
-            import flash_attn
-            logger.info("Flash attention detected and will be used")
-        except ImportError:
+        if use_flash_attention and not find_spec("flash_attn"):
+            logger.warning("flash-attn not found. Will continue without flash attention.")
+            logger.warning("To use flash attention, install with: pip install flash-attn --no-build-isolation")
             use_flash_attention = False
-            logger.warning("Flash attention not available, falling back to standard attention")
             
         # First detect if we have a GPU
         if torch.cuda.is_available():
@@ -321,13 +311,24 @@ def load_dataset_with_mapping(dataset_config):
         
         # Add prompt_number field that increments based on original order
         def add_prompt_numbers(examples, indices):
-            # Defensive check to ensure indices is not None
+            # Defensive check to ensure indices is not None and is iterable
             if indices is None:
                 logger.warning("Warning: indices is None in add_prompt_numbers, using empty list")
                 indices = []
+            elif isinstance(indices, int):
+                # Handle case where indices is a single integer
+                logger.warning(f"Warning: indices is an integer ({indices}) in add_prompt_numbers, converting to list")
+                indices = [indices]
             
-            # Create a new field with the dataset index as the prompt number, starting at 1
-            examples["prompt_number"] = [idx + 1 for idx in indices]  # Adding 1 to make it 1-indexed
+            # Ensure indices is always a list/iterable
+            try:
+                # Create a new field with the dataset index as the prompt number, starting at 1
+                examples["prompt_number"] = [idx + 1 for idx in indices]  # Adding 1 to make it 1-indexed
+            except TypeError:
+                # Fallback for non-iterable types
+                logger.warning(f"Warning: non-iterable indices in add_prompt_numbers: {type(indices)}, using default")
+                examples["prompt_number"] = [1] * len(next(iter(examples.values())))
+                
             return examples
         
         # Add prompt numbers to the dataset based on original order
@@ -358,37 +359,73 @@ def load_dataset_with_mapping(dataset_config):
             dataset = Dataset.from_list(updated_examples)
             logger.info(f"Successfully added prompt_number field using fallback method")
         
-        # Verify expected columns exist
-        expected_columns = {"id", "conversations"}
-        for col in expected_columns:
+        # Rename 'id' to 'article_id' if it exists
+        if 'id' in dataset.column_names and 'article_id' not in dataset.column_names:
+            logger.info("Renaming 'id' column to 'article_id'")
+            dataset = dataset.rename_column('id', 'article_id')
+        
+        # Reorder columns to make prompt_number first if it exists
+        if 'prompt_number' in dataset.column_names:
+            logger.info("Reordering columns to place prompt_number first")
+            # Get current column names
+            current_columns = dataset.column_names
+            # Create new column order with prompt_number first
+            new_column_order = ['prompt_number'] + [col for col in current_columns if col != 'prompt_number']
+            # Reorder columns
+            dataset = dataset.select_columns(new_column_order)
+        
+        # Verify all new column names for logging
+        logger.info(f"Dataset loaded successfully with {len(dataset)} examples")
+        logger.info(f"Dataset columns: {dataset.column_names}")
+        
+        # Verify dataset is not empty
+        if len(dataset) == 0:
+            logger.error("Dataset is empty! This will cause errors during training.")
+            raise ValueError("Empty dataset loaded")
+            
+        # Check for required columns
+        required_columns = ['conversations']
+        for col in required_columns:
             if col not in dataset.column_names:
-                # If "conversations" is missing but "text" exists, it might need conversion
-                if col == "conversations" and "text" in dataset.column_names:
-                    logger.info("Converting 'text' field to 'conversations' format")
-                    
-                    def convert_text_to_conversations(example):
-                        # Check if text is already a list of conversation turns
-                        if isinstance(example.get("text"), list):
-                            return {"conversations": example["text"]}
-                        # Otherwise, create a simple conversation with the text as user message
-                        else:
-                            return {
-                                "conversations": [
-                                    {"role": "user", "content": str(example.get("text", ""))}
-                                ]
-                            }
-                    
-                    dataset = dataset.map(convert_text_to_conversations)
-                else:
-                    logger.warning(f"Expected column '{col}' not found in dataset")
+                logger.error(f"Required column '{col}' not found in dataset!")
+                raise ValueError(f"Required column '{col}' missing from dataset")
         
-        # Note: Explicitly NOT sorting the dataset to preserve original order
-        logger.info("Preserving original dataset order (no sorting)")
+        # Verify expected columns exist
+        expected_columns = {"article_id", "conversations", "prompt_number"}
+        missing_columns = expected_columns - set(dataset.column_names)
+        if missing_columns:
+            logger.warning(f"Some expected columns are missing: {missing_columns}")
+            
+            # If "conversations" is missing but "text" exists, attempt conversion
+            if "conversations" not in dataset.column_names and "text" in dataset.column_names:
+                logger.info("Converting 'text' field to 'conversations' format")
+                
+                def convert_text_to_conversations(example):
+                    # Check if text is already a list of conversation turns
+                    if isinstance(example.get("text"), list):
+                        example["conversations"] = example["text"]
+                    # Otherwise, create a simple conversation with the text as user message
+                    else:
+                        example["conversations"] = [
+                            {"role": "user", "content": str(example.get("text", ""))}
+                        ]
+                    return example
+                
+                dataset = dataset.map(convert_text_to_conversations)
+                logger.info("Successfully converted 'text' to 'conversations'")
         
-        # Check data ordering requirements
+        # Verify data ordering requirements
         processing_config = dataset_config.get("dataset", {}).get("processing", {})
         data_loading_config = dataset_config.get("data_loading", {})
         
+        # Check if sorting is required
+        sort_by_article_id = processing_config.get("sort_by_article_id", False)
+        if sort_by_article_id and 'article_id' in dataset.column_names:
+            logger.info("Sorting dataset by article_id as specified in config")
+            dataset = dataset.sort("article_id")
+            sorted_ids = [example['article_id'] for example in dataset.select(range(min(5, len(dataset))))]
+            logger.info(f"First few article_ids after sorting: {sorted_ids}")
+        
         # Flag consolidation - we only need one flag to control sequence preservation
         # Default to True to ensure safety
         preserve_sequence = processing_config.get("preserve_entry_sequence", True)
@@ -450,17 +487,18 @@ def load_dataset_with_mapping(dataset_config):
                             logger.warning(f"Error accessing dataset at index {i}: {e}")
                     
                     if sample_examples:
-                        if all(isinstance(example.get('id', ''), (int, str)) for example in sample_examples):
-                            sample_ids = [example.get('id', '') for example in sample_examples if 'id' in example]
+                        id_field = 'article_id' if 'article_id' in dataset.column_names else 'id'
+                        if all(isinstance(example.get(id_field, ''), (int, str)) for example in sample_examples):
+                            sample_ids = [example.get(id_field, '') for example in sample_examples if id_field in example]
                             
                             if sample_ids and all(isinstance(id, int) or (isinstance(id, str) and id.isdigit()) for id in sample_ids):
                                 numeric_ids = [int(id) if isinstance(id, str) else id for id in sample_ids]
                                 if len(numeric_ids) > 1:
                                     is_ordered = all(numeric_ids[i] <= numeric_ids[i+1] for i in range(len(numeric_ids)-1))
                                     if not is_ordered:
-                                        logger.warning("WARNING: Sample IDs are not in sequential order.")
+                                        logger.warning(f"WARNING: Sample {id_field}s are not in sequential order.")
                                     else:
-                                        logger.info("Sample IDs appear to be in sequential order.")
+                                        logger.info(f"Sample {id_field}s appear to be in sequential order.")
                 except Exception as e:
                     logger.warning(f"Error checking ID sequence: {e}")
         except Exception as e:
@@ -472,19 +510,19 @@ def load_dataset_with_mapping(dataset_config):
                 # Safely get first few samples
                 first_few_indices = range(min(5, len(dataset)))
                 sample_prompt_numbers = []
-                sample_ids = []
+                sample_article_ids = []
                 
                 for i in first_few_indices:
                     try:
                         example = dataset[i]
                         if 'prompt_number' in example:
                             sample_prompt_numbers.append(example['prompt_number'])
-                        if 'id' in example:
-                            sample_ids.append(example['id'])
+                        if 'article_id' in example:
+                            sample_article_ids.append(example['article_id'])
                     except Exception as e:
                         logger.warning(f"Error accessing sample at index {i}: {e}")
                 
-                logger.info(f"First few samples - Prompt numbers: {sample_prompt_numbers}, IDs: {sample_ids}")
+                logger.info(f"First few samples - Prompt numbers: {sample_prompt_numbers}, Article IDs: {sample_article_ids}")
             
                 # Log conversation structure without full content
                 if len(dataset) > 0:
@@ -510,6 +548,74 @@ def load_dataset_with_mapping(dataset_config):
         
         logger.info(f"Dataset loaded successfully with {len(dataset)} examples")
         logger.info(f"Dataset columns: {dataset.column_names}")
+        
+        # Verify dataset is not empty
+        if len(dataset) == 0:
+            logger.error("Dataset is empty! Cannot proceed with training.")
+            return dataset
+            
+        # Check for required columns
+        required_cols = ['conversations', 'prompt_number']
+        for col in required_cols:
+            if col not in dataset.column_names:
+                logger.error(f"Required column '{col}' missing from dataset. Cannot proceed with training.")
+                return dataset
+        
+        # Validate at least one sample can be processed
+        try:
+            if len(dataset) > 0:
+                sample = dataset[0]
+                if 'conversations' not in sample or not sample['conversations']:
+                    logger.error("First sample has no conversations! Data format may be incorrect.")
+                    return dataset
+                if not isinstance(sample['conversations'], list):
+                    logger.error(f"Conversations field should be a list but got {type(sample['conversations'])}")
+                    return dataset
+        except Exception as e:
+            logger.error(f"Error validating first sample: {e}")
+            return dataset
+        
+        # Add metadata if specified
+        metadata_config = dataset_config.get("data_formatting", {}).get("metadata_handling", {})
+        if metadata_config:
+            include_article_id = metadata_config.get("include_article_id", False)
+            include_prompt_number = metadata_config.get("include_prompt_number", False)
+            metadata_format = metadata_config.get("metadata_format", "")
+            
+            if (include_article_id or include_prompt_number) and metadata_format:
+                logger.info("Adding metadata to conversations")
+                
+                def add_metadata(example):
+                    if not example.get("conversations"):
+                        return example
+                    
+                    # Prepare metadata
+                    metadata = metadata_format
+                    if include_article_id and "article_id" in example:
+                        metadata = metadata.replace("{article_id}", str(example.get("article_id", "")))
+                    if include_prompt_number and "prompt_number" in example:
+                        metadata = metadata.replace("{prompt_number}", str(example.get("prompt_number", "")))
+                    
+                    # Add system message with metadata if not empty
+                    if metadata.strip():
+                        if example["conversations"] and isinstance(example["conversations"], list):
+                            # Check if first message is already a system message
+                            if (isinstance(example["conversations"][0], dict) and 
+                                example["conversations"][0].get("role") == "system"):
+                                # Append to existing system message
+                                example["conversations"][0]["content"] += f"\n\nMetadata: {metadata}"
+                            else:
+                                # Add new system message at the beginning
+                                example["conversations"].insert(0, {
+                                    "role": "system",
+                                    "content": f"Metadata: {metadata}"
+                                })
+                    
+                    return example
+                
+                dataset = dataset.map(add_metadata)
+                logger.info("Metadata added to conversations")
+        
         return dataset
         
     except Exception as e:
@@ -752,13 +858,13 @@ class LoggingCallback(TrainerCallback):
                                                         is_sequence_maintained = False
                                                 
                                                 # Also compare IDs as a backup check
-                                                elif ('id' in orig_sample and 
-                                                      'id' in current_sample and 
-                                                      orig_sample['id'] is not None and 
-                                                      current_sample['id'] is not None):
+                                                elif ('article_id' in orig_sample and 
+                                                      'article_id' in current_sample and 
+                                                      orig_sample['article_id'] is not None and 
+                                                      current_sample['article_id'] is not None):
                                                     
-                                                    if orig_sample['id'] != current_sample['id']:
-                                                        log_info(f"WARNING: Sequence integrity compromised! Sample {i} ID changed from {orig_sample['id']} to {current_sample['id']}")
+                                                    if orig_sample['article_id'] != current_sample['article_id']:
+                                                        log_info(f"WARNING: Sequence integrity compromised! Sample {i} article_id changed from {orig_sample['article_id']} to {current_sample['article_id']}")
                                                         is_sequence_maintained = False
                                                 
                                                 # Compare input fingerprints
@@ -899,12 +1005,11 @@ def check_dependencies():
         missing_packages.append("peft>=0.9.0")
     
     # Optional packages - don't add to missing list, just log
-    try:
-        import flash_attn
+    if find_spec("flash_attn"):
         logger.info("flash-attn found. Flash attention will be used for faster training.")
-    except ImportError:
+    else:
         logger.warning("flash-attn not found. Training will work but may be slower.")
-        # Don't add to missing packages since it's optional and can cause build issues
+        logger.warning("To use flash attention, install with: pip install flash-attn --no-build-isolation")
     
     # If critical packages are missing, exit with instructions
     if missing_packages:
@@ -918,115 +1023,44 @@ def check_dependencies():
 
 def main():
     # Set up logging
-    log_info("Starting Phi-4 fine-tuning process")
-    
-    # Log hardware information
-    log_info(f"Hardware detection: CUDA {'available' if CUDA_AVAILABLE else 'not available'}")
-    if CUDA_AVAILABLE:
-        log_info(f"Found {NUM_GPUS} GPUs")
-        for i in range(NUM_GPUS):
-            log_info(f"  GPU {i}: {torch.cuda.get_device_name(i)}")
-    else:
-        log_info("Running on CPU (training will be very slow)")
+    logger.info("Starting training process")
     
     # Parse arguments
     args = parse_args()
     
+    # Load environment variables
+    load_env_variables()
+    
+    # Load configuration
+    try:
+        transformers_config = load_configs(args.config)
+        hardware_config = transformers_config.get("hardware", {})
+        dataset_config = transformers_config.get("dataset", {})
+        logger.info("Configuration loaded successfully")
+    except Exception as e:
+        logger.error(f"Error loading configuration: {e}")
+        return 1
+    
     # Check dependencies
     if not check_dependencies():
         logger.error("Aborting due to missing critical dependencies")
         return 1
     
-    # Load environment variables
-    load_env_variables()
-    
     # Check if we're in distributed mode
     is_distributed = "WORLD_SIZE" in os.environ and int(os.environ.get("WORLD_SIZE", "1")) > 1
     if is_distributed:
-        log_info(f"Running in distributed mode with world size: {os.environ.get('WORLD_SIZE')}")
+        local_rank = int(os.environ.get("LOCAL_RANK", "0"))
+        log_info(f"Running in distributed mode with {os.environ.get('WORLD_SIZE')} processes, local_rank: {local_rank}")
     else:
         log_info("Running in non-distributed mode (single process)")
     
-    # Load all configurations - do this once
-    try:
-        configs = load_configs(args.config_dir)
-        
-        # Extract specific configs immediately after loading
-        if not configs:
-            logger.error("Failed to load configuration")
-            return 1
-        
-        # Store configurations in clear variables
-        transformers_config = configs.get("transformers", {})
-        hardware_config = configs.get("hardware", {})
-        dataset_config = configs.get("dataset", {})
-            
-        # Verify configuration sections exist
-        if not transformers_config:
-            logger.error("transformers_config.json not found or invalid")
-            return 1
-            
-        if not hardware_config:
-            logger.warning("Hardware configuration section not found in transformers_config.json. Using default hardware configuration.")
-            
-        if not dataset_config:
-            logger.error("Dataset configuration section not found in transformers_config.json")
-            return 1
-            
-        # Validate model configuration
-        model_name = (transformers_config.get("model", {}).get("name") or 
-                     transformers_config.get("model_name_or_path") or 
-                     transformers_config.get("model_name"))
-        
-        if not model_name:
-            logger.error("Model name not specified in configuration")
-            logger.error("Please ensure 'name' is specified under 'model' in transformers_config.json")
-            return 1
-            
-        log_info(f"Using model: {model_name}")
-        log_info("All configurations loaded successfully")
-        
-        # Apply hardware-specific settings if available
-        if hardware_config:
-            # Get training optimizations from hardware config
-            training_opts = hardware_config.get("training_optimizations", {})
-            
-            # Apply batch size and gradient accumulation settings
-            if training_opts.get("per_device_batch_size") and transformers_config.get("training"):
-                batch_size = training_opts.get("per_device_batch_size")
-                transformers_config["training"]["per_device_train_batch_size"] = batch_size
-                log_info(f"Applied hardware-optimized batch size: {batch_size}")
-                
-            if training_opts.get("gradient_accumulation_steps") and transformers_config.get("training"):
-                grad_steps = training_opts.get("gradient_accumulation_steps")
-                transformers_config["training"]["gradient_accumulation_steps"] = grad_steps
-                log_info(f"Applied hardware-optimized gradient accumulation: {grad_steps}")
-            
-            # Apply memory optimizations
-            memory_opts = training_opts.get("memory_optimizations", {})
-            if memory_opts.get("use_gradient_checkpointing") is not None and transformers_config.get("training"):
-                grad_ckpt = memory_opts.get("use_gradient_checkpointing")
-                transformers_config["training"]["gradient_checkpointing"] = grad_ckpt
-                log_info(f"Applied hardware-optimized gradient checkpointing: {grad_ckpt}")
-            
-            # Apply system settings
-            system_settings = hardware_config.get("system_settings", {})
-            if system_settings.get("dataloader_num_workers") is not None:
-                workers = system_settings.get("dataloader_num_workers")
-                log_info(f"Using {workers} dataloader workers from hardware config")
-            
-            # Get distribution strategy
-            multi_gpu_strategy = training_opts.get("multi_gpu_strategy", "data_parallel")
-            log_info(f"Hardware config specifies {multi_gpu_strategy} for multi-GPU training")
-                
-    except Exception as e:
-        logger.error(f"Error loading configurations: {e}")
-        return 1
-    
     # Set random seed for reproducibility
     seed = transformers_config.get("seed", 42)
     set_seed(seed)
-    log_info(f"Set random seed to {seed} for reproducibility")
+    logger.info(f"Set random seed to {seed}")
+    
+    # Load model and tokenizer using the consolidated config
+    model, tokenizer = load_model_and_tokenizer(transformers_config)
     
     # Empty CUDA cache to ensure clean state
     if CUDA_AVAILABLE:
@@ -1043,17 +1077,13 @@ def main():
             log_info(f"Set CUDA memory allocation limit to expandable with max_split_size_mb:128")
     
     try:
-        log_info("Loading model and tokenizer...")
-        model, tokenizer = load_model_and_tokenizer(transformers_config)
-        log_info("Model and tokenizer loaded successfully")
+        log_info("Loading dataset...")
+        dataset = load_dataset_with_mapping(dataset_config)
+        log_info(f"Dataset loaded with {len(dataset)} examples")
         
-        # Load dataset with proper mapping
-        try:
-            log_info(f"Loading dataset from {dataset_config.get('dataset', {}).get('name', '')}")
-            dataset = load_dataset_with_mapping(dataset_config)
-            log_info(f"Dataset loaded with {len(dataset)} examples")
-        except Exception as e:
-            logger.error(f"Error loading dataset: {e}")
+        # Minimal validation before proceeding
+        if dataset is None or len(dataset) == 0:
+            logger.error("Dataset is empty or None! Cannot proceed with training.")
             return 1
         
         # Create data collator

transformers_config.json

@@ -134,10 +134,11 @@
       "name": "George-API/cognitive-data",
       "split": "train",
       "column_mapping": {
-        "conversations": "text"
+        "conversations": "text",
+        "article_id": "id"
       },
       "processing": {
-        "sort_by_id": true,
+        "sort_by_article_id": true,
         "maintain_paper_order": true,
         "preserve_entry_sequence": true,
         "max_seq_length": 2048
@@ -152,9 +153,9 @@
         "user": "Human: {content}\n\n"
       },
       "metadata_handling": {
-        "include_paper_id": true,
-        "include_chunk_number": true,
-        "metadata_format": "Paper ID: {paper_id} | Chunk: {chunk_number}"
+        "include_article_id": true,
+        "include_prompt_number": true,
+        "metadata_format": "Article ID: {article_id} | Prompt: {prompt_number}"
       }
     },
     "data_loading": {