Merge branch 'refs/heads/feat/workflow-parallel-support' into deploy/dev

.github/workflows/build-push.yml

@@ -125,7 +125,6 @@ jobs:
         with:
           images: ${{ env[matrix.image_name_env] }}
           tags: |
-            type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/') }}
             type=ref,event=branch
             type=sha,enable=true,priority=100,prefix=,suffix=,format=long
             type=raw,value=${{ github.ref_name }},enable=${{ startsWith(github.ref, 'refs/tags/') }}

api/configs/packaging/__init__.py

@@ -9,7 +9,7 @@ class PackagingInfo(BaseSettings):
 
     CURRENT_VERSION: str = Field(
         description="Dify version",
-        default="0.7.3",
+        default="0.8.0-beta1",
     )
 
     COMMIT_SHA: str = Field(

api/core/workflow/graph_engine/entities/graph.py

@@ -416,13 +416,33 @@ def _recursively_add_parallels(
                         parallel.end_to_node_id = outside_parallel_target_node_ids.pop()
 
         for graph_edge in target_node_edges:
+            current_parallel = None
+            if parallel:
+                current_parallel = parallel
+            elif parent_parallel:
+                if not parent_parallel.end_to_node_id or (parent_parallel.end_to_node_id and graph_edge.target_node_id != parent_parallel.end_to_node_id):
+                    current_parallel = parent_parallel
+                else:
+                    # fetch parent parallel's parent parallel
+                    parent_parallel_parent_parallel_id = parent_parallel.parent_parallel_id
+                    if parent_parallel_parent_parallel_id:
+                        parent_parallel_parent_parallel = parallel_mapping.get(parent_parallel_parent_parallel_id)
+                        if (
+                            parent_parallel_parent_parallel 
+                            and (
+                                not parent_parallel_parent_parallel.end_to_node_id
+                                 or (parent_parallel_parent_parallel.end_to_node_id and graph_edge.target_node_id != parent_parallel_parent_parallel.end_to_node_id)
+                            )
+                        ):
+                            current_parallel = parent_parallel_parent_parallel
+
             cls._recursively_add_parallels(
                 edge_mapping=edge_mapping,
                 reverse_edge_mapping=reverse_edge_mapping,
                 start_node_id=graph_edge.target_node_id,
                 parallel_mapping=parallel_mapping,
                 node_parallel_mapping=node_parallel_mapping,
-                parent_parallel=parallel if parallel else parent_parallel
+                parent_parallel=current_parallel
             )
 
     @classmethod

api/services/app_dsl_service.py

@@ -13,8 +13,9 @@
 
 logger = logging.getLogger(__name__)
 
-current_dsl_version = "0.1.1"
+current_dsl_version = "0.1.2"
 dsl_to_dify_version_mapping: dict[str, str] = {
+    "0.1.2": "0.8.0",
     "0.1.1": "0.6.0",  # dsl version -> from dify version
 }
 

docker-legacy/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3'
 services:
   # API service
   api:
-    image: langgenius/dify-api:0.7.3
+    image: langgenius/dify-api:0.8.0-beta1
     restart: always
     environment:
       # Startup mode, 'api' starts the API server.
@@ -229,7 +229,7 @@ services:
   # worker service
   # The Celery worker for processing the queue.
   worker:
-    image: langgenius/dify-api:0.7.3
+    image: langgenius/dify-api:0.8.0-beta1
     restart: always
     environment:
       CONSOLE_WEB_URL: ''
@@ -400,7 +400,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:0.7.3
+    image: langgenius/dify-web:0.8.0-beta1
     restart: always
     environment:
       # The base URL of console application api server, refers to the Console base URL of WEB service if console domain is

docker/docker-compose.yaml

@@ -191,7 +191,7 @@ x-shared-env: &shared-api-worker-env
 services:
   # API service
   api:
-    image: langgenius/dify-api:0.7.3
+    image: langgenius/dify-api:0.8.0-beta1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -211,7 +211,7 @@ services:
   # worker service
   # The Celery worker for processing the queue.
   worker:
-    image: langgenius/dify-api:0.7.3
+    image: langgenius/dify-api:0.8.0-beta1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -230,7 +230,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:0.7.3
+    image: langgenius/dify-web:0.8.0-beta1
     restart: always
     environment:
       CONSOLE_API_URL: ${CONSOLE_API_URL:-}

web/app/components/workflow/hooks/use-workflow-run.ts

@@ -141,7 +141,6 @@ export const useWorkflowRun = () => {
       tracing: [],
       resultText: '',
     })
-    let iterationLength = 0
 
     let ttsUrl = ''
     let ttsIsPublic = false
@@ -250,13 +249,13 @@ export const useWorkflowRun = () => {
             transform,
           } = store.getState()
           const nodes = getNodes()
-          const nodeParentId = nodes.find(node => node.id === data.node_id)!.parentId
-          if (nodeParentId) {
+          const node = nodes.find(node => node.id === data.node_id)
+          if (node?.parentId) {
             setWorkflowRunningData(produce(workflowRunningData!, (draft) => {
               const tracing = draft.tracing!
-              const iterations = tracing[tracing.length - 1]
-              const currIteration = iterations.details![iterations.details!.length - 1]
-              currIteration.push({
+              const iterations = tracing.find(trace => trace.node_id === node?.parentId)
+              const currIteration = iterations?.details![node.data.iteration_index] || iterations?.details![iterations.details!.length - 1]
+              currIteration?.push({
                 ...data,
                 status: NodeRunningStatus.Running,
               } as any)
@@ -316,7 +315,7 @@ export const useWorkflowRun = () => {
           if (nodeParentId) {
             setWorkflowRunningData(produce(workflowRunningData!, (draft) => {
               const tracing = draft.tracing!
-              const iterations = tracing[tracing.length - 1] // the iteration node
+              const iterations = tracing.find(trace => trace.node_id === nodeParentId) // the iteration node
 
               if (iterations && iterations.details) {
                 const iterationIndex = data.execution_metadata?.iteration_index || 0
@@ -325,7 +324,7 @@ export const useWorkflowRun = () => {
 
                 const currIteration = iterations.details[iterationIndex]
                 const nodeIndex = currIteration.findIndex(node =>
-                  node.node_id === data.node_id,
+                  node.node_id === data.node_id && node.execution_metadata?.parallel_id === data.execution_metadata?.parallel_id,
                 )
                 if (data.status === NodeRunningStatus.Succeeded) {
                   if (nodeIndex !== -1) {
@@ -391,7 +390,6 @@ export const useWorkflowRun = () => {
               details: [],
             } as any)
           }))
-          iterationLength = data.metadata.iterator_length
 
           const {
             setViewport,
@@ -437,13 +435,13 @@ export const useWorkflowRun = () => {
           } = store.getState()
 
           setWorkflowRunningData(produce(workflowRunningData!, (draft) => {
-            const iteration = draft.tracing![draft.tracing!.length - 1]
-            if (iteration.details!.length >= iterationLength)
-              return
-
-            iteration.details!.push([])
+            const iteration = draft.tracing!.find(trace => trace.node_id === data.node_id)
+            if (iteration) {
+              if (iteration.details!.length >= iteration.metadata.iterator_length!)
+                return
+            }
+            iteration?.details!.push([])
           }))
-
           const nodes = getNodes()
           const newNodes = produce(nodes, (draft) => {
             const currentNode = draft.find(node => node.id === data.node_id)!
@@ -469,11 +467,13 @@ export const useWorkflowRun = () => {
           const nodes = getNodes()
           setWorkflowRunningData(produce(workflowRunningData!, (draft) => {
             const tracing = draft.tracing!
-            tracing[tracing.length - 1] = {
-              ...tracing[tracing.length - 1],
-              ...data,
-              status: NodeRunningStatus.Succeeded,
-            } as any
+            const currIterationNode = tracing.find(trace => trace.node_id === data.node_id)
+            if (currIterationNode) {
+              Object.assign(currIterationNode, {
+                ...data,
+                status: NodeRunningStatus.Succeeded,
+              })
+            }
           }))
 
           const newNodes = produce(nodes, (draft) => {

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "dify-web",
-  "version": "0.7.3",
+  "version": "0.8.0-beta1",
   "private": true,
   "engines": {
     "node": ">=18.17.0"