mirror of
https://github.com/mindverse/Second-Me.git
synced 2026-08-25 08:31:43 +00:00
Resume training at breakpoint
This commit is contained in:
yanmuyuan
parent
c3642b0f2b
commit
b219475e04
3 changed files with 407 additions and 251 deletions
|
|
@ -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):
|
||||
"""
|
||||
更新进度消息
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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():
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue