jyapayne/dify
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: Change _queue_manager to public attribute queue_manager in task pipelines (#23747)

Browse source
This commit is contained in:
-LAN- 2025-08-11 18:18:07 +08:00 • committed by GitHub
commit 332e8e68ee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 11 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

10
api/core/app/apps/advanced_chat/generate_task_pipeline.py
View file

@ -568,7 +568,7 @@ class AdvancedChatAppGenerateTaskPipeline:
) )
yield workflow_finish_resp 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( def _handle_workflow_partial_success_event(
self, self,
@ -600,7 +600,7 @@ class AdvancedChatAppGenerateTaskPipeline:
) )
yield workflow_finish_resp 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( def _handle_workflow_failed_event(
self, self,
@ -845,7 +845,7 @@ class AdvancedChatAppGenerateTaskPipeline:
# Initialize graph runtime state # Initialize graph runtime state
graph_runtime_state: Optional[GraphRuntimeState] = None 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 event = queue_message.event
match 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:
if self._base_task_pipeline._output_moderation_handler.should_direct_output(): 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._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 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 QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
) )
return True return True

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

@ -711,7 +711,7 @@ class WorkflowAppGenerateTaskPipeline:
# Initialize graph runtime state # Initialize graph runtime state
graph_runtime_state = None 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 event = queue_message.event
match event: match event:

4
api/core/app/task_pipeline/based_generate_task_pipeline.py
View file

@ -37,7 +37,7 @@ class BasedGenerateTaskPipeline:
stream: bool, stream: bool,
) -> None: ) -> None:
self._application_generate_entity = application_generate_entity self._application_generate_entity = application_generate_entity
self._queue_manager = queue_manager self.queue_manager = queue_manager
self._start_at = time.perf_counter() self._start_at = time.perf_counter()
self._output_moderation_handler = self._init_output_moderation() self._output_moderation_handler = self._init_output_moderation()
self._stream = stream self._stream = stream
@ -113,7 +113,7 @@ class BasedGenerateTaskPipeline:
tenant_id=app_config.tenant_id, tenant_id=app_config.tenant_id,
app_id=app_config.app_id, app_id=app_config.app_id,
rule=ModerationRule(type=sensitive_word_avoidance.type, config=sensitive_word_avoidance.config), rule=ModerationRule(type=sensitive_word_avoidance.type, config=sensitive_word_avoidance.config),
queue_manager=self._queue_manager, queue_manager=self.queue_manager,
) )
return None return None

6
api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py
View file

@ -257,7 +257,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
Process stream response. Process stream response.
:return: :return:
""" """
for message in self._queue_manager.listen(): for message in self.queue_manager.listen():
if publisher: if publisher:
publisher.publish(message) publisher.publish(message)
event = message.event event = message.event
@ -499,7 +499,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
if self._output_moderation_handler.should_direct_output(): if self._output_moderation_handler.should_direct_output():
# stop subscribe new token when output moderation 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._task_state.llm_result.message.content = self._output_moderation_handler.get_final_output()
self._queue_manager.publish( self.queue_manager.publish(
QueueLLMChunkEvent( QueueLLMChunkEvent(
chunk=LLMResultChunk( chunk=LLMResultChunk(
model=self._task_state.llm_result.model, model=self._task_state.llm_result.model,
@ -513,7 +513,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
PublishFrom.TASK_PIPELINE, PublishFrom.TASK_PIPELINE,
) )
self._queue_manager.publish( self.queue_manager.publish(
QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
) )
return True return True