From 75fb40274d3352d54260a802d4a0c063577f9c3e Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Mon, 24 Aug 2026 08:59:29 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E6=96=AD=E7=82=B9=E7=BB=AD=E8=B7=91=E6=97=B6=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E8=AE=B0=E5=BD=95=E5=A4=8D=E7=94=A8=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- workflow/service/flow/flow_ws_exec.go | 15 ++++++++++ workflow/service/flow/lambda_node.go | 4 ++- workflow/service/flow/lambda_value_source.go | 31 +++++++++++--------- 3 files changed, 35 insertions(+), 15 deletions(-) diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index db78c13..81c86d5 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -315,6 +315,7 @@ func flowContentEqual(a, b *entity.FlowInfo) bool { func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, status flow.FlowExecutionStatus, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { var nodeGroupId = uuid.NewString() if g.IsEmpty(execId) { + glog.Infof(ctx, "工作流全新执行execute，无历史记录") execId, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ SessionId: conn.SessionId, FlowId: req.FlowId, @@ -330,6 +331,7 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta // FlowExecutionStatus 是 *int8 别名，Code() 返回包级指针，直接 == 是地址比较恒为 false， // 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() { + glog.Infof(ctx, "工作流断点续跑execute，execId: %v", execId) _, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{ Id: execId, NodeGroupId: nodeGroupId, @@ -339,6 +341,19 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta if err != nil { return } + } else { + glog.Infof(ctx, "工作流全新执行execute，lastExec: %v", execId) + execId, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ + SessionId: conn.SessionId, + FlowId: req.FlowId, + NodeGroupId: nodeGroupId, + Status: flow.FlowExecutionStatusRunning.Code(), + RequestParams: req.FlowContent, + }) + if err != nil || g.IsEmpty(execId) { + glog.Errorf(ctx, "工作流执行记录创建失败: %v", err) + return + } } } diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go index df6100e..a0a488e 100644 --- a/workflow/service/flow/lambda_node.go +++ b/workflow/service/flow/lambda_node.go @@ -137,7 +137,9 @@ func ModelLambda(ctx context.Context, input any) (any, error) { // 4.5 视频模型节点返回多个视频时，自动调用视频合成工具（concat_videos）合并为单条； // 已显式配置 concat_videos 后置工具时跳过，避免重复合并 if nodeInput.Config.PostTool != media.ProcessorName && len(outputRes) > 1 && isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId) { + g.Log().Debugf(ctx, "modelId1:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes) outputRes, err = invokePostTool(ctx, media.ProcessorName, outputRes, map[string]any{"callback_url": "callback_url", "upload": true}) + g.Log().Debugf(ctx, "modelId2:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes) if err != nil { return nil, err } @@ -148,7 +150,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { return nil, err } } - g.Log().Debugf(ctx, "modelId:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes) + g.Log().Debugf(ctx, "modelId3:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes) nodeInput.Config.OutputResult = outputRes return nodeInput, nil } diff --git a/workflow/service/flow/lambda_value_source.go b/workflow/service/flow/lambda_value_source.go index 932f262..e0f90c9 100644 --- a/workflow/service/flow/lambda_value_source.go +++ b/workflow/service/flow/lambda_value_source.go @@ -186,21 +186,24 @@ func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, field string } } default: + // templates 是模型节点在前端配置的静态输出模板，不在 OutputResult 中，需单独取 if field == "templates" { - value = nodeConfig.Templates - } else { - for _, output := range nodeConfig.OutputResult { - // 模型节点输出记录是单 key 的字面量扁平 key（如 "choices.attrs[0].attrs.delta.attrs.content"）， - // gjson 会把 . 和 [0] 当结构路径解析，无法命中字面量 key，故先按字面量 key 直接取值； - // 未命中再回退 gjson 路径查询（兼容真正嵌套的输出结构）。 - if v, has := output[field]; has { - value = v - } else { - value = gjson.Get(gconv.String(output), field).Value() - } - if !g.IsEmpty(value) { - return value, gjson.Get(gconv.String(output), CleanFieldPath("refsName")).Value(), true - } + if !g.IsEmpty(nodeConfig.Templates) { + return nodeConfig.Templates, "", true + } + return nil, nil, false + } + for _, output := range nodeConfig.OutputResult { + // 模型节点输出记录是单 key 的字面量扁平 key（如 "choices.attrs[0].attrs.delta.attrs.content"）， + // gjson 会把 . 和 [0] 当结构路径解析，无法命中字面量 key，故先按字面量 key 直接取值； + // 未命中再回退 gjson 路径查询（兼容真正嵌套的输出结构）。 + if v, has := output[field]; has { + value = v + } else { + value = gjson.Get(gconv.String(output), field).Value() + } + if !g.IsEmpty(value) { + return value, gjson.Get(gconv.String(output), CleanFieldPath("refsName")).Value(), true } } }