refactor: remove redundant comments from task pipeline

Clean up code by removing unnecessary comments that were added during previous refactoring iterations. The code is now self-documenting and these comments were redundant.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
pull/22600/head
-LAN- 10 months ago
parent ed7ca673d9
commit d4dc1478ef
No known key found for this signature in database
GPG Key ID: 6BA0D108DED011FF

@ -166,7 +166,6 @@ class AdvancedChatAppGenerateTaskPipeline:
Process generate task pipeline. Process generate task pipeline.
:return: :return:
""" """
# start generate conversation name thread
self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name( self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
conversation_id=self._conversation_id, query=self._application_generate_entity.query conversation_id=self._conversation_id, query=self._application_generate_entity.query
) )
@ -258,15 +257,12 @@ class AdvancedChatAppGenerateTaskPipeline:
yield response yield response
start_listener_time = time.time() start_listener_time = time.time()
# timeout
while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT: while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
try: try:
if not tts_publisher: if not tts_publisher:
break break
audio_trunk = tts_publisher.check_and_get_audio() audio_trunk = tts_publisher.check_and_get_audio()
if audio_trunk is None: if audio_trunk is None:
# release cpu
# sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
time.sleep(TTS_AUTO_PLAY_YIELD_CPU_TIME) time.sleep(TTS_AUTO_PLAY_YIELD_CPU_TIME)
continue continue
if audio_trunk.status == "finish": if audio_trunk.status == "finish":
@ -848,20 +844,16 @@ class AdvancedChatAppGenerateTaskPipeline:
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
# Use elegant pattern matching for event dispatch
match event: match event:
# Handle QueueWorkflowStartedEvent specially - it has a side effect
case QueueWorkflowStartedEvent(): case QueueWorkflowStartedEvent():
graph_runtime_state = event.graph_runtime_state # Side effect! graph_runtime_state = event.graph_runtime_state
yield from self._handle_workflow_started_event(event) yield from self._handle_workflow_started_event(event)
# Handle QueueTextChunkEvent specially - needs queue_message
case QueueTextChunkEvent(): case QueueTextChunkEvent():
yield from self._handle_text_chunk_event( yield from self._handle_text_chunk_event(
event, tts_publisher=tts_publisher, queue_message=queue_message event, tts_publisher=tts_publisher, queue_message=queue_message
) )
# Handle events that cause loop breaks
case QueueErrorEvent(): case QueueErrorEvent():
yield from self._handle_error_event(event) yield from self._handle_error_event(event)
break break
@ -890,9 +882,7 @@ class AdvancedChatAppGenerateTaskPipeline:
) )
): ):
yield from responses yield from responses
# Continue with next event (elegant fallthrough)
# Cleanup - publish None when task finished
if tts_publisher: if tts_publisher:
tts_publisher.publish(None) tts_publisher.publish(None)
@ -964,7 +954,6 @@ 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():
# stop subscribe new token when output moderation 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

@ -711,20 +711,16 @@ class WorkflowAppGenerateTaskPipeline:
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
# Use elegant pattern matching for event dispatch
match event: match event:
# Handle QueueWorkflowStartedEvent specially - it has a side effect
case QueueWorkflowStartedEvent(): case QueueWorkflowStartedEvent():
graph_runtime_state = event.graph_runtime_state # Side effect! graph_runtime_state = event.graph_runtime_state
yield from self._handle_workflow_started_event(event) yield from self._handle_workflow_started_event(event)
# Handle QueueTextChunkEvent specially - needs queue_message
case QueueTextChunkEvent(): case QueueTextChunkEvent():
yield from self._handle_text_chunk_event( yield from self._handle_text_chunk_event(
event, tts_publisher=tts_publisher, queue_message=queue_message event, tts_publisher=tts_publisher, queue_message=queue_message
) )
# Handle events that cause loop breaks
case QueueErrorEvent(): case QueueErrorEvent():
yield from self._handle_error_event(event) yield from self._handle_error_event(event)
break break
@ -741,9 +737,7 @@ class WorkflowAppGenerateTaskPipeline:
) )
): ):
yield from responses yield from responses
# Continue with next event (elegant fallthrough)
# Cleanup - publish None when task finished
if tts_publisher: if tts_publisher:
tts_publisher.publish(None) tts_publisher.publish(None)

Loading…
Cancel
Save