From 332e8e68ee0961a193237de9aaa5164661e06801 Mon Sep 17 00:00:00 2001 From: -LAN- Date: Mon, 11 Aug 2025 18:18:07 +0800 Subject: [PATCH] refactor: Change _queue_manager to public attribute queue_manager in task pipelines (#23747) --- .../app/apps/advanced_chat/generate_task_pipeline.py | 10 +++++----- api/core/app/apps/workflow/generate_task_pipeline.py | 2 +- .../app/task_pipeline/based_generate_task_pipeline.py | 4 ++-- .../easy_ui_based_generate_task_pipeline.py | 6 +++--- 4 files changed, 11 insertions(+), 11 deletions(-) diff --git a/api/core/app/apps/advanced_chat/generate_task_pipeline.py b/api/core/app/apps/advanced_chat/generate_task_pipeline.py index abb8db34de..5db7539926 100644 --- a/api/core/app/apps/advanced_chat/generate_task_pipeline.py +++ b/api/core/app/apps/advanced_chat/generate_task_pipeline.py @@ -568,7 +568,7 @@ class AdvancedChatAppGenerateTaskPipeline: ) yield workflow_finish_resp - self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) + self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) def _handle_workflow_partial_success_event( self, @@ -600,7 +600,7 @@ class AdvancedChatAppGenerateTaskPipeline: ) yield workflow_finish_resp - self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) + self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE) def _handle_workflow_failed_event( self, @@ -845,7 +845,7 @@ class AdvancedChatAppGenerateTaskPipeline: # Initialize graph runtime state graph_runtime_state: Optional[GraphRuntimeState] = None - for queue_message in self._base_task_pipeline._queue_manager.listen(): + for queue_message in self._base_task_pipeline.queue_manager.listen(): event = queue_message.event match event: @@ -959,11 +959,11 @@ class AdvancedChatAppGenerateTaskPipeline: if self._base_task_pipeline._output_moderation_handler: if self._base_task_pipeline._output_moderation_handler.should_direct_output(): self._task_state.answer = self._base_task_pipeline._output_moderation_handler.get_final_output() - self._base_task_pipeline._queue_manager.publish( + self._base_task_pipeline.queue_manager.publish( QueueTextChunkEvent(text=self._task_state.answer), PublishFrom.TASK_PIPELINE ) - self._base_task_pipeline._queue_manager.publish( + self._base_task_pipeline.queue_manager.publish( QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE ) return True diff --git a/api/core/app/apps/workflow/generate_task_pipeline.py b/api/core/app/apps/workflow/generate_task_pipeline.py index b1e9a340bd..537c070adf 100644 --- a/api/core/app/apps/workflow/generate_task_pipeline.py +++ b/api/core/app/apps/workflow/generate_task_pipeline.py @@ -711,7 +711,7 @@ class WorkflowAppGenerateTaskPipeline: # Initialize graph runtime state graph_runtime_state = None - for queue_message in self._base_task_pipeline._queue_manager.listen(): + for queue_message in self._base_task_pipeline.queue_manager.listen(): event = queue_message.event match event: diff --git a/api/core/app/task_pipeline/based_generate_task_pipeline.py b/api/core/app/task_pipeline/based_generate_task_pipeline.py index 3ed0c3352f..014c7fd4f5 100644 --- a/api/core/app/task_pipeline/based_generate_task_pipeline.py +++ b/api/core/app/task_pipeline/based_generate_task_pipeline.py @@ -37,7 +37,7 @@ class BasedGenerateTaskPipeline: stream: bool, ) -> None: self._application_generate_entity = application_generate_entity - self._queue_manager = queue_manager + self.queue_manager = queue_manager self._start_at = time.perf_counter() self._output_moderation_handler = self._init_output_moderation() self._stream = stream @@ -113,7 +113,7 @@ class BasedGenerateTaskPipeline: tenant_id=app_config.tenant_id, app_id=app_config.app_id, rule=ModerationRule(type=sensitive_word_avoidance.type, config=sensitive_word_avoidance.config), - queue_manager=self._queue_manager, + queue_manager=self.queue_manager, ) return None diff --git a/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py b/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py index 888434798a..56131d99c9 100644 --- a/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py +++ b/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py @@ -257,7 +257,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline): Process stream response. :return: """ - for message in self._queue_manager.listen(): + for message in self.queue_manager.listen(): if publisher: publisher.publish(message) event = message.event @@ -499,7 +499,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline): if self._output_moderation_handler.should_direct_output(): # stop subscribe new token when output moderation should direct output self._task_state.llm_result.message.content = self._output_moderation_handler.get_final_output() - self._queue_manager.publish( + self.queue_manager.publish( QueueLLMChunkEvent( chunk=LLMResultChunk( model=self._task_state.llm_result.model, @@ -513,7 +513,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline): PublishFrom.TASK_PIPELINE, ) - self._queue_manager.publish( + self.queue_manager.publish( QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE ) return True