diff --git a/service/task/worker.go b/service/task/worker.go index 580391c..036b9cd 100644 --- a/service/task/worker.go +++ b/service/task/worker.go @@ -18,6 +18,7 @@ import ( "time" "gitea.redpowerfuture.com/red-future/common/beans" + "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/encoding/gjson" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/util/gconv" @@ -123,7 +124,7 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa // 3) 处理提示词相关数据解析涵盖重试 // ============================================ if req.BuildType == public.BuildTypePrompt { - mapped, err = w.parseAndRetry(ctx, mapped, task, model, maxRetry) + mapped, err = w.parseAndRetry(ctx, mapped, model, task, maxRetry) if err != nil { task.TextResult = mapped w.failTask(ctx, task, startTime, err.Error()) @@ -248,15 +249,28 @@ func NotifyAsyncResult(taskID string, result map[string]any, err error) { //} // parseAndRetry 解析模型返回结果,并重试 -func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, task *entity.ModelGatewayTask, model *entity.ModelGatewayModel, maxRetry int) (map[string]any, error) { +func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, model *entity.ModelGatewayModel, task *entity.ModelGatewayTask, maxRetry int) (map[string]any, error) { + // 获取构建模型的必填字段 + user, err := utils.GetUserInfo(ctx) + if err != nil { + return nil, err + } + buildModel, err := dao.ModelGatewayModels.Get(ctx, &entity.ModelGatewayModel{ + SQLBaseDO: beans.SQLBaseDO{TenantId: user.TenantId, Creator: user.UserName}, + ModelName: task.BuildModelName, + }) + if err != nil { + return nil, err + } + var lastErr error for attempt := 0; attempt <= maxRetry; attempt++ { if attempt > 0 { g.Log().Infof(ctx, "[执行任务][重试] JSON解析 第%d/%d次 taskId=%s", attempt, maxRetry, task.TaskID) } - // 解析 + 校验 - parsed, err := util.ParseAndValidate(body, model.RequiredFields) + // 解析 + 校验(用构建模型的 RequiredFields) + parsed, err := util.ParseAndValidate(body, buildModel.RequiredFields) if err == nil { return parsed, nil } @@ -279,7 +293,6 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta continue } - // 响应映射 var rawResp map[string]any if err := json.Unmarshal(rawData, &rawResp); err != nil { g.Log().Warningf(ctx, "[执行任务][Unmarshal失败] taskId=%s err=%v", task.TaskID, err) @@ -291,20 +304,17 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta continue } - // parseAndRetry 重试时 - if len(model.BillingConfig) > 0 { - requestData := task.BillingData[0] // 请求数据从第一个元素取 - + // 计费 + if len(model.BillingConfig) > 0 && len(task.BillingData) > 0 { + requestData := task.BillingData[0] retryData := make(map[string]any) for k, v := range requestData { retryData[k] = v } - responseData := util.ExtractResponseBilling(model.BillingConfig, mapped) for k, v := range responseData { retryData[k] = v } - billingResult := util.CalculateBilling(model.BillingConfig, retryData) if billingResult != nil { task.BillingData = append(task.BillingData, billingResult)