From bae916e01939d32a8b648be6cce0f874f997df8f Mon Sep 17 00:00:00 2001
From: qhd <1766646056@qq.com>
Date: Tue, 25 Aug 2026 10:20:47 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20ModelLambda=20=E6=AE=B5=E7=BA=A7?=
 =?UTF-8?q?=E7=BB=AD=E8=B7=91=E2=80=94=E2=80=94=E5=8F=AA=E7=94=9F=E6=88=90?=
 =?UTF-8?q?=E7=BC=BA=E5=A4=B1=E6=AE=B5=E3=80=81=E5=A4=B1=E8=B4=A5=E9=87=8D?=
 =?UTF-8?q?=E8=AF=951=E6=AC=A1=E3=80=81=E6=88=90=E5=8A=9F=E8=90=BD?=
 =?UTF-8?q?=E5=BA=93=E3=80=81=E6=8C=89=E5=BA=8F=E5=90=88=E5=B9=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 workflow/service/flow/lambda_node.go | 68 ++++++++++++++++++++++++----
 1 file changed, 58 insertions(+), 10 deletions(-)

diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go
index a0a488e..9609c0f 100644
--- a/workflow/service/flow/lambda_node.go
+++ b/workflow/service/flow/lambda_node.go
@@ -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 {