diff --git a/workflow/service/flow/lambda_segment_resume.go b/workflow/service/flow/lambda_segment_resume.go new file mode 100644 index 0000000..b3b12ed --- /dev/null +++ b/workflow/service/flow/lambda_segment_resume.go @@ -0,0 +1,55 @@ +package flow + +import ( + "sort" + + "ai-agent/workflow/model/entity" + + "github.com/gogf/gf/v2/util/gconv" +) + +// segmentGenerateMaxAttempts 视频段生成最大尝试次数(失败自动重试 1 次,共 2 次尝试),参数化可调 +const segmentGenerateMaxAttempts = 2 + +// planSegmentResume 段级续跑决策:把 paramsList 各段映射到"是否需重新生成"。 +// savedMap 为该节点已成功段(段序号 → {key,url});段在表中缺失或地址为空则需重新生成。 +// 返回值与 paramsList 对齐。全新执行(savedMap 为 nil/空)时全部需生成。 +func planSegmentResume(paramsList []map[string]any, savedMap map[int]entity.SegmentRef) (idxList []int, needGen []bool) { + idxList = make([]int, len(paramsList)) + needGen = make([]bool, len(paramsList)) + for i, params := range paramsList { + idx := gconv.Int(params["segment_index"]) + idxList[i] = idx + ref, ok := savedMap[idx] + needGen[i] = !ok || ref.URL == "" + } + return +} + +// mergeSegmentOutputs 把复用段与新生段按段序号升序合并为 concat 输入列表(列表顺序即拼接顺序)。 +// 复用段重建 {key:url} 记录(key 保持模型原输出字段,避免下游引用失配);新生段沿用模型原输出。 +// 任一段既无复用又无生成结果(生成空)则跳过——与现有"空段贡献空"行为一致,最终由 concat 校验兜底。 +func mergeSegmentOutputs(idxList []int, needGen []bool, newRes [][]map[string]any, savedMap map[int]entity.SegmentRef) []map[string]any { + type segOutput struct { + idx int + recs []map[string]any + } + out := make([]segOutput, 0, len(idxList)) + for i, idx := range idxList { + if needGen[i] { + if len(newRes[i]) > 0 { + out = append(out, segOutput{idx: idx, recs: newRes[i]}) + } + continue + } + if ref, ok := savedMap[idx]; ok && ref.URL != "" { + out = append(out, segOutput{idx: idx, recs: []map[string]any{{ref.Key: ref.URL}}}) + } + } + sort.SliceStable(out, func(a, b int) bool { return out[a].idx < out[b].idx }) + var merged []map[string]any + for _, o := range out { + merged = append(merged, o.recs...) + } + return merged +} \ No newline at end of file diff --git a/workflow/service/flow/lambda_segment_resume_test.go b/workflow/service/flow/lambda_segment_resume_test.go new file mode 100644 index 0000000..15ffacb --- /dev/null +++ b/workflow/service/flow/lambda_segment_resume_test.go @@ -0,0 +1,102 @@ +package flow + +import ( + "testing" + + "ai-agent/workflow/model/entity" +) + +// planSegmentResume:段序号从 params 的 segment_index 读出;已成功段复用,缺失/空地址需生成 +func TestPlanSegmentResume(t *testing.T) { + paramsList := []map[string]any{ + {"segment_index": 1}, {"segment_index": 2}, {"segment_index": 3}, {"segment_index": 4}, {"segment_index": 5}, + } + saved := map[int]entity.SegmentRef{ + 1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"}, + 3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"}, + } + idxList, needGen := planSegmentResume(paramsList, saved) + if len(idxList) != 5 || len(needGen) != 5 { + t.Fatalf("长度不符: idxList=%v needGen=%v", idxList, needGen) + } + want := []bool{false, false, false, true, false} // 仅段4需重新生成 + for i := range want { + if needGen[i] != want[i] { + t.Fatalf("needGen[%d]=%v 期望 %v", i, needGen[i], want[i]) + } + if idxList[i] != i+1 { + t.Fatalf("idxList[%d]=%d 期望 %d", i, idxList[i], i+1) + } + } +} + +// planSegmentResume:无已成功段(全新执行)→ 全部需生成 +func TestPlanSegmentResumeEmptySaved(t *testing.T) { + paramsList := []map[string]any{{"segment_index": 0}, {"segment_index": 1}} + _, needGen := planSegmentResume(paramsList, nil) + for i := range needGen { + if !needGen[i] { + t.Fatalf("needGen[%d] 期望 true(无复用)", i) + } + } +} + +// mergeSegmentOutputs:复用段 + 新生段按段序号升序,顺序即拼接顺序 +func TestMergeSegmentOutputs(t *testing.T) { + idxList := []int{1, 2, 3, 4, 5} + needGen := []bool{false, false, false, true, false} + newRes := make([][]map[string]any, 5) + newRes[3] = []map[string]any{{"video_url": "new4"}} + saved := map[int]entity.SegmentRef{ + 1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"}, + 3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"}, + } + got := mergeSegmentOutputs(idxList, needGen, newRes, saved) + if len(got) != 5 { + t.Fatalf("合并后应有 5 段,实际 %d: %v", len(got), got) + } + wantURLs := []string{"u1", "u2", "u3", "new4", "u5"} + for i, rec := range got { + if rec["video_url"] != wantURLs[i] { + t.Fatalf("第 %d 段=%v 期望 %s", i, rec, wantURLs[i]) + } + } +} + +// mergeSegmentOutputs:复用段 key 保持模型原字段,重建记录与新生记录 key 不一致也不影响顺序 +func TestMergeSegmentOutputsMixedKeys(t *testing.T) { + idxList := []int{1, 2, 3} + needGen := []bool{false, true, false} + newRes := make([][]map[string]any, 3) + newRes[1] = []map[string]any{{"video_oss_url": "new2"}} + saved := map[int]entity.SegmentRef{ + 1: {Key: "video_url", URL: "u1"}, 3: {Key: "file_url", URL: "u3"}, + } + got := mergeSegmentOutputs(idxList, needGen, newRes, saved) + if len(got) != 3 { + t.Fatalf("合并后应有 3 段,实际 %d: %v", len(got), got) + } + if got[0]["video_url"] != "u1" || got[1]["video_oss_url"] != "new2" || got[2]["file_url"] != "u3" { + t.Fatalf("合并结果不符: %v", got) + } +} + +// mergeSegmentOutputs:paramsList 乱序时仍按段序号升序输出(顺序保证) +func TestMergeSegmentOutputsUnorderedParams(t *testing.T) { + idxList := []int{3, 1, 2} + needGen := []bool{true, false, true} + newRes := make([][]map[string]any, 3) + newRes[0] = []map[string]any{{"video_url": "new3"}} + newRes[2] = []map[string]any{{"video_url": "new2"}} + saved := map[int]entity.SegmentRef{1: {Key: "video_url", URL: "u1"}} + got := mergeSegmentOutputs(idxList, needGen, newRes, saved) + if len(got) != 3 { + t.Fatalf("合并后应有 3 段,实际 %d: %v", len(got), got) + } + wantURLs := []string{"u1", "new2", "new3"} + for i, rec := range got { + if rec["video_url"] != wantURLs[i] { + t.Fatalf("第 %d 段=%v 期望 %s(应段序升序)", i, rec, wantURLs[i]) + } + } +} diff --git a/workflow/service/flow/processor/builtin/media/media.go b/workflow/service/flow/processor/builtin/media/media.go index e794fb6..b4b1844 100644 --- a/workflow/service/flow/processor/builtin/media/media.go +++ b/workflow/service/flow/processor/builtin/media/media.go @@ -301,3 +301,13 @@ func normalizeVideoURL(ctx context.Context, url string) string { } return prefix + url } + +// FindVideoKey 返回视频 URL 所在字段的 key(规则同 findVideoKey),供工作流段级续跑重建输出记录保持 key 一致 +func FindVideoKey(params map[string]any) string { + return findVideoKey(params) +} + +// FindVideoURL 从模型返回参数中提取视频 URL(规则同 findVideoURL),供段级续跑落库 +func FindVideoURL(ctx context.Context, params map[string]any) string { + return findVideoURL(ctx, params) +}