diff --git a/lpm_kernel/api/domains/cloud_service/cloud_progress_holder.py b/lpm_kernel/api/domains/cloud_service/cloud_progress_holder.py index cc0b6e8..73bd405 100644 --- a/lpm_kernel/api/domains/cloud_service/cloud_progress_holder.py +++ b/lpm_kernel/api/domains/cloud_service/cloud_progress_holder.py @@ -581,8 +581,66 @@ class CloudProgressHolder: Returns: Dict: Progress data """ - return self.progress.to_dict() + return self.progress.data + def is_stage_completed(self, stage_name): + """ + 检查指定阶段是否已完成 + + Args: + stage_name: 阶段名称（格式化后的，如"activating_the_memory_matrix"） + + Returns: + bool: 如果阶段已完成返回True，否则返回False + """ + try: + # 加载最新的进度数据 + self._load_progress() + + # 获取阶段对象 + stage = self.progress.stage_map.get(stage_name) + if not stage: + logger.warning(f"Stage {stage_name} not found in progress data") + return False + + # 检查阶段状态 + return stage.get("status") == CloudStatus.COMPLETED and stage.get("progress") == 100.0 + except Exception as e: + logger.error(f"Error checking stage completion status: {str(e)}") + return False + + def is_step_completed(self, stage_name, step_name): + """ + 检查指定阶段中的特定步骤是否已完成 + + Args: + stage_name: 阶段名称（格式化后的，如"activating_the_memory_matrix"） + step_name: 步骤名称（如"list_documents"） + + Returns: + bool: 如果步骤已完成返回True，否则返回False + """ + try: + # 加载最新的进度数据 + self._load_progress() + + # 获取阶段对象 + stage = self.progress.stage_map.get(stage_name) + if not stage: + logger.warning(f"Stage {stage_name} not found in progress data") + return False + + # 查找步骤 + for step in stage.get("steps", []): + if step.get("name") == step_name: + return step.get("completed", False) and step.get("status") == CloudStatus.COMPLETED + + logger.warning(f"Step {step_name} not found in stage {stage_name}") + return False + except Exception as e: + logger.error(f"Error checking step completion status: {str(e)}") + return False + def update_message(self, message: str): """ 更新进度消息 diff --git a/lpm_kernel/api/domains/cloud_service/cloud_trainprocess_service.py b/lpm_kernel/api/domains/cloud_service/cloud_trainprocess_service.py index 2c9da9b..fbc187e 100644 --- a/lpm_kernel/api/domains/cloud_service/cloud_trainprocess_service.py +++ b/lpm_kernel/api/domains/cloud_service/cloud_trainprocess_service.py @@ -1,31 +1,15 @@ -import os -import re import time import json -from typing import Optional, Dict, Any, List -from pathlib import Path -import threading import os +import time +from pathlib import Path +from typing import Dict, Any -from lpm_kernel.api.domains.cloud_service.service import CloudService from lpm_kernel.api.domains.cloud_service.cloud_process_step import CloudProcessStep from lpm_kernel.api.domains.cloud_service.cloud_progress_holder import CloudProgressHolder, CloudStatus -from lpm_kernel.api.domains.trainprocess.training_params_manager import TrainingParamsManager +from lpm_kernel.api.domains.cloud_service.service import CloudService from lpm_kernel.api.domains.trainprocess.trainprocess_service import TrainProcessService -from lpm_kernel.api.services.user_llm_config_service import UserLLMConfigService -from lpm_kernel.configs.config import Config - -from lpm_kernel.kernel.l1.l1_manager import ( - extract_notes_from_documents, - document_service, - get_latest_status_bio, - get_latest_global_bio, - generate_l1_from_l0 -) -from lpm_kernel.kernel.chunk_service import ChunkService -from lpm_kernel.file_data.chunker import DocumentChunker from lpm_kernel.configs.logging import get_train_process_logger -from lpm_kernel.kernel.note_service import NoteService logger = get_train_process_logger() @@ -71,35 +55,43 @@ class CloudTrainProcessService(TrainProcessService): self.cloud_service = CloudService() @classmethod - def get_instance(cls, current_model_name: str = None, base_model=None, training_type=None, hyper_parameters=None): - """ - Get the current instance of CloudTrainProcessService + def get_instance(cls): + """Get the current instance of CloudTrainProcessService - Args: - current_model_name: Optional model name to update the instance with - base_model: Base model for cloud training (only used when creating a new instance) - training_type: Type of training (only used when creating a new instance) - hyper_parameters: Training hyperparameters (only used when creating a new instance) - Returns: CloudTrainProcessService: The singleton instance """ - if cls._instance is None: - if current_model_name is None: - logger.warning("current_model_name must be provided when creating a new instance") - return None - if base_model is None or training_type is None or hyper_parameters is None: - logger.warning("base_model, training_type, and hyper_parameters must be provided when creating a new instance") - return None - return cls(current_model_name=current_model_name, base_model=base_model, - training_type=training_type, hyper_parameters=hyper_parameters) - - if current_model_name is not None: - # Update the existing instance with new model name - cls._instance.model_name = current_model_name - cls._instance.progress = CloudProgressHolder(current_model_name) + + if cls._instance is not None: + return cls._instance + + try: - return cls._instance + params_file = Path("data/cloud_progress/cloud_training_params.json") + if params_file.exists(): + with open(params_file, "r", encoding="utf-8") as f: + params = json.load(f) + + model_name = params.get("model_name") + base_model = params.get("base_model") + training_type = params.get("training_type", "efficient_sft") + hyper_parameters = params.get("hyper_parameters", {}) + + if model_name and base_model: + logger.info(f"Loaded training parameters for model {model_name} from file") + + cls._instance = cls(current_model_name=model_name, + base_model=base_model, + training_type=training_type, + hyper_parameters=hyper_parameters) + return cls._instance + else: + logger.warning("Invalid training parameters in file: missing model_name or base_model") + except Exception as e: + logger.warning(f"Failed to load training parameters from file: {str(e)}") + + logger.warning("No valid training parameters found in file") + return None @@ -110,43 +102,71 @@ class CloudTrainProcessService(TrainProcessService): logger.info("Executing memory matrix activation steps...") stage_name = "activating_the_memory_matrix" stage = self.progress.progress.stage_map.get(stage_name) - total_steps = len(stage["steps"]) if stage else 4 - # 1. 列出文档 - logger.info("Step 1.1: Listing documents...") - if not super().list_documents(): - logger.error("Failed to list documents") - return False + # 检查该阶段是否已完成，如果已完成则跳过 + if self.progress.is_stage_completed(stage_name): + logger.info(f"Stage '{stage_name}' already completed, skipping...") + else: + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False - # 更新第一步完成后的进度 - if stage: - stage["progress"] = 25.0 # 第一步完成，进度25% - stage["status"] = CloudStatus.IN_PROGRESS - # 更新步骤状态 - if len(stage["steps"]) > 0: - stage["steps"][0]["completed"] = True - stage["steps"][0]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 25% after completing list_documents") - self._update_overall_progress() + # 1. 列出文档 + logger.info("Step 1.1: Listing documents...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "list_documents"): + logger.info("Step 'list_documents' already completed, skipping...") + else: + if not super().list_documents(): + logger.error("Failed to list documents") + return False - # 2. 生成文档嵌入 - logger.info("Step 1.2: Generating document embeddings...") - if not super().generate_document_embeddings(): - logger.error("Failed to generate document embeddings") - return False + # 更新第一步完成后的进度 + if stage: + stage["progress"] = 25.0 # 第一步完成，进度25% + stage["status"] = CloudStatus.IN_PROGRESS + # 更新步骤状态 + if len(stage["steps"]) > 0: + stage["steps"][0]["completed"] = True + stage["steps"][0]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 25% after completing list_documents") + self._update_overall_progress() - # 更新第二步完成后的进度 - if stage: - stage["progress"] = 50.0 # 第二步完成，进度50% - # 更新步骤状态 - if len(stage["steps"]) > 1: - stage["steps"][1]["completed"] = True - stage["steps"][1]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 50% after completing generate_document_embeddings") - self._update_overall_progress() + # 2. 生成文档嵌入 + logger.info("Step 1.2: Generating document embeddings...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "generate_document_embeddings"): + logger.info("Step 'generate_document_embeddings' already completed, skipping...") + else: + if not super().generate_document_embeddings(): + logger.error("Failed to generate document embeddings") + return False + + # 更新第二步完成后的进度 + if stage: + stage["progress"] = 50.0 # 第二步完成，进度50% + # 更新步骤状态 + if len(stage["steps"]) > 1: + stage["steps"][1]["completed"] = True + stage["steps"][1]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 50% after completing generate_document_embeddings") + self._update_overall_progress() # 3. 处理文档分块 logger.info("Step 1.3: Processing document chunks...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False if not super().process_chunks(): logger.error("Failed to process document chunks") return False @@ -163,6 +183,10 @@ class CloudTrainProcessService(TrainProcessService): # 4. 生成分块嵌入 logger.info("Step 1.4: Generating chunk embeddings...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False if not super().chunk_embedding(): logger.error("Failed to generate chunk embeddings") return False @@ -182,113 +206,167 @@ class CloudTrainProcessService(TrainProcessService): logger.info("Executing life narrative synthesis steps...") stage_name = "synthesize_your_life_narrative" stage = self.progress.progress.stage_map.get(stage_name) - total_steps = len(stage["steps"]) if stage else 3 - # 1. 提取维度主题 - logger.info("Step 2.1: Extracting dimensional topics...") - if not super().extract_dimensional_topics(): - logger.error("Failed to extract dimensional topics") - return False + # 检查该阶段是否已完成，如果已完成则跳过 + if self.progress.is_stage_completed(stage_name): + logger.info(f"Stage '{stage_name}' already completed, skipping...") + else: + # 1. 提取维度主题 + logger.info("Step 2.1: Extracting dimensional topics...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "extract_dimensional_topics"): + logger.info("Step 'extract_dimensional_topics' already completed, skipping...") + else: + if not super().extract_dimensional_topics(): + logger.error("Failed to extract dimensional topics") + return False - # 更新第一步完成后的进度 - if stage: - stage["progress"] = 33.0 # 第一步完成，进度33% - stage["status"] = CloudStatus.IN_PROGRESS - # 更新步骤状态 - if len(stage["steps"]) > 0: - stage["steps"][0]["completed"] = True - stage["steps"][0]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 33% after completing extract_dimensional_topics") - self._update_overall_progress() + # 更新第一步完成后的进度 + if stage: + stage["progress"] = 33.0 # 第一步完成，进度33% + stage["status"] = CloudStatus.IN_PROGRESS + # 更新步骤状态 + if len(stage["steps"]) > 0: + stage["steps"][0]["completed"] = True + stage["steps"][0]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 33% after completing extract_dimensional_topics") + self._update_overall_progress() - # 2. 生成传记 - logger.info("Step 2.2: Generating biography...") - if not super().generate_biography(): - logger.error("Failed to generate biography") - return False + # 2. 生成传记 + logger.info("Step 2.2: Generating biography...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "generate_biography"): + logger.info("Step 'generate_biography' already completed, skipping...") + else: + if not super().generate_biography(): + logger.error("Failed to generate biography") + return False - # 更新第二步完成后的进度 - if stage: - stage["progress"] = 66.0 # 第二步完成，进度66% - # 更新步骤状态 - if len(stage["steps"]) > 1: - stage["steps"][1]["completed"] = True - stage["steps"][1]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 66% after completing generate_biography") - self._update_overall_progress() + # 更新第二步完成后的进度 + if stage: + stage["progress"] = 66.0 # 第二步完成，进度66% + # 更新步骤状态 + if len(stage["steps"]) > 1: + stage["steps"][1]["completed"] = True + stage["steps"][1]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 66% after completing generate_biography") + self._update_overall_progress() - # 3. 映射实体网络 - logger.info("Step 2.3: Mapping entity network...") - if not super().map_your_entity_network(): - logger.error("Failed to map entity network") - return False + # 3. 映射实体网络 + logger.info("Step 2.3: Mapping entity network...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "map_your_entity_network"): + logger.info("Step 'map_your_entity_network' already completed, skipping...") + else: + if not super().map_your_entity_network(): + logger.error("Failed to map entity network") + return False - # 更新第二阶段完成后的进度为100%并标记为已完成 - if stage: - stage["progress"] = 100.0 # 全部完成，进度100% - stage["status"] = CloudStatus.COMPLETED - # 更新最后一个步骤状态 - if len(stage["steps"]) > 2: - stage["steps"][2]["completed"] = True - stage["steps"][2]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 100% and status to COMPLETED") - self._update_overall_progress() + # 更新第二阶段完成后的进度为100%并标记为已完成 + if stage: + stage["progress"] = 100.0 # 全部完成，进度100% + stage["status"] = CloudStatus.COMPLETED + # 更新最后一个步骤状态 + if len(stage["steps"]) > 2: + stage["steps"][2]["completed"] = True + stage["steps"][2]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 100% and status to COMPLETED") + self._update_overall_progress() # 执行第三阶段的步骤（Prepare Training Data for Deep Comprehension） logger.info("Executing training data preparation steps...") stage_name = "prepare_training_data_for_deep_comprehension" stage = self.progress.progress.stage_map.get(stage_name) - total_steps = len(stage["steps"]) if stage else 3 - # 1. 解码偏好模式 - logger.info("Step 3.1: Decoding preference patterns...") - if not super().decode_preference_patterns(): - logger.error("Failed to decode preference patterns") - return False + # 检查该阶段是否已完成，如果已完成则跳过 + if self.progress.is_stage_completed(stage_name): + logger.info(f"Stage '{stage_name}' already completed, skipping...") + else: + # 1. 解码偏好模式 + logger.info("Step 3.1: Decoding preference patterns...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "decode_preference_patterns"): + logger.info("Step 'decode_preference_patterns' already completed, skipping...") + else: + if not super().decode_preference_patterns(): + logger.error("Failed to decode preference patterns") + return False - # 更新第一步完成后的进度 - if stage: - stage["progress"] = 33.0 # 第一步完成，进度33% - stage["status"] = CloudStatus.IN_PROGRESS - # 更新步骤状态 - if len(stage["steps"]) > 0: - stage["steps"][0]["completed"] = True - stage["steps"][0]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 33% after completing decode_preference_patterns") - self._update_overall_progress() + # 更新第一步完成后的进度 + if stage: + stage["progress"] = 33.0 # 第一步完成，进度33% + stage["status"] = CloudStatus.IN_PROGRESS + # 更新步骤状态 + if len(stage["steps"]) > 0: + stage["steps"][0]["completed"] = True + stage["steps"][0]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 33% after completing decode_preference_patterns") + self._update_overall_progress() - # 2. 强化身份 - logger.info("Step 3.2: Reinforcing identity...") - if not super().reinforce_identity(): - logger.error("Failed to reinforce identity") - return False + # 2. 强化身份 + logger.info("Step 3.2: Reinforcing identity...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "reinforce_identity"): + logger.info("Step 'reinforce_identity' already completed, skipping...") + else: + if not super().reinforce_identity(): + logger.error("Failed to reinforce identity") + return False - # 更新第二步完成后的进度 - if stage: - stage["progress"] = 66.0 # 第二步完成，进度66% - # 更新步骤状态 - if len(stage["steps"]) > 1: - stage["steps"][1]["completed"] = True - stage["steps"][1]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 66% after completing reinforce_identity") - self._update_overall_progress() + # 更新第二步完成后的进度 + if stage: + stage["progress"] = 66.0 # 第二步完成，进度66% + # 更新步骤状态 + if len(stage["steps"]) > 1: + stage["steps"][1]["completed"] = True + stage["steps"][1]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 66% after completing reinforce_identity") + self._update_overall_progress() - # 3. 增强内容保留 - logger.info("Step 3.3: Augmenting content retention...") - if not super().augment_content_retention(): - logger.error("Failed to augment content retention") - return False + # 3. 增强内容保留 + logger.info("Step 3.3: Augmenting content retention...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling data preparation") + return False + # 检查该步骤是否已完成 + if self.progress.is_step_completed(stage_name, "augment_content_retention"): + logger.info("Step 'augment_content_retention' already completed, skipping...") + else: + if not super().augment_content_retention(): + logger.error("Failed to augment content retention") + return False - # 更新第三阶段完成后的进度为100%并标记为已完成 - if stage: - stage["progress"] = 100.0 # 全部完成，进度100% - stage["status"] = CloudStatus.COMPLETED - # 更新最后一个步骤状态 - if len(stage["steps"]) > 2: - stage["steps"][2]["completed"] = True - stage["steps"][2]["status"] = CloudStatus.COMPLETED - logger.info(f"Updated {stage_name} progress to 100% and status to COMPLETED") - self._update_overall_progress() + # 更新第三阶段完成后的进度为100%并标记为已完成 + if stage: + stage["progress"] = 100.0 # 全部完成，进度100% + stage["status"] = CloudStatus.COMPLETED + # 更新最后一个步骤状态 + if len(stage["steps"]) > 2: + stage["steps"][2]["completed"] = True + stage["steps"][2]["status"] = CloudStatus.COMPLETED + logger.info(f"Updated {stage_name} progress to 100% and status to COMPLETED") + self._update_overall_progress() # 计算并更新整体进度 self._update_overall_progress() @@ -304,58 +382,62 @@ class CloudTrainProcessService(TrainProcessService): if step["name"].lower().replace(" ", "_") == current_stage and step["current_step"]: stage_name = current_stage step_name = step["current_step"].lower().replace(" ", "_") - self.progress.mark_step_status(stage_name, step_name, CloudStatus.FAILED, f"Error: {str(e)}") + self.progress.mark_step_status(stage_name, step_name, CloudStatus.FAILED) break return False - - def get_training_status(self) -> Dict[str, Any]: - """Get current training status""" - return self.progress.get_progress() - def start_process(self) -> bool: """Start the cloud training process using CloudService""" + self.is_stopped = False + + self.current_pid = os.getpid() + logger.info(f"Cloud training process started with PID: {self.current_pid}") + logger.info(f"Using base_model: {self.base_model}, training_type: {self.training_type}") + logger.info(f"CloudService initialized with API key: {self.cloud_service.api_key is not None}") + + logger.info("Step 1: Preparing training data...") + + success = self.prepare_training_data() + logger.info(f"Training data preparation result: {success}") + if not success: + logger.error("Failed to prepare training data") + return False + + deploy_success = self.cloud_deploy() + logger.info(f"Cloud deploy result: {deploy_success}") + if not deploy_success: + logger.error("Failed to cloud deploy") + return False + + return True + + def cloud_deploy(self) -> bool: try: - self.is_stopped = False - # Store the current process PID - self.current_pid = os.getpid() - logger.info(f"Cloud training process started with PID: {self.current_pid}") - logger.info(f"Using base_model: {self.base_model}, training_type: {self.training_type}") - logger.info(f"CloudService initialized with API key: {self.cloud_service.api_key is not None}") - - # 1. 准备训练数据（生成L2级别数据） - logger.info("Step 1: Preparing training data...") - - success = self.prepare_training_data() - logger.info(f"Training data preparation result: {success}") - if not success: - logger.error("Failed to prepare training data") - return False - - - # 7. 上传训练数据 logger.info("Step 7: Uploading training data...") + # 检查是否已停止 + if self.is_stopped: + logger.info("Process has been stopped, cancelling cloud deployment") + return False + self.progress.mark_step_status(CloudProcessStep.UPLOAD_TRAINING_DATA, CloudStatus.IN_PROGRESS) try: - # 直接使用upload_training_file的默认参数 file_id = self.cloud_service.upload_training_file() logger.info(f"File upload result: file_id={file_id}") except Exception as e: logger.error(f"Exception during file upload: {str(e)}", exc_info=True) self.progress.mark_step_status(CloudProcessStep.UPLOAD_TRAINING_DATA, CloudStatus.FAILED) return False - + if not file_id: logger.error("Failed to upload training data") self.progress.mark_step_status(CloudProcessStep.UPLOAD_TRAINING_DATA, CloudStatus.FAILED) return False self.progress.mark_step_status(CloudProcessStep.UPLOAD_TRAINING_DATA, CloudStatus.COMPLETED) - - # 8. 创建微调任务 + logger.info("Step 8: Creating fine-tune job...") self.progress.mark_step_status(CloudProcessStep.CREATE_FINE_TUNE_JOB, CloudStatus.IN_PROGRESS) - + try: success_id = self.cloud_service.create_fine_tune_job( base_model=self.base_model, @@ -364,18 +446,18 @@ class CloudTrainProcessService(TrainProcessService): ) try: current_dir = Path(__file__).parent - + job_file_path = current_dir / "job_id.json" - + job_info = { "job_id": success_id, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "status": "completed" } - + with open(job_file_path, "w") as f: json.dump(job_info, f, indent=2) - + logger.info(f"Job ID information saved to {job_file_path}") except Exception as e: logger.error(f"Failed to write job ID to file: {str(e)}", exc_info=True) @@ -385,24 +467,24 @@ class CloudTrainProcessService(TrainProcessService): logger.error(f"Exception during fine-tune job creation: {str(e)}", exc_info=True) self.progress.mark_step_status(CloudProcessStep.CREATE_FINE_TUNE_JOB, CloudStatus.FAILED) return False - + if success_id is None: logger.error("Failed to create fine-tune job") self.progress.mark_step_status(CloudProcessStep.CREATE_FINE_TUNE_JOB, CloudStatus.FAILED) return False - + self.job_id = success_id logger.info(f"Job ID set: {self.job_id}") - # 将job_id保存到进度条中 + self.progress.job_id = self.job_id self.progress.progress.data["job_id"] = self.job_id self.progress.mark_step_status(CloudProcessStep.CREATE_FINE_TUNE_JOB, CloudStatus.COMPLETED) - - # 9. 等待微调完成 + logger.info("Step 9: Waiting for fine-tune job to complete...") + self.progress.mark_step_status(CloudProcessStep.WAIT_FOR_FINE_TUNE_COMPLETION, CloudStatus.IN_PROGRESS) self._wait_for_completion_thread(self.cloud_service, self.job_id) - + logger.info("Cloud training process completed successfully") return True except Exception as e: @@ -410,7 +492,7 @@ class CloudTrainProcessService(TrainProcessService): if self.current_step: self.progress.mark_step_status(self.current_step, CloudStatus.FAILED, f"Error: {str(e)}") return False - + def _wait_for_completion_thread(self, cloud_service, job_id): try: logger.info(f"Async thread: waiting for job {job_id} to complete") @@ -459,36 +541,31 @@ class CloudTrainProcessService(TrainProcessService): logger.error(f"Fine-tuning job failed") except Exception as e: logger.error(f"Error in async wait thread: {str(e)}", exc_info=True) - self.progress.mark_step_status("cloud_training", "wait_for_fine-tune_completion", CloudStatus.FAILED, f"Error: {str(e)}") + self.progress.mark_step_status("cloud_training", "wait_for_fine-tune_completion", CloudStatus.FAILED) def _update_overall_progress(self): """Calculate and update the overall progress based on the stages' progress""" try: - # 获取所有阶段 - stages = self.progress.data["stages"] + stages = self.progress.progress.data["stages"] total_stages = len(stages) completed_stages = 0 total_progress = 0.0 - - # 计算已完成的阶段数和总进度 + for stage in stages: total_progress += stage["progress"] if stage["status"] == CloudStatus.COMPLETED: completed_stages += 1 - - # 计算整体进度（所有阶段进度的平均值） + if total_stages > 0: overall_progress = total_progress / total_stages else: overall_progress = 0.0 - - # 更新整体进度 - self.progress.data["overall_progress"] = overall_progress + + self.progress.progress.data["overall_progress"] = overall_progress logger.info(f"Updated overall progress to {overall_progress:.2f}%") - - # 如果所有阶段都已完成，将整体状态设置为已完成 + if completed_stages == total_stages: - self.progress.data["status"] = CloudStatus.COMPLETED + self.progress.progress.data["status"] = CloudStatus.COMPLETED logger.info("All stages completed, setting overall status to COMPLETED") # 保存进度 @@ -552,9 +629,6 @@ class CloudTrainProcessService(TrainProcessService): break break - # 设置整体消息 - self.progress.set_message("Cloud process cancelled by user") - return any_operation_succeeded or not (self.job_id) except Exception as e: diff --git a/lpm_kernel/api/domains/cloud_service/routes.py b/lpm_kernel/api/domains/cloud_service/routes.py index ca87547..91e4905 100644 --- a/lpm_kernel/api/domains/cloud_service/routes.py +++ b/lpm_kernel/api/domains/cloud_service/routes.py @@ -10,6 +10,7 @@ import time import json import tempfile import threading +from datetime import datetime from pathlib import Path from ....configs.config import Config from .service import CloudService @@ -98,6 +99,36 @@ def list_available_models(): # ============= Cloud Training Process Routes ============= +@cloud_bp.route("/train/resume", methods=["POST"]) +def resume_cloud_training(): + """Resume cloud training process from the last checkpoint + + Request: JSON object, containing: + - model_name: str, optional, the model name to resume training for + - base_model: str, optional, the base model to use for fine-tuning + - training_type: str, optional, the training type to use + - hyper_parameters: dict, optional, hyperparameters for training + + If model_name is not provided, the system will attempt to resume the most recent training process. + """ + try: + + cloud_train_service = CloudTrainProcessService.get_instance() + + if cloud_train_service is None: + logger.warning("No training parameters found in file") + return jsonify(APIResponse.error("No training parameters found. Please use /train/start endpoint for initial training.")) + + thread = threading.Thread(target=cloud_train_service.start_process) + thread.daemon = True + thread.start() + + return jsonify(APIResponse.success("Cloud resume training process started successfully")) + + except Exception as e: + logger.error(f"Failed to resume cloud training: {str(e)}", exc_info=True) + return jsonify(APIResponse.error(f"Failed to resume cloud training: {str(e)}")) + @cloud_bp.route("/train/stop", methods=["POST"]) def stop_cloud_training(): """Stop cloud training process @@ -108,44 +139,19 @@ def stop_cloud_training(): If model_name is not provided, the system will attempt to stop the most recent training process. """ try: - data = request.json or {} - model_name = data.get("model_name") - - # 如果没有提供model_name，尝试从job_id.json文件中获取最近的训练任务 - if not model_name: - try: - current_dir = Path(__file__).parent - job_file_path = current_dir / "job_id.json" - - if job_file_path.exists(): - with open(job_file_path, "r") as f: - job_info = json.load(f) - job_id = job_info.get("job_id") - if job_id: - logger.info(f"Found job_id {job_id} from job_id.json") - # 使用时间戳作为模型名称 - model_name = time.strftime("%Y%m%d_%H%M%S") - else: - logger.warning("No job_id.json file found") - except Exception as e: - logger.error(f"Failed to read job ID from file: {str(e)}", exc_info=True) - - if not model_name: - return jsonify(APIResponse.error("No model_name provided and no active training job found")) - - # 获取CloudTrainProcessService实例 - train_service = CloudTrainProcessService.get_instance(current_model_name=model_name) + + train_service = CloudTrainProcessService.get_instance() if not train_service: - return jsonify(APIResponse.error(f"No training service found for model: {model_name}")) + return jsonify(APIResponse.error("No training parameters found. Please use /train/start endpoint for initial training.")) # 停止训练进程 success = train_service.stop_process() if success: - return jsonify(APIResponse.success(message=f"Cloud training process for model {model_name} stopped successfully")) + return jsonify(APIResponse.success(message=f"Cloud training process stopped successfully")) else: - return jsonify(APIResponse.error(f"Failed to stop cloud training process for model {model_name}")) + return jsonify(APIResponse.error(f"Failed to stop cloud training process")) except Exception as e: logger.error(f"Failed to stop cloud training process: {str(e)}", exc_info=True) @@ -164,6 +170,24 @@ def start_cloud_training(): training_type = data.get("training_type", "efficient_sft") hyper_parameters = data.get("hyper_parameters", {}) + training_params = { + "model_name": model_name, + "base_model": base_model, + "training_type": training_type, + "hyper_parameters": hyper_parameters, + "created_at": datetime.now().isoformat() + } + + + params_dir = Path("data/cloud_progress") + params_dir.mkdir(parents=True, exist_ok=True) + + params_file = params_dir / "cloud_training_params.json" + with open(params_file, "w", encoding="utf-8") as f: + json.dump(training_params, f, indent=2, ensure_ascii=False) + + logger.info(f"Training parameters saved to {params_file}") + train_service = CloudTrainProcessService(current_model_name=model_name, base_model=base_model, training_type=training_type, hyper_parameters=hyper_parameters) def async_train_process():