refactor(workflow_cycle_manager): Refactor `_workflow_iteration_start_to_stream_response`

Signed-off-by: -LAN- <laipz8200@outlook.com>
pull/20067/head
-LAN- 1 year ago
parent 5da9bc636e
commit 123a2cee5f
No known key found for this signature in database
GPG Key ID: 6BA0D108DED011FF

@ -416,16 +416,11 @@ class AdvancedChatAppGenerateTaskPipeline:
if not self._workflow_run_id: if not self._workflow_run_id:
raise ValueError("workflow run not initialized.") raise ValueError("workflow run not initialized.")
with Session(db.engine, expire_on_commit=False) as session: iter_start_resp = self._workflow_cycle_manager._workflow_iteration_start_to_stream_response(
workflow_run = self._workflow_cycle_manager._get_workflow_run( task_id=self._application_generate_entity.task_id,
session=session, workflow_run_id=self._workflow_run_id workflow_execution_id=self._workflow_run_id,
) event=event,
iter_start_resp = self._workflow_cycle_manager._workflow_iteration_start_to_stream_response( )
session=session,
task_id=self._application_generate_entity.task_id,
workflow_run=workflow_run,
event=event,
)
yield iter_start_resp yield iter_start_resp
elif isinstance(event, QueueIterationNextEvent): elif isinstance(event, QueueIterationNextEvent):

@ -370,16 +370,11 @@ class WorkflowAppGenerateTaskPipeline:
if not self._workflow_run_id: if not self._workflow_run_id:
raise ValueError("workflow run not initialized.") raise ValueError("workflow run not initialized.")
with Session(db.engine, expire_on_commit=False) as session: iter_start_resp = self._workflow_cycle_manager._workflow_iteration_start_to_stream_response(
workflow_run = self._workflow_cycle_manager._get_workflow_run( task_id=self._application_generate_entity.task_id,
session=session, workflow_run_id=self._workflow_run_id workflow_execution_id=self._workflow_run_id,
) event=event,
iter_start_resp = self._workflow_cycle_manager._workflow_iteration_start_to_stream_response( )
session=session,
task_id=self._application_generate_entity.task_id,
workflow_run=workflow_run,
event=event,
)
yield iter_start_resp yield iter_start_resp

@ -675,12 +675,15 @@ class WorkflowCycleManager:
) )
def _workflow_iteration_start_to_stream_response( def _workflow_iteration_start_to_stream_response(
self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueIterationStartEvent self,
*,
task_id: str,
workflow_execution_id: str,
event: QueueIterationStartEvent,
) -> IterationNodeStartStreamResponse: ) -> IterationNodeStartStreamResponse:
_ = session
return IterationNodeStartStreamResponse( return IterationNodeStartStreamResponse(
task_id=task_id, task_id=task_id,
workflow_run_id=workflow_run.id, workflow_run_id=workflow_execution_id,
data=IterationNodeStartStreamResponse.Data( data=IterationNodeStartStreamResponse.Data(
id=event.node_id, id=event.node_id,
node_id=event.node_id, node_id=event.node_id,

Loading…
Cancel
Save