Feat/iteration single run time (langgenius#10512)

Author: Nov1c444

api/core/app/apps/workflow_app_runner.py

@@ -361,6 +361,7 @@ def _handle_event(self, workflow_entry: WorkflowEntry, event: GraphEngineEvent)
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     output=event.pre_iteration_output,
                     parallel_mode_run_id=event.parallel_mode_run_id,
+                    duration=event.duration,
                 )
             )
         elif isinstance(event, (IterationRunSucceededEvent | IterationRunFailedEvent)):

api/core/app/entities/queue_entities.py

@@ -111,6 +111,7 @@ class QueueIterationNextEvent(AppQueueEvent):
     """iteratoin run in parallel mode run id"""
     node_run_index: int
     output: Optional[Any] = None  # output for the current iteration
+    duration: Optional[float] = None
 
     @field_validator("output", mode="before")
     @classmethod
@@ -307,6 +308,8 @@ class QueueNodeSucceededEvent(AppQueueEvent):
     execution_metadata: Optional[dict[NodeRunMetadataKey, Any]] = None
 
     error: Optional[str] = None
+    """single iteration duration map"""
+    iteration_duration_map: Optional[dict[str, float]] = None
 
 
 class QueueNodeInIterationFailedEvent(AppQueueEvent):

api/core/app/entities/task_entities.py

@@ -434,6 +434,7 @@ class Data(BaseModel):
         parallel_id: Optional[str] = None
         parallel_start_node_id: Optional[str] = None
         parallel_mode_run_id: Optional[str] = None
+        duration: Optional[float] = None
 
     event: StreamEvent = StreamEvent.ITERATION_NEXT
     workflow_run_id: str

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -624,6 +624,7 @@ def _workflow_iteration_next_to_stream_response(
                 parallel_id=event.parallel_id,
                 parallel_start_node_id=event.parallel_start_node_id,
                 parallel_mode_run_id=event.parallel_mode_run_id,
+                duration=event.duration,
             ),
         )
 

api/core/workflow/entities/node_entities.py

@@ -24,6 +24,7 @@ class NodeRunMetadataKey(str, Enum):
     PARENT_PARALLEL_ID = "parent_parallel_id"
     PARENT_PARALLEL_START_NODE_ID = "parent_parallel_start_node_id"
     PARALLEL_MODE_RUN_ID = "parallel_mode_run_id"
+    ITERATION_DURATION_MAP = "iteration_duration_map"  # single iteration duration if iteration node runs
 
 
 class NodeRunResult(BaseModel):

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

@@ -148,6 +148,7 @@ class IterationRunStartedEvent(BaseIterationEvent):
 class IterationRunNextEvent(BaseIterationEvent):
     index: int = Field(..., description="index")
     pre_iteration_output: Optional[Any] = Field(None, description="pre iteration output")
+    duration: Optional[float] = Field(None, description="duration")
 
 
 class IterationRunSucceededEvent(BaseIterationEvent):
@@ -156,6 +157,7 @@ class IterationRunSucceededEvent(BaseIterationEvent):
     outputs: Optional[dict[str, Any]] = None
     metadata: Optional[dict[str, Any]] = None
     steps: int = 0
+    iteration_duration_map: Optional[dict[str, float]] = None
 
 
 class IterationRunFailedEvent(BaseIterationEvent):

api/core/workflow/nodes/iteration/iteration_node.py

@@ -156,6 +156,7 @@ def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
             index=0,
             pre_iteration_output=None,
         )
+        iter_run_map: dict[str, float] = {}
         outputs: list[Any] = [None] * len(iterator_list_value)
         try:
             if self.node_data.is_parallel:
@@ -175,6 +176,7 @@ def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
                         iteration_graph,
                         index,
                         item,
+                        iter_run_map,
                     )
                     future.add_done_callback(thread_pool.task_done_callback)
                     futures.append(future)
@@ -213,6 +215,7 @@ def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
                         start_at,
                         graph_engine,
                         iteration_graph,
+                        iter_run_map,
                     )
             if self.node_data.error_handle_mode == ErrorHandleMode.REMOVE_ABNORMAL_OUTPUT:
                 outputs = [output for output in outputs if output is not None]
@@ -230,7 +233,9 @@ def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
 
             yield RunCompletedEvent(
                 run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.SUCCEEDED, outputs={"output": jsonable_encoder(outputs)}
+                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                    outputs={"output": jsonable_encoder(outputs)},
+                    metadata={NodeRunMetadataKey.ITERATION_DURATION_MAP: iter_run_map},
                 )
             )
         except IterationNodeError as e:
@@ -356,15 +361,19 @@ def _run_single_iter(
         start_at: datetime,
         graph_engine: "GraphEngine",
         iteration_graph: Graph,
+        iter_run_map: dict[str, float],
         parallel_mode_run_id: Optional[str] = None,
     ) -> Generator[NodeEvent | InNodeEvent, None, None]:
         """
         run single iteration
         """
+        iter_start_at = datetime.now(timezone.utc).replace(tzinfo=None)
+
         try:
             rst = graph_engine.run()
             # get current iteration index
             current_index = variable_pool.get([self.node_id, "index"]).value
+            iteration_run_id = parallel_mode_run_id if parallel_mode_run_id is not None else f"{current_index}"
             next_index = int(current_index) + 1
 
             if current_index is None:
@@ -431,6 +440,8 @@ def _run_single_iter(
                             variable_pool.add([self.node_id, "index"], next_index)
                             if next_index < len(iterator_list_value):
                                 variable_pool.add([self.node_id, "item"], iterator_list_value[next_index])
+                            duration = (datetime.now(timezone.utc).replace(tzinfo=None) - iter_start_at).total_seconds()
+                            iter_run_map[iteration_run_id] = duration
                             yield IterationRunNextEvent(
                                 iteration_id=self.id,
                                 iteration_node_id=self.node_id,
@@ -439,6 +450,7 @@ def _run_single_iter(
                                 index=next_index,
                                 parallel_mode_run_id=parallel_mode_run_id,
                                 pre_iteration_output=None,
+                                duration=duration,
                             )
                             return
                         elif self.node_data.error_handle_mode == ErrorHandleMode.REMOVE_ABNORMAL_OUTPUT:
@@ -449,6 +461,8 @@ def _run_single_iter(
 
                             if next_index < len(iterator_list_value):
                                 variable_pool.add([self.node_id, "item"], iterator_list_value[next_index])
+                            duration = (datetime.now(timezone.utc).replace(tzinfo=None) - iter_start_at).total_seconds()
+                            iter_run_map[iteration_run_id] = duration
                             yield IterationRunNextEvent(
                                 iteration_id=self.id,
                                 iteration_node_id=self.node_id,
@@ -457,6 +471,7 @@ def _run_single_iter(
                                 index=next_index,
                                 parallel_mode_run_id=parallel_mode_run_id,
                                 pre_iteration_output=None,
+                                duration=duration,
                             )
                             return
                         elif self.node_data.error_handle_mode == ErrorHandleMode.TERMINATED:
@@ -485,6 +500,8 @@ def _run_single_iter(
 
             if next_index < len(iterator_list_value):
                 variable_pool.add([self.node_id, "item"], iterator_list_value[next_index])
+            duration = (datetime.now(timezone.utc).replace(tzinfo=None) - iter_start_at).total_seconds()
+            iter_run_map[iteration_run_id] = duration
             yield IterationRunNextEvent(
                 iteration_id=self.id,
                 iteration_node_id=self.node_id,
@@ -493,6 +510,7 @@ def _run_single_iter(
                 index=next_index,
                 parallel_mode_run_id=parallel_mode_run_id,
                 pre_iteration_output=jsonable_encoder(current_iteration_output) if current_iteration_output else None,
+                duration=duration,
             )
 
         except IterationNodeError as e:
@@ -528,6 +546,7 @@ def _run_single_iter_parallel(
         iteration_graph: Graph,
         index: int,
         item: Any,
+        iter_run_map: dict[str, float],
     ) -> Generator[NodeEvent | InNodeEvent, None, None]:
         """
         run single iteration in parallel mode
@@ -546,6 +565,7 @@ def _run_single_iter_parallel(
                 start_at=start_at,
                 graph_engine=graph_engine_copy,
                 iteration_graph=iteration_graph,
+                iter_run_map=iter_run_map,
                 parallel_mode_run_id=parallel_mode_run_id,
             ):
                 q.put(event)

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

@@ -445,6 +445,7 @@ export const useWorkflowRun = () => {
               ...data,
               status: NodeRunningStatus.Running,
               details: [],
+              iterDurationMap: {},
             } as any)
           }))
 
@@ -496,6 +497,8 @@ export const useWorkflowRun = () => {
           setWorkflowRunningData(produce(workflowRunningData!, (draft) => {
             const iteration = draft.tracing!.find(trace => trace.node_id === data.node_id)
             if (iteration) {
+              if (iteration.iterDurationMap && data.duration)
+                iteration.iterDurationMap[data.parallel_mode_run_id ?? `${data.index - 1}`] = data.duration
               if (iteration.details!.length >= iteration.metadata.iterator_length!)
                 return
             }

web/app/components/workflow/nodes/_base/node.tsx

@@ -193,7 +193,7 @@ const BaseNode: FC<BaseNodeProps> = ({
           {
             data._iterationLength && data._iterationIndex && data._runningStatus === NodeRunningStatus.Running && (
               <div className='mr-1.5 text-xs font-medium text-primary-600'>
-                {data._iterationIndex}/{data._iterationLength}
+                {data._iterationIndex > data._iterationLength ? data._iterationLength : data._iterationIndex}/{data._iterationLength}
               </div>
             )
           }

web/app/components/workflow/panel/workflow-preview.tsx

@@ -28,7 +28,7 @@ import IterationResultPanel from '../run/iteration-result-panel'
 import InputsPanel from './inputs-panel'
 import cn from '@/utils/classnames'
 import Loading from '@/app/components/base/loading'
-import type { NodeTracing } from '@/types/workflow'
+import type { IterationDurationMap, NodeTracing } from '@/types/workflow'
 
 const WorkflowPreview = () => {
   const { t } = useTranslation()
@@ -53,12 +53,14 @@ const WorkflowPreview = () => {
   }, [workflowRunningData])
 
   const [iterationRunResult, setIterationRunResult] = useState<NodeTracing[][]>([])
+  const [iterDurationMap, setIterDurationMap] = useState<IterationDurationMap>({})
   const [isShowIterationDetail, {
     setTrue: doShowIterationDetail,
     setFalse: doHideIterationDetail,
   }] = useBoolean(false)
 
-  const handleShowIterationDetail = useCallback((detail: NodeTracing[][]) => {
+  const handleShowIterationDetail = useCallback((detail: NodeTracing[][], iterationDurationMap: IterationDurationMap) => {
+    setIterDurationMap(iterationDurationMap)
     setIterationRunResult(detail)
     doShowIterationDetail()
   }, [doShowIterationDetail])
@@ -72,6 +74,7 @@ const WorkflowPreview = () => {
           list={iterationRunResult}
           onHide={doHideIterationDetail}
           onBack={doHideIterationDetail}
+          iterDurationMap={iterDurationMap}
         />
       </div>
     )
@@ -94,6 +97,7 @@ const WorkflowPreview = () => {
               list={iterationRunResult}
               onHide={doHideIterationDetail}
               onBack={doHideIterationDetail}
+              iterDurationMap={iterDurationMap}
             />
           )
           : (