diff --git a/workflow/model/dto/session/exec_workflow_dto.go b/workflow/model/dto/session/exec_workflow_dto.go index 98557fc..ab8c87a 100644 --- a/workflow/model/dto/session/exec_workflow_dto.go +++ b/workflow/model/dto/session/exec_workflow_dto.go @@ -20,12 +20,13 @@ type DeleteExecWorkflowReq struct { } type UpdateWorkflowReq struct { - Id int64 `json:"id" v:"required#工作流执行记录ID不能为空"` - NodeGroupId string `json:"nodeGroupId" description:"节点组ID"` - Status flow.FlowExecutionStatus `json:"status" description:"状态:1-运行中,2-成功,3-失败"` - Duration int64 `json:"duration" description:"执行时长(秒)"` - TotalTokens int `json:"totalTokens" description:"总token消耗"` - TotalFee float64 `json:"totalFee" description:"总费用"` - ErrorMessage string `json:"errorMessage" description:"错误信息(友好提示)"` - Error string `json:"error" description:"错误明细(原始错误)"` + Id int64 `json:"id" v:"required#工作流执行记录ID不能为空"` + NodeGroupId string `json:"nodeGroupId" description:"节点组ID"` + Status flow.FlowExecutionStatus `json:"status" description:"状态:1-运行中,2-成功,3-失败"` + Duration int64 `json:"duration" description:"执行时长(秒)"` + TotalTokens int `json:"totalTokens" description:"总token消耗"` + TotalFee float64 `json:"totalFee" description:"总费用"` + ErrorMessage string `json:"errorMessage" description:"错误信息(友好提示)"` + Error string `json:"error" description:"错误明细(原始错误)"` + RequestParams *entity.FlowInfo `json:"requestParams" description:"请求参数"` } diff --git a/workflow/service/flow/flow_graph_builder.go b/workflow/service/flow/flow_graph_builder.go index c0be8fe..9910901 100644 --- a/workflow/service/flow/flow_graph_builder.go +++ b/workflow/service/flow/flow_graph_builder.go @@ -143,7 +143,7 @@ func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.Flo // BuildGraphFromFlowContent 根据前端保存的工作流JSON,自动构建执行图并编译 func BuildGraphFromFlowContent(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, compose.Runnable[any, any], error) { nodeList, graph := BuildGraph(ctx, flowContent) - compile, err := graph.Compile(ctx, compose.WithGraphName("auto_build_workflow"), compose.WithCheckPointStore(NewDbCheckPointStore())) + compile, err := graph.Compile(ctx, compose.WithGraphName("auto_build_workflow"), compose.WithCheckPointStore(NewDbCheckPointStore()), compose.WithNodeTriggerMode(compose.AllPredecessor)) return nodeList, compile, err } @@ -225,7 +225,7 @@ func registerNodeToGraph(graph *compose.Graph[any, any], flowNode entity.FlowNod case node.NodeTypeForm: _ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(FormLambda))) case node.NodeTypeDataMerge: - _ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(DataMergeLambda)), compose.WithGraphCompileOptions(compose.WithNodeTriggerMode(compose.AllPredecessor))) + _ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(DataMergeLambda))) case node.NodeTypeSubFlow: _ = graph.AddLambdaNode(flowNode.Id, compose.InvokableLambda(wrapLambda(SubFlowLambda))) case node.NodeTypeHttp: diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index 205ebca..7c5153b 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -268,19 +268,41 @@ func flowContentEqual(a, b *entity.FlowInfo) bool { return bytes.Equal(ab, bb) } -// execute 执行工作流(首次执行) +// execute 执行工作流(首次执行;同会话+同工作流最近一次执行为失败状态时复用该记录重新执行,不新建数据) func execute(ctx context.Context, conn *wsCommon.WsConnection, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { var nodeGroupId = uuid.NewString() - id, 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 { - return + + // 复用失败记录:查询会话+工作流最近一次执行,若为失败状态则复用同一条记录(更新状态+节点组+本次请求参数), + // 全新执行(forceNewRun,参数取本次请求),避免重跑新建数据;查询出错按无记录处理走新建 + lastExec, qErr := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, conn.SessionId, req.FlowId) + if qErr != nil { + glog.Errorf(ctx, "查询最近工作流执行记录失败: %v", qErr) + lastExec = nil } + if lastExec != nil && *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() { + id = lastExec.Id + _, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{ + Id: id, + NodeGroupId: nodeGroupId, + Status: flow.FlowExecutionStatusRunning.Code(), + RequestParams: req.FlowContent, + }) + if err != nil { + return + } + } else { + id, 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 { + return + } + } + _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ "id": id, }})