From e98bc4020381f0c653f8bfd39e5f7c60c108aeaa Mon Sep 17 00:00:00 2001 From: somunslotus Date: Mon, 26 Jan 2026 14:32:32 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A8=E7=90=86=E6=8E=A5=E5=8F=A3=E6=9B=B4?= =?UTF-8?q?=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/handler/handler.go | 35 +++++++++ pkg/models/handler.go | 37 ++++++++++ pkg/models/task.go | 61 +++++++++++++++ pkg/service/common/task_builder.go | 65 ++++++++++++++++ pkg/service/inference/inference_service.go | 86 ++++++++++++++++++++++ pkg/service/interface.go | 2 + 6 files changed, 286 insertions(+) diff --git a/pkg/handler/handler.go b/pkg/handler/handler.go index 55c7ce4..cf1ee3a 100644 --- a/pkg/handler/handler.go +++ b/pkg/handler/handler.go @@ -164,6 +164,41 @@ func (h *Handler) QueryStatus(c *gin.Context) { c.JSON(http.StatusOK, SuccessResponse(status)) } +// SubmitInferenceTask 处理推理任务提交请求 +func (h *Handler) SubmitInferenceTask(c *gin.Context) { + var request models.InferenceSubmitTaskRequest + if err := c.ShouldBindJSON(&request); err != nil { + c.JSON(http.StatusOK, BadRequestResponse("请求参数错误: "+err.Error())) + return + } + + // 获取认证信息 + authData, err := h.authService.GetToken(c.Request.Context()) + if err != nil { + c.JSON(http.StatusOK, InternalServerErrorResponse("获取认证信息失败: "+err.Error())) + return + } + + // 获取集群ID（从 resource 中获取，如果没有则使用默认值） + clusterID := request.Resource.ClusterID + if clusterID == "" { + clusterID, err = h.authService.GetClusterID(c.Request.Context(), "default") + if err != nil { + c.JSON(http.StatusOK, InternalServerErrorResponse("获取集群ID失败: "+err.Error())) + return + } + } + + // 提交推理任务（包含准备步骤） + response, err := h.inferenceService.SubmitInferenceTask(c.Request.Context(), authData, &request, clusterID) + if err != nil { + c.JSON(http.StatusOK, InternalServerErrorResponse("提交推理任务失败: "+err.Error())) + return + } + + c.JSON(http.StatusOK, response) +} + // HealthCheck 处理健康检查请求 func (h *Handler) HealthCheck(c *gin.Context) { c.JSON(http.StatusOK, SuccessResponse(gin.H{ diff --git a/pkg/models/handler.go b/pkg/models/handler.go index 32e8c38..35c3dde 100644 --- a/pkg/models/handler.go +++ b/pkg/models/handler.go @@ -12,3 +12,40 @@ type SubmitTaskRequest struct { type UploadCodeRequest struct { RunConfig RunConfig `json:"run_config"` // 运行配置 } + +// InferenceSubmitTaskRequest 推理任务提交请求结构体 +type InferenceSubmitTaskRequest struct { + Version string `json:"version"` // 版本 + Description string `json:"description"` // 描述 + Model DataResourceConfig `json:"model"` // 模型信息 + SubModel *DataResourceConfig `json:"sub_model"` // 增量模型信息（可选） + Image ImageInfo `json:"image"` // 镜像信息 + CodeConfig CodeConfig `json:"code_config"` // 代码配置 + ResourceType string `json:"resource_type"` // 资源类型 + Resource ResourceConfig `json:"resource"` // 资源配置 + Command string `json:"command"` // 启动命令 +} + +// ImageInfo 镜像信息 +type ImageInfo struct { + ImageID int `json:"imageID"` + Name string `json:"name"` + CreateTime string `json:"createTime"` + ClusterImages []ClusterImage `json:"clusterImages"` +} + +// ClusterImage 集群镜像信息 +type ClusterImage struct { + ImageID int `json:"imageID"` + ClusterID string `json:"clusterID"` + OriginImageType string `json:"originImageType"` + OriginImageID string `json:"originImageID"` + OriginImageName string `json:"originImageName"` + Cards []Card `json:"cards"` +} + +// Card 卡片信息 +type Card struct { + OriginImageID string `json:"originImageID"` + Card string `json:"card"` +} diff --git a/pkg/models/task.go b/pkg/models/task.go index c00d0d0..b524772 100644 --- a/pkg/models/task.go +++ b/pkg/models/task.go @@ -275,3 +275,64 @@ type InferenceTaskDetailData struct { Status string `json:"status"` CreatedTime string `json:"createdTime"` } + +// InferenceSubmitTaskResponse 推理任务提交响应 +type InferenceSubmitTaskResponse struct { + Code int `json:"code"` // 状态码 + Msg string `json:"msg"` // 消息 + Data InferenceSubmitTaskResponseData `json:"data"` // 响应数据 +} + +// InferenceSubmitTaskResponseData 推理任务提交响应数据 +type InferenceSubmitTaskResponseData struct { + TaskInfo InferenceTaskInfo `json:"taskInfo"` // 任务信息 + ResultInfo InferenceResultInfo `json:"resultInfo"` // 结果信息 +} + +// InferenceTaskInfo 推理任务信息 +type InferenceTaskInfo struct { + UserID int `json:"userID"` // 用户ID + JobSetInfo InferenceJobSetInfo `json:"jobSetInfo"` // 任务集信息 +} + +// InferenceJobSetInfo 推理任务集信息 +type InferenceJobSetInfo struct { + Jobs []InferenceJob `json:"jobs"` // 任务列表 +} + +// InferenceJob 推理任务 +type InferenceJob struct { + LocalJobID string `json:"localJobID"` // 本地任务ID + Name string `json:"name"` // 任务名称 + Description string `json:"description"` // 任务描述 + Type string `json:"type"` // 任务类型 + Files *InferenceJobFiles `json:"files"` // 文件信息 + JobResources *InferenceJobResources `json:"jobResources"` // 资源信息 +} + +// InferenceJobFiles 推理任务文件 +type InferenceJobFiles struct { + Model FileBinding `json:"model"` // 模型绑定 + SubModel *FileBinding `json:"sub_model,omitempty"` // 增量模型绑定（可选） + Image ImageBinding `json:"image"` // 镜像绑定 +} + +// InferenceJobResources 推理任务资源 +type InferenceJobResources struct { + ScheduleStrategy string `json:"scheduleStrategy"` // 调度策略 + Clusters []InferenceCluster `json:"clusters"` // 集群列表 +} + +// InferenceCluster 推理集群信息 +type InferenceCluster struct { + ClusterID string `json:"clusterID"` // 集群ID + Runtime Runtime `json:"runtime"` // 运行时配置 + Code CodeInfo `json:"code"` // 代码信息 + Resources []ResourcesItem `json:"resources"` // 资源列表 +} + +// InferenceResultInfo 推理结果信息 +type InferenceResultInfo struct { + LocalJobID string `json:"localJobID"` // 本地任务ID + JobSetID string `json:"jobSetID"` // 任务集ID +} diff --git a/pkg/service/common/task_builder.go b/pkg/service/common/task_builder.go index bb79074..67e41b9 100644 --- a/pkg/service/common/task_builder.go +++ b/pkg/service/common/task_builder.go @@ -86,3 +86,68 @@ func BuildSubmitTaskRequest(ctx context.Context, authData *models.AuthData, conf }, } } + +// BuildInferenceSubmitTaskRequest 构建推理任务提交请求 +func BuildInferenceSubmitTaskRequest(ctx context.Context, authData *models.AuthData, + codeBindingID, modelBindingID, subModelBindingID int, imageID int, clusterID string, + resourceType string, resources []models.ResourcesItem, command string, description string) models.InferenceSubmitTaskResponseData { + + // 构建文件信息 + files := models.InferenceJobFiles{ + Model: models.FileBinding{ + Type: "Binding", + BindingID: modelBindingID, + }, + Image: models.ImageBinding{ + Type: "Image", + ImageID: imageID, + }, + } + + // 如果有增量模型，添加到文件信息中 + if subModelBindingID > 0 { + files.SubModel = &models.FileBinding{ + Type: "Binding", + BindingID: subModelBindingID, + } + } + + // 构建任务 + job := models.InferenceJob{ + LocalJobID: models.MainTaskID, + Name: "inference-task" + preparation.GenerateUniqueID(), + Description: description, + Type: "PCM_Inference", + Files: &files, + JobResources: &models.InferenceJobResources{ + ScheduleStrategy: "dataLocality", + Clusters: []models.InferenceCluster{ + { + ClusterID: clusterID, + Runtime: models.Runtime{ + Envs: make(map[string]string), + Params: make(map[string]string), + }, + Code: models.CodeInfo{ + Type: "Binding", + BindingID: codeBindingID, + }, + Resources: resources, + }, + }, + }, + } + + return models.InferenceSubmitTaskResponseData{ + TaskInfo: models.InferenceTaskInfo{ + UserID: authData.JsmUserInfo.Data.UserID, + JobSetInfo: models.InferenceJobSetInfo{ + Jobs: []models.InferenceJob{job}, + }, + }, + ResultInfo: models.InferenceResultInfo{ + LocalJobID: models.MainTaskID, + JobSetID: "", // 这个会在提交后由服务器返回 + }, + } +} diff --git a/pkg/service/inference/inference_service.go b/pkg/service/inference/inference_service.go index 5af183c..fea259d 100644 --- a/pkg/service/inference/inference_service.go +++ b/pkg/service/inference/inference_service.go @@ -52,3 +52,89 @@ func (s *inferenceService) StopTask(ctx context.Context, authData *models.AuthDa // 这里需要实现停止任务的逻辑 return fmt.Errorf("StopTask not implemented") } + +// SubmitInferenceTask 提交推理任务（包含准备步骤） +func (s *inferenceService) SubmitInferenceTask(ctx context.Context, authData *models.AuthData, + request *models.InferenceSubmitTaskRequest, clusterID string) (*models.InferenceSubmitTaskResponse, error) { + + // 1. 准备代码 + fmt.Println("开始准备代码") + codeConfig := models.CodeConfig{ + GitUrl: request.CodeConfig.GitUrl, + GitBranch: request.CodeConfig.GitBranch, + MountPath: request.CodeConfig.MountPath, + } + runConfig := &models.RunConfig{ + CodeConfig: codeConfig, + Image: request.Image.ImageID, + Command: request.Command, + Resource: request.Resource, + } + codeID, err := s.preparationService.PrepareCode(ctx, authData, runConfig, clusterID) + if err != nil { + return nil, fmt.Errorf("准备代码失败: %w", err) + } + fmt.Println("代码准备成功，ID:", codeID) + + // 2. 准备模型 + fmt.Println("开始准备模型") + // 如果 MountPath 为空，使用 Path 作为 MountPath + modelConfig := request.Model + if modelConfig.MountPath == "" { + modelConfig.MountPath = modelConfig.Path + } + modelID, err := s.preparationService.PrepareModel(ctx, authData, modelConfig, clusterID) + if err != nil { + return nil, fmt.Errorf("准备模型失败: %w", err) + } + fmt.Println("模型准备成功，ID:", modelID) + + // 3. 准备增量模型（如果有） + var subModelID int + if request.SubModel != nil { + fmt.Println("开始准备增量模型") + // 如果 MountPath 为空，使用 Path 作为 MountPath + subModelConfig := *request.SubModel + if subModelConfig.MountPath == "" { + subModelConfig.MountPath = subModelConfig.Path + } + subModelID, err = s.preparationService.PrepareModel(ctx, authData, subModelConfig, clusterID) + if err != nil { + return nil, fmt.Errorf("准备增量模型失败: %w", err) + } + fmt.Println("增量模型准备成功，ID:", subModelID) + } + + // 4. 构建推理任务响应数据（用于返回） + taskData := common.BuildInferenceSubmitTaskRequest( + ctx, authData, + codeID, modelID, subModelID, + request.Image.ImageID, + clusterID, + request.ResourceType, + request.Resource.Resources, + request.Command, + request.Description, + ) + + // 5. 使用 taskSubmitter 提交任务 + // 注意：taskSubmitter 会使用 BuildSubmitTaskRequest，它构建的是训练任务格式 + // 对于推理任务，我们需要传入 BindDatasetID=0，这样 Files 中的 Dataset 会是空的 + jobSetID, err := s.taskSubmitter.SubmitTask(ctx, authData, runConfig, clusterID, &models.BindResultSet{ + BindCodeID: codeID, + BindModelID: modelID, + BindDatasetID: 0, // 推理任务不需要数据集 + }) + if err != nil { + return nil, fmt.Errorf("提交任务失败: %w", err) + } + + // 更新结果信息 + taskData.ResultInfo.JobSetID = jobSetID + + return &models.InferenceSubmitTaskResponse{ + Code: 200, + Msg: "", + Data: taskData, + }, nil +} diff --git a/pkg/service/interface.go b/pkg/service/interface.go index 2bd336c..d5b6cf1 100644 --- a/pkg/service/interface.go +++ b/pkg/service/interface.go @@ -30,6 +30,8 @@ type TaskService interface { type InferenceService interface { SubmitTask(ctx context.Context, authData *models.AuthData, config *models.RunConfig, clusterID string, bindResultSet *models.BindResultSet) (string, error) + SubmitInferenceTask(ctx context.Context, authData *models.AuthData, + request *models.InferenceSubmitTaskRequest, clusterID string) (*models.InferenceSubmitTaskResponse, error) GetTaskStatus(ctx context.Context, authData *models.AuthData, jobSetID string) (*models.InferenceTaskDetailResponse, error) StopTask(ctx context.Context, authData *models.AuthData, jobSetID string) error }