Refactor/remove db from cycle manager (#20455)

Signed-off-by: -LAN- <laipz8200@outlook.com>
2025-05-30 04:34:13 +08:00
parent cd0a05f114
commit 482e50aae9
81 changed files with 345 additions and 362 deletions
--- a/api/core/app/apps/advanced_chat/app_generator.py
+++ b/api/core/app/apps/advanced_chat/app_generator.py
@@ -27,8 +27,8 @@ from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
 from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
-from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
--- a/api/core/app/apps/advanced_chat/generate_task_pipeline.py
+++ b/api/core/app/apps/advanced_chat/generate_task_pipeline.py
@@ -62,21 +62,19 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus, WorkflowType
 from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
-from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_cycle_manager import WorkflowCycleManager
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from models import Conversation, EndUser, Message, MessageFile
 from models.account import Account
 from models.enums import CreatorUserRole
-from models.workflow import (
-    Workflow,
-    WorkflowRunStatus,
-)
+from models.workflow import Workflow

 logger = logging.getLogger(__name__)

@@ -128,6 +126,12 @@ class AdvancedChatAppGenerateTaskPipeline:
                SystemVariableKey.WORKFLOW_ID: workflow.id,
                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_run_id,
            },
+            workflow_info=CycleManagerWorkflowInfo(
+                workflow_id=workflow.id,
+                workflow_type=WorkflowType(workflow.type),
+                version=workflow.version,
+                graph_data=workflow.graph_dict,
+            ),
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
        )
@@ -302,15 +306,12 @@ class AdvancedChatAppGenerateTaskPipeline:

                with Session(db.engine, expire_on_commit=False) as session:
                    # init workflow run
-                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start(
-                        session=session,
-                        workflow_id=self._workflow_id,
-                    )
-                    self._workflow_run_id = workflow_execution.id
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
+                    self._workflow_run_id = workflow_execution.id_
                    message = self._get_message(session=session)
                    if not message:
                        raise ValueError(f"Message not found: {self._message_id}")
-                    message.workflow_run_id = workflow_execution.id
+                    message.workflow_run_id = workflow_execution.id_
                    workflow_start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
                        task_id=self._application_generate_entity.task_id,
                        workflow_execution=workflow_execution,
@@ -550,7 +551,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                        workflow_run_id=self._workflow_run_id,
                        total_tokens=graph_runtime_state.total_tokens,
                        total_steps=graph_runtime_state.node_run_steps,
-                        status=WorkflowRunStatus.FAILED,
+                        status=WorkflowExecutionStatus.FAILED,
                        error_message=event.error,
                        conversation_id=self._conversation_id,
                        trace_manager=trace_manager,
@@ -576,7 +577,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                            workflow_run_id=self._workflow_run_id,
                            total_tokens=graph_runtime_state.total_tokens,
                            total_steps=graph_runtime_state.node_run_steps,
-                            status=WorkflowRunStatus.STOPPED,
+                            status=WorkflowExecutionStatus.STOPPED,
                            error_message=event.get_stop_reason(),
                            conversation_id=self._conversation_id,
                            trace_manager=trace_manager,
--- a/api/core/app/apps/common/workflow_response_converter.py
+++ b/api/core/app/apps/common/workflow_response_converter.py
@@ -44,15 +44,14 @@ from core.app.entities.task_entities import (
 )
 from core.file import FILE_MODEL_IDENTITY, File
 from core.tools.tool_manager import ToolManager
-from core.workflow.entities.node_execution_entities import NodeExecution
-from core.workflow.entities.workflow_execution_entities import WorkflowExecution
+from core.workflow.entities.workflow_execution import WorkflowExecution
+from core.workflow.entities.workflow_node_execution import NodeExecution, WorkflowNodeExecutionStatus
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.tool.entities import ToolNodeData
 from models import (
    Account,
    CreatorUserRole,
    EndUser,
-    WorkflowNodeExecutionStatus,
    WorkflowRun,
 )

@@ -73,11 +72,10 @@ class WorkflowResponseConverter:
    ) -> WorkflowStartStreamResponse:
        return WorkflowStartStreamResponse(
            task_id=task_id,
-            workflow_run_id=workflow_execution.id,
+            workflow_run_id=workflow_execution.id_,
            data=WorkflowStartStreamResponse.Data(
-                id=workflow_execution.id,
+                id=workflow_execution.id_,
                workflow_id=workflow_execution.workflow_id,
-                sequence_number=workflow_execution.sequence_number,
                inputs=workflow_execution.inputs,
                created_at=int(workflow_execution.started_at.timestamp()),
            ),
@@ -91,7 +89,7 @@ class WorkflowResponseConverter:
        workflow_execution: WorkflowExecution,
    ) -> WorkflowFinishStreamResponse:
        created_by = None
-        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id))
+        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
        assert workflow_run is not None
        if workflow_run.created_by_role == CreatorUserRole.ACCOUNT:
            stmt = select(Account).where(Account.id == workflow_run.created_by)
@@ -122,11 +120,10 @@ class WorkflowResponseConverter:

        return WorkflowFinishStreamResponse(
            task_id=task_id,
-            workflow_run_id=workflow_execution.id,
+            workflow_run_id=workflow_execution.id_,
            data=WorkflowFinishStreamResponse.Data(
-                id=workflow_execution.id,
+                id=workflow_execution.id_,
                workflow_id=workflow_execution.workflow_id,
-                sequence_number=workflow_execution.sequence_number,
                status=workflow_execution.status,
                outputs=workflow_execution.outputs,
                error=workflow_execution.error_message,
--- a/api/core/app/apps/workflow/app_generator.py
+++ b/api/core/app/apps/workflow/app_generator.py
@@ -25,8 +25,8 @@ from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
 from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
-from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
 from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
@@ -132,7 +132,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            invoke_from=invoke_from,
            call_depth=call_depth,
            trace_manager=trace_manager,
-            workflow_run_id=workflow_run_id,
+            workflow_execution_id=workflow_run_id,
        )

        contexts.plugin_tool_providers.set({})
@@ -279,7 +279,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            single_iteration_run=WorkflowAppGenerateEntity.SingleIterationRunEntity(
                node_id=node_id, inputs=args["inputs"]
            ),
-            workflow_run_id=str(uuid.uuid4()),
+            workflow_execution_id=str(uuid.uuid4()),
        )
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())
@@ -355,7 +355,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            invoke_from=InvokeFrom.DEBUGGER,
            extras={"auto_generate_conversation_name": False},
            single_loop_run=WorkflowAppGenerateEntity.SingleLoopRunEntity(node_id=node_id, inputs=args["inputs"]),
-            workflow_run_id=str(uuid.uuid4()),
+            workflow_execution_id=str(uuid.uuid4()),
        )
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())
--- a/api/core/app/apps/workflow/app_runner.py
+++ b/api/core/app/apps/workflow/app_runner.py
@@ -95,7 +95,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                SystemVariableKey.USER_ID: user_id,
                SystemVariableKey.APP_ID: app_config.app_id,
                SystemVariableKey.WORKFLOW_ID: app_config.workflow_id,
-                SystemVariableKey.WORKFLOW_RUN_ID: self.application_generate_entity.workflow_run_id,
+                SystemVariableKey.WORKFLOW_RUN_ID: self.application_generate_entity.workflow_execution_id,
            }

            variable_pool = VariablePool(
--- a/api/core/app/apps/workflow/generate_task_pipeline.py
+++ b/api/core/app/apps/workflow/generate_task_pipeline.py
@@ -55,11 +55,11 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.workflow.entities.workflow_execution_entities import WorkflowExecution
+from core.workflow.entities.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
 from core.workflow.enums import SystemVariableKey
-from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_cycle_manager import WorkflowCycleManager
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from extensions.ext_database import db
 from models.account import Account
 from models.enums import CreatorUserRole
@@ -69,7 +69,6 @@ from models.workflow import (
    WorkflowAppLog,
    WorkflowAppLogCreatedFrom,
    WorkflowRun,
-    WorkflowRunStatus,
 )

 logger = logging.getLogger(__name__)
@@ -114,8 +113,14 @@ class WorkflowAppGenerateTaskPipeline:
                SystemVariableKey.USER_ID: user_session_id,
                SystemVariableKey.APP_ID: application_generate_entity.app_config.app_id,
                SystemVariableKey.WORKFLOW_ID: workflow.id,
-                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_run_id,
+                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_execution_id,
            },
+            workflow_info=CycleManagerWorkflowInfo(
+                workflow_id=workflow.id,
+                workflow_type=WorkflowType(workflow.type),
+                version=workflow.version,
+                graph_data=workflow.graph_dict,
+            ),
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
        )
@@ -266,17 +271,13 @@ class WorkflowAppGenerateTaskPipeline:
                # override graph runtime state
                graph_runtime_state = event.graph_runtime_state

-                with Session(db.engine, expire_on_commit=False) as session:
-                    # init workflow run
-                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start(
-                        session=session,
-                        workflow_id=self._workflow_id,
-                    )
-                    self._workflow_run_id = workflow_execution.id
-                    start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution=workflow_execution,
-                    )
+                # init workflow run
+                workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
+                self._workflow_run_id = workflow_execution.id_
+                start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution=workflow_execution,
+                )

                yield start_resp
            elif isinstance(
@@ -511,9 +512,9 @@ class WorkflowAppGenerateTaskPipeline:
                        workflow_run_id=self._workflow_run_id,
                        total_tokens=graph_runtime_state.total_tokens,
                        total_steps=graph_runtime_state.node_run_steps,
-                        status=WorkflowRunStatus.FAILED
+                        status=WorkflowExecutionStatus.FAILED
                        if isinstance(event, QueueWorkflowFailedEvent)
-                        else WorkflowRunStatus.STOPPED,
+                        else WorkflowExecutionStatus.STOPPED,
                        error_message=event.error
                        if isinstance(event, QueueWorkflowFailedEvent)
                        else event.get_stop_reason(),
@@ -557,7 +558,7 @@ class WorkflowAppGenerateTaskPipeline:
            tts_publisher.publish(None)

    def _save_workflow_app_log(self, *, session: Session, workflow_execution: WorkflowExecution) -> None:
-        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id))
+        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
        assert workflow_run is not None
        invoke_from = self._application_generate_entity.invoke_from
        if invoke_from == InvokeFrom.SERVICE_API:
--- a/api/core/app/apps/workflow_app_runner.py
+++ b/api/core/app/apps/workflow_app_runner.py
@@ -29,8 +29,8 @@ from core.app.entities.queue_entities import (
    QueueWorkflowStartedEvent,
    QueueWorkflowSucceededEvent,
 )
-from core.workflow.entities.node_entities import NodeRunMetadataKey
 from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.entities.workflow_node_execution import NodeRunMetadataKey
 from core.workflow.graph_engine.entities.event import (
    AgentLogEvent,
    GraphEngineEvent,
--- a/api/core/app/entities/app_invoke_entities.py
+++ b/api/core/app/entities/app_invoke_entities.py
@@ -76,6 +76,8 @@ class AppGenerateEntity(BaseModel):
    App Generate Entity.
    """

+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
    task_id: str

    # app config
@@ -99,9 +101,6 @@ class AppGenerateEntity(BaseModel):
    # tracing instance
    trace_manager: Optional[TraceQueueManager] = None

-    class Config:
-        arbitrary_types_allowed = True
-

 class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
    """
@@ -205,7 +204,7 @@ class WorkflowAppGenerateEntity(AppGenerateEntity):

    # app config
    app_config: WorkflowUIBasedAppConfig
-    workflow_run_id: str
+    workflow_execution_id: str

    class SingleIterationRunEntity(BaseModel):
        """
--- a/api/core/app/entities/queue_entities.py
+++ b/api/core/app/entities/queue_entities.py
@@ -6,7 +6,8 @@ from typing import Any, Optional
 from pydantic import BaseModel

 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
-from core.workflow.entities.node_entities import AgentNodeStrategyInit, NodeRunMetadataKey
+from core.workflow.entities.node_entities import AgentNodeStrategyInit
+from core.workflow.entities.workflow_node_execution import NodeRunMetadataKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.base import BaseNodeData
--- a/api/core/app/entities/task_entities.py
+++ b/api/core/app/entities/task_entities.py
@@ -6,8 +6,8 @@ from pydantic import BaseModel, ConfigDict

 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.workflow.entities.node_entities import AgentNodeStrategyInit, NodeRunMetadataKey
-from models.workflow import WorkflowNodeExecutionStatus
+from core.workflow.entities.node_entities import AgentNodeStrategyInit
+from core.workflow.entities.workflow_node_execution import NodeRunMetadataKey, WorkflowNodeExecutionStatus


 class TaskState(BaseModel):
@@ -189,7 +189,6 @@ class WorkflowStartStreamResponse(StreamResponse):

        id: str
        workflow_id: str
-        sequence_number: int
        inputs: Mapping[str, Any]
        created_at: int

@@ -210,7 +209,6 @@ class WorkflowFinishStreamResponse(StreamResponse):

        id: str
        workflow_id: str
-        sequence_number: int
        status: str
        outputs: Optional[Mapping[str, Any]] = None
        error: Optional[str] = None