feat: ModelLambda 段级续跑——只生成缺失段、失败重试1次、成功落库、按序合并

2026-08-25 10:20:47 +08:00
parent 2d7f6888a6
commit bae916e019
1 changed files with 58 additions and 10 deletions
@@ -6,6 +6,7 @@ import (
 	"ai-agent/workflow/consts/model"
 	"ai-agent/workflow/consts/node"
 	"ai-agent/workflow/consts/public"
+	flowDao "ai-agent/workflow/dao/flow"
 	nodeDao "ai-agent/workflow/dao/node"
 	sessionDao "ai-agent/workflow/dao/session"
 	flowDto "ai-agent/workflow/model/dto/flow"
@@ -75,36 +76,83 @@ func ModelLambda(ctx context.Context, input any) (any, error) {
 	var totalTokens int64
 	var totalCost float64
 	if len(paramsList) > 1 {
-		// 异步批量执行：并发请求模型，等待全部返回后再继续，避免下游读到空结果
+		// 段级续跑仅在"多段 + 视频模型"启用；非视频分段（批量文本等）走原逻辑零影响
+		segVideo := isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId)
+
+		// 续跑（!ForceNewRun）时读取该节点已成功段；全新执行不查（BuildExecution 已清旧段），saved 为 nil → 全量重生成
+		var saved map[int]entity.SegmentRef
+		if !nodeInput.Global.ForceNewRun && segVideo {
+			saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		idxList, needGen := planSegmentResume(paramsList, saved)
+
 		results := make([][]map[string]any, len(paramsList))
 		tokenRes := make([]*gateway.ModelCallRes, len(paramsList))
 		errs := make([]error, len(paramsList))
 		isInference := make([]bool, len(paramsList))
 		var wg sync.WaitGroup
 		for i, params := range paramsList {
+			if !needGen[i] {
+				continue
+			}
 			wg.Add(1)
 			go func(i int, params map[string]any) {
 				defer wg.Done()
-				results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
+				// 视频段每段失败自动重试 1 次（共 2 次尝试）；非视频保持单次调用
+				for attempt := 0; attempt < segmentGenerateMaxAttempts; attempt++ {
+					results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
+					if errs[i] == nil || !segVideo {
+						break
+					}
+				}
 			}(i, params)
 		}
 		wg.Wait()
+
+		// 每段成功立即落库：即使后续某段失败导致节点失败，成功段也保留供下次续跑复用
+		if segVideo {
+			for i := range results {
+				if !needGen[i] || errs[i] != nil {
+					continue
+				}
+				for _, rec := range results[i] {
+					key := media.FindVideoKey(rec)
+					url := media.FindVideoURL(ctx, rec)
+					if key == "" || url == "" {
+						continue
+					}
+					if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i], key, url); err != nil {
+						return nil, err
+					}
+				}
+			}
+		}
+
+		// 仍有失败段 → 节点失败（成功段已落库，供下次 reExecute 复用）
 		for i := range results {
-			if errs[i] != nil {
+			if needGen[i] && errs[i] != nil {
 				return nil, errs[i]
 			}
-			if tokenRes[i] != nil {
+			if needGen[i] && tokenRes[i] != nil {
 				totalTokens += tokenRes[i].TotalTokens
 				totalCost += tokenRes[i].Cost
 			}
 		}
-		// 推理模型（分批同一模型，isInference 各批一致）：分批结果拼到单个字段（单条输出记录）；
-		// 非推理模型保持逐条展平
-		if isInference[0] {
-			outputRes = mergeInferenceBatchResults(results)
+
+		if segVideo {
+			// 复用段 + 新生段按段序号升序合并，concat 按列表顺序拼接 → 顺序保证
+			outputRes = mergeSegmentOutputs(idxList, needGen, results, saved)
 		} else {
-			for _, res := range results {
-				outputRes = append(outputRes, res...)
+			if isInference[0] {
+				outputRes = mergeInferenceBatchResults(results)
+			} else {
+				for _, res := range results {
+					outputRes = append(outputRes, res...)
+				}
 			}
 		}
 	} else {