fix(task): 修复任务处理中的参数传递和数据解析问题

This commit is contained in:
WangLiZhao
2026-07-01 18:28:00 +08:00
parent 84421b4ee5
commit 7e849b1c0f
+21 -11
View File
@@ -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)