outman/dify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: replace Exception with ValueError for uninitialized workflow run and queue listening errors

Browse Source 
Signed-off-by: -LAN- <laipz8200@outlook.com>
This commit is contained in:
-LAN- 2024-12-23 12:14:28 +08:00
parent cdc854d15b
commit fc15b95b81
No known key found for this signature in database
GPG Key ID: 6BA0D108DED011FF
2 changed files with 17 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
api/core/app/apps/advanced_chat/generate_task_pipeline.py
View File

@ -180,7 +180,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
else: else:
continue continue
raise Exception("Queue listening stopped unexpectedly.") raise ValueError("queue listening stopped unexpectedly.")
def _to_stream_response( def _to_stream_response(
self, generator: Generator[StreamResponse, None, None] self, generator: Generator[StreamResponse, None, None]
@ -295,6 +295,8 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
event, event,
QueueNodeRetryEvent, QueueNodeRetryEvent,
): ):
if not workflow_run:
raise ValueError("workflow run not initialized.")
workflow_node_execution = self._handle_workflow_node_execution_retried( workflow_node_execution = self._handle_workflow_node_execution_retried(
workflow_run=workflow_run, event=event workflow_run=workflow_run, event=event
) )
@ -309,7 +311,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
yield response yield response
elif isinstance(event, QueueNodeStartedEvent): elif isinstance(event, QueueNodeStartedEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
workflow_node_execution = self._handle_node_execution_start(workflow_run=workflow_run, event=event) workflow_node_execution = self._handle_node_execution_start(workflow_run=workflow_run, event=event)
@ -350,45 +352,45 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
elif isinstance(event, QueueParallelBranchRunStartedEvent): elif isinstance(event, QueueParallelBranchRunStartedEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
yield self._workflow_parallel_branch_start_to_stream_response( yield self._workflow_parallel_branch_start_to_stream_response(
task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event
) )
elif isinstance(event, QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent): elif isinstance(event, QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
yield self._workflow_parallel_branch_finished_to_stream_response( yield self._workflow_parallel_branch_finished_to_stream_response(
task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event
) )
elif isinstance(event, QueueIterationStartEvent): elif isinstance(event, QueueIterationStartEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
yield self._workflow_iteration_start_to_stream_response( yield self._workflow_iteration_start_to_stream_response(
task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event
) )
elif isinstance(event, QueueIterationNextEvent): elif isinstance(event, QueueIterationNextEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
yield self._workflow_iteration_next_to_stream_response( yield self._workflow_iteration_next_to_stream_response(
task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event
) )
elif isinstance(event, QueueIterationCompletedEvent): elif isinstance(event, QueueIterationCompletedEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
yield self._workflow_iteration_completed_to_stream_response( yield self._workflow_iteration_completed_to_stream_response(
task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event task_id=self._application_generate_entity.task_id, workflow_run=workflow_run, event=event
) )
elif isinstance(event, QueueWorkflowSucceededEvent): elif isinstance(event, QueueWorkflowSucceededEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
if not graph_runtime_state: if not graph_runtime_state:
raise Exception("Graph runtime state not initialized.") raise ValueError("workflow run not initialized.")
workflow_run = self._handle_workflow_run_success( workflow_run = self._handle_workflow_run_success(
workflow_run=workflow_run, workflow_run=workflow_run,
@ -407,10 +409,10 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
self._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) self._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
elif isinstance(event, QueueWorkflowPartialSuccessEvent): elif isinstance(event, QueueWorkflowPartialSuccessEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
if not graph_runtime_state: if not graph_runtime_state:
raise Exception("Graph runtime state not initialized.") raise ValueError("graph runtime state not initialized.")
workflow_run = self._handle_workflow_run_partial_success( workflow_run = self._handle_workflow_run_partial_success(
workflow_run=workflow_run, workflow_run=workflow_run,
@ -430,10 +432,10 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
self._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) self._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
elif isinstance(event, QueueWorkflowFailedEvent): elif isinstance(event, QueueWorkflowFailedEvent):
if not workflow_run: if not workflow_run:
raise Exception("Workflow run not initialized.") raise ValueError("workflow run not initialized.")
if not graph_runtime_state: if not graph_runtime_state:
raise Exception("Graph runtime state not initialized.") raise ValueError("graph runtime state not initialized.")
workflow_run = self._handle_workflow_run_failed( workflow_run = self._handle_workflow_run_failed(
workflow_run=workflow_run, workflow_run=workflow_run,
@ -523,7 +525,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
yield self._message_replace_to_stream_response(answer=event.text) yield self._message_replace_to_stream_response(answer=event.text)
elif isinstance(event, QueueAdvancedChatMessageEndEvent): elif isinstance(event, QueueAdvancedChatMessageEndEvent):
if not graph_runtime_state: if not graph_runtime_state:
raise Exception("Graph runtime state not initialized.") raise ValueError("graph runtime state not initialized.")
output_moderation_answer = self._handle_output_moderation_when_task_finished(self._task_state.answer) output_moderation_answer = self._handle_output_moderation_when_task_finished(self._task_state.answer)
if output_moderation_answer: if output_moderation_answer:

2
api/core/app/apps/workflow/generate_task_pipeline.py
View File

@ -155,7 +155,7 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
else: else:
continue continue
raise Exception("Queue listening stopped unexpectedly.") raise ValueError("queue listening stopped unexpectedly.")
def _to_stream_response( def _to_stream_response(
self, generator: Generator[StreamResponse, None, None] self, generator: Generator[StreamResponse, None, None]