91 lines
2.8 KiB
Go
91 lines
2.8 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"model-gateway/dao"
|
|
"model-gateway/model/dto"
|
|
"model-gateway/service/httpclient"
|
|
modelUtils "model-gateway/service/utils"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/util/gconv"
|
|
)
|
|
|
|
var ModelTaskStart = &modelTaskStartService{}
|
|
|
|
type modelTaskStartService struct{}
|
|
|
|
// CreateTask 创建任务
|
|
func (s *modelTaskStartService) CreateTask(ctx context.Context, req *dto.CallModelTaskStartReq) (res *dto.ModelCallRes, err error) {
|
|
startTime := time.Now()
|
|
attempt := 0
|
|
|
|
id := req.Id
|
|
modelInfo := req.ModelInfo
|
|
newRequestParams := req.RequestParams
|
|
|
|
LOOP:
|
|
// 6) 模型请求
|
|
modelRespBody, err := httpclient.ModelHttpNormalRequest(ctx, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams)
|
|
if err != nil {
|
|
if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, "", err.Error(), "") {
|
|
attempt++
|
|
if waitErr := retryWait(ctx, attempt); waitErr != nil {
|
|
return nil, waitErr
|
|
}
|
|
goto LOOP
|
|
}
|
|
return nil, fmt.Errorf("模型请求失败: %v", err)
|
|
}
|
|
if modelRespBody == nil {
|
|
return nil, fmt.Errorf("模型返回参数是空")
|
|
}
|
|
// 7) 更新视频任务信息（统一字段路径 GetByPath 基于该对象读取）
|
|
var respObj map[string]any
|
|
if err = json.Unmarshal(modelRespBody, &respObj); err != nil {
|
|
return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
|
|
}
|
|
updateModelReq := dto.UpdateModelTaskStartReq{
|
|
Id: id,
|
|
OriginalResponseParams: respObj,
|
|
}
|
|
docMsg := new(dto.ModelCallRes)
|
|
docMsg.TaskId = id
|
|
// 按模型 ErrorMessageMapping 解析错误响应，无错误返回空串
|
|
var errCode string
|
|
if errCode, docMsg.ErrorMsg, err = parseModelError(modelRespBody, modelInfo.ErrorMessageMapping); err != nil {
|
|
return nil, fmt.Errorf("模型返回参数解析失败：%v", err)
|
|
}
|
|
if docMsg.ErrorMsg != "" {
|
|
if attempt < modelCallMaxRetries && shouldRetryWithMemory(ctx, modelInfo, errCode, docMsg.ErrorMsg, string(modelRespBody)) {
|
|
attempt++
|
|
if waitErr := retryWait(ctx, attempt); waitErr != nil {
|
|
return nil, waitErr
|
|
}
|
|
goto LOOP
|
|
}
|
|
updateModelReq.ErrorMsg = docMsg.ErrorMsg
|
|
}
|
|
if docMsg.ErrorMsg == "" {
|
|
taskIDPath := modelUtils.CleanFieldPath(modelInfo.AsyncTaskMapping.TaskId)
|
|
docMsg.Content = map[string]any{
|
|
"respBody": modelUtils.GetByPathValue(respObj, taskIDPath),
|
|
}
|
|
}
|
|
if !g.IsEmpty(docMsg.Content) {
|
|
updateModelReq.ResponseParams = docMsg.Content
|
|
updateModelReq.TaskId = gconv.String(docMsg.Content["respBody"])
|
|
}
|
|
updateModelReq.DurationSeconds = int64(time.Since(startTime).Seconds())
|
|
// 8) 更新模型视频任务信息
|
|
_, err = dao.ModelTaskStart.Update(ctx, &updateModelReq)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("更新模型视频任务信息失败: %v", err)
|
|
}
|
|
|
|
return docMsg, nil
|
|
}
|