feat:add tts-streaming config and future (#5492)

api/core/app/apps/advanced_chat/app_generator_tts_publisher.py

@@ -0,0 +1,135 @@
+import base64
+import concurrent.futures
+import logging
+import queue
+import re
+import threading
+
+from core.app.entities.queue_entities import QueueAgentMessageEvent, QueueLLMChunkEvent, QueueTextChunkEvent
+from core.model_manager import ModelManager
+from core.model_runtime.entities.model_entities import ModelType
+
+
+class AudioTrunk:
+    def __init__(self, status: str, audio):
+        self.audio = audio
+        self.status = status
+
+
+def _invoiceTTS(text_content: str, model_instance, tenant_id: str, voice: str):
+    if not text_content or text_content.isspace():
+        return
+    return model_instance.invoke_tts(
+        content_text=text_content.strip(),
+        user="responding_tts",
+        tenant_id=tenant_id,
+        voice=voice
+    )
+
+
+def _process_future(future_queue, audio_queue):
+    while True:
+        try:
+            future = future_queue.get()
+            if future is None:
+                break
+            for audio in future.result():
+                audio_base64 = base64.b64encode(bytes(audio))
+                audio_queue.put(AudioTrunk("responding", audio=audio_base64))
+        except Exception as e:
+            logging.getLogger(__name__).warning(e)
+            break
+    audio_queue.put(AudioTrunk("finish", b''))
+
+
+class AppGeneratorTTSPublisher:
+
+    def __init__(self, tenant_id: str, voice: str):
+        self.logger = logging.getLogger(__name__)
+        self.tenant_id = tenant_id
+        self.msg_text = ''
+        self._audio_queue = queue.Queue()
+        self._msg_queue = queue.Queue()
+        self.match = re.compile(r'[。.!?]')
+        self.model_manager = ModelManager()
+        self.model_instance = self.model_manager.get_default_model_instance(
+            tenant_id=self.tenant_id,
+            model_type=ModelType.TTS
+        )
+        self.voices = self.model_instance.get_tts_voices()
+        values = [voice.get('value') for voice in self.voices]
+        self.voice = voice
+        if not voice or voice not in values:
+            self.voice = self.voices[0].get('value')
+        self.MAX_SENTENCE = 2
+        self._last_audio_event = None
+        self._runtime_thread = threading.Thread(target=self._runtime).start()
+        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
+
+    def publish(self, message):
+        try:
+            self._msg_queue.put(message)
+        except Exception as e:
+            self.logger.warning(e)
+
+    def _runtime(self):
+        future_queue = queue.Queue()
+        threading.Thread(target=_process_future, args=(future_queue, self._audio_queue)).start()
+        while True:
+            try:
+                message = self._msg_queue.get()
+                if message is None:
+                    if self.msg_text and len(self.msg_text.strip()) > 0:
+                        futures_result = self.executor.submit(_invoiceTTS, self.msg_text,
+                                                              self.model_instance, self.tenant_id, self.voice)
+                        future_queue.put(futures_result)
+                    break
+                elif isinstance(message.event, QueueAgentMessageEvent | QueueLLMChunkEvent):
+                    self.msg_text += message.event.chunk.delta.message.content
+                elif isinstance(message.event, QueueTextChunkEvent):
+                    self.msg_text += message.event.text
+                self.last_message = message
+                sentence_arr, text_tmp = self._extract_sentence(self.msg_text)
+                if len(sentence_arr) >= min(self.MAX_SENTENCE, 7):
+                    self.MAX_SENTENCE += 1
+                    text_content = ''.join(sentence_arr)
+                    futures_result = self.executor.submit(_invoiceTTS, text_content,
+                                                          self.model_instance,
+                                                          self.tenant_id,
+                                                          self.voice)
+                    future_queue.put(futures_result)
+                    if text_tmp:
+                        self.msg_text = text_tmp
+                    else:
+                        self.msg_text = ''
+
+            except Exception as e:
+                self.logger.warning(e)
+                break
+        future_queue.put(None)
+
+    def checkAndGetAudio(self) -> AudioTrunk | None:
+        try:
+            if self._last_audio_event and self._last_audio_event.status == "finish":
+                if self.executor:
+                    self.executor.shutdown(wait=False)
+                return self.last_message
+            audio = self._audio_queue.get_nowait()
+            if audio and audio.status == "finish":
+                self.executor.shutdown(wait=False)
+                self._runtime_thread = None
+            if audio:
+                self._last_audio_event = audio
+            return audio
+        except queue.Empty:
+            return None
+
+    def _extract_sentence(self, org_text):
+        tx = self.match.finditer(org_text)
+        start = 0
+        result = []
+        for i in tx:
+            end = i.regs[0][1]
+            result.append(org_text[start:end])
+            start = end
+        return result, org_text[start:]

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -4,6 +4,8 @@ import time
 from collections.abc import Generator
 from typing import Any, Optional, Union, cast
 
+from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
+from core.app.apps.advanced_chat.app_generator_tts_publisher import AppGeneratorTTSPublisher, AudioTrunk
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
@@ -33,6 +35,8 @@ from core.app.entities.task_entities import (
     ChatbotAppStreamResponse,
     ChatflowStreamGenerateRoute,
     ErrorStreamResponse,
+    MessageAudioEndStreamResponse,
+    MessageAudioStreamResponse,
     MessageEndStreamResponse,
     StreamResponse,
 )
@@ -71,13 +75,13 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
     _iteration_nested_relations: dict[str, list[str]]
 
     def __init__(
-        self, application_generate_entity: AdvancedChatAppGenerateEntity,
-        workflow: Workflow,
-        queue_manager: AppQueueManager,
-        conversation: Conversation,
-        message: Message,
-        user: Union[Account, EndUser],
-        stream: bool
+            self, application_generate_entity: AdvancedChatAppGenerateEntity,
+            workflow: Workflow,
+            queue_manager: AppQueueManager,
+            conversation: Conversation,
+            message: Message,
+            user: Union[Account, EndUser],
+            stream: bool
     ) -> None:
         """
         Initialize AdvancedChatAppGenerateTaskPipeline.
@@ -129,7 +133,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
             self._application_generate_entity.query
         )
 
-        generator = self._process_stream_response(
+        generator = self._wrapper_process_stream_response(
             trace_manager=self._application_generate_entity.trace_manager
         )
         if self._stream:
@@ -138,7 +142,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
             return self._to_blocking_response(generator)
 
     def _to_blocking_response(self, generator: Generator[StreamResponse, None, None]) \
-        -> ChatbotAppBlockingResponse:
+            -> ChatbotAppBlockingResponse:
         """
         Process blocking response.
         :return:
@@ -169,7 +173,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
         raise Exception('Queue listening stopped unexpectedly.')
 
     def _to_stream_response(self, generator: Generator[StreamResponse, None, None]) \
-        -> Generator[ChatbotAppStreamResponse, None, None]:
+            -> Generator[ChatbotAppStreamResponse, None, None]:
         """
         To stream response.
         :return:
@@ -182,14 +186,68 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                 stream_response=stream_response
             )
 
+    def _listenAudioMsg(self, publisher, task_id: str):
+        if not publisher:
+            return None
+        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        if audio_msg and audio_msg.status != "finish":
+            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
+        return None
+
+    def _wrapper_process_stream_response(self, trace_manager: Optional[TraceQueueManager] = None) -> \
+            Generator[StreamResponse, None, None]:
+
+        publisher = None
+        task_id = self._application_generate_entity.task_id
+        tenant_id = self._application_generate_entity.app_config.tenant_id
+        features_dict = self._workflow.features_dict
+
+        if features_dict.get('text_to_speech') and features_dict['text_to_speech'].get('enabled') and features_dict[
+                'text_to_speech'].get('autoPlay') == 'enabled':
+            publisher = AppGeneratorTTSPublisher(tenant_id, features_dict['text_to_speech'].get('voice'))
+        for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
+            while True:
+                audio_response = self._listenAudioMsg(publisher, task_id=task_id)
+                if audio_response:
+                    yield audio_response
+                else:
+                    break
+            yield response
+
+        start_listener_time = time.time()
+        # timeout
+        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
+            try:
+                if not publisher:
+                    break
+                audio_trunk = publisher.checkAndGetAudio()
+                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)
+                    continue
+                if audio_trunk.status == "finish":
+                    break
+                else:
+                    start_listener_time = time.time()
+                    yield MessageAudioStreamResponse(audio=audio_trunk.audio, task_id=task_id)
+            except Exception as e:
+                logger.error(e)
+                break
+        yield MessageAudioEndStreamResponse(audio='', task_id=task_id)
+
     def _process_stream_response(
-        self, trace_manager: Optional[TraceQueueManager] = None
+            self,
+            publisher: AppGeneratorTTSPublisher,
+            trace_manager: Optional[TraceQueueManager] = None
     ) -> Generator[StreamResponse, None, None]:
         """
         Process stream response.
         :return:
         """
         for message in self._queue_manager.listen():
+            if publisher:
+                publisher.publish(message=message)
             event = message.event
 
             if isinstance(event, QueueErrorEvent):
@@ -301,7 +359,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                     continue
 
                 if not self._is_stream_out_support(
-                    event=event
+                        event=event
                 ):
                     continue
 
@@ -318,7 +376,8 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                 yield self._ping_stream_response()
             else:
                 continue
-
+        if publisher:
+            publisher.publish(None)
         if self._conversation_name_generate_thread:
             self._conversation_name_generate_thread.join()
 
@@ -402,7 +461,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
         return stream_generate_routes
 
     def _get_answer_start_at_node_ids(self, graph: dict, target_node_id: str) \
-        -> list[str]:
+            -> list[str]:
         """
         Get answer start at node id.
         :param graph: graph
@@ -457,7 +516,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                 start_node_id = target_node_id
                 start_node_ids.append(start_node_id)
             elif node_type == NodeType.START.value or \
-                node_iteration_id is not None and iteration_start_node_id == source_node.get('id'):
+                    node_iteration_id is not None and iteration_start_node_id == source_node.get('id'):
                 start_node_id = source_node_id
                 start_node_ids.append(start_node_id)
             else:
@@ -515,7 +574,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
 
             # all route chunks are generated
             if self._task_state.current_stream_generate_state.current_route_position == len(
-                self._task_state.current_stream_generate_state.generate_route
+                    self._task_state.current_stream_generate_state.generate_route
             ):
                 self._task_state.current_stream_generate_state = None
 
@@ -525,7 +584,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
         :return:
         """
         if not self._task_state.current_stream_generate_state:
-            return None
+            return
 
         route_chunks = self._task_state.current_stream_generate_state.generate_route[
                        self._task_state.current_stream_generate_state.current_route_position:]
@@ -573,7 +632,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                     # get route chunk node execution info
                     route_chunk_node_execution_info = self._task_state.ran_node_execution_infos[route_chunk_node_id]
                     if (route_chunk_node_execution_info.node_type == NodeType.LLM
-                        and latest_node_execution_info.node_type == NodeType.LLM):
+                            and latest_node_execution_info.node_type == NodeType.LLM):
                         # only LLM support chunk stream output
                         self._task_state.current_stream_generate_state.current_route_position += 1
                         continue
@@ -643,7 +702,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
 
         # all route chunks are generated
         if self._task_state.current_stream_generate_state.current_route_position == len(
-            self._task_state.current_stream_generate_state.generate_route
+                self._task_state.current_stream_generate_state.generate_route
         ):
             self._task_state.current_stream_generate_state = None
 

api/core/app/apps/base_app_queue_manager.py

@@ -51,7 +51,6 @@ class AppQueueManager:
         listen_timeout = current_app.config.get("APP_MAX_EXECUTION_TIME")
         start_time = time.time()
         last_ping_time = 0
-
         while True:
             try:
                 message = self._q.get(timeout=1)

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -1,7 +1,10 @@
 import logging
+import time
 from collections.abc import Generator
 from typing import Any, Optional, Union
 
+from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
+from core.app.apps.advanced_chat.app_generator_tts_publisher import AppGeneratorTTSPublisher, AudioTrunk
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import (
     InvokeFrom,
@@ -25,6 +28,8 @@ from core.app.entities.queue_entities import (
 )
 from core.app.entities.task_entities import (
     ErrorStreamResponse,
+    MessageAudioEndStreamResponse,
+    MessageAudioStreamResponse,
     StreamResponse,
     TextChunkStreamResponse,
     TextReplaceStreamResponse,
@@ -105,7 +110,7 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
         db.session.refresh(self._user)
         db.session.close()
 
-        generator = self._process_stream_response(
+        generator = self._wrapper_process_stream_response(
             trace_manager=self._application_generate_entity.trace_manager
         )
         if self._stream:
@@ -161,8 +166,58 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
                 stream_response=stream_response
             )
 
+    def _listenAudioMsg(self, publisher, task_id: str):
+        if not publisher:
+            return None
+        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        if audio_msg and audio_msg.status != "finish":
+            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
+        return None
+
+    def _wrapper_process_stream_response(self, trace_manager: Optional[TraceQueueManager] = None) -> \
+            Generator[StreamResponse, None, None]:
+
+        publisher = None
+        task_id = self._application_generate_entity.task_id
+        tenant_id = self._application_generate_entity.app_config.tenant_id
+        features_dict = self._workflow.features_dict
+
+        if features_dict.get('text_to_speech') and features_dict['text_to_speech'].get('enabled') and features_dict[
+                'text_to_speech'].get('autoPlay') == 'enabled':
+            publisher = AppGeneratorTTSPublisher(tenant_id, features_dict['text_to_speech'].get('voice'))
+        for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
+            while True:
+                audio_response = self._listenAudioMsg(publisher, task_id=task_id)
+                if audio_response:
+                    yield audio_response
+                else:
+                    break
+            yield response
+
+        start_listener_time = time.time()
+        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
+            try:
+                if not publisher:
+                    break
+                audio_trunk = publisher.checkAndGetAudio()
+                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)
+                    continue
+                if audio_trunk.status == "finish":
+                    break
+                else:
+                    yield MessageAudioStreamResponse(audio=audio_trunk.audio, task_id=task_id)
+            except Exception as e:
+                logger.error(e)
+                break
+        yield MessageAudioEndStreamResponse(audio='', task_id=task_id)
+
+
     def _process_stream_response(
         self,
+        publisher: AppGeneratorTTSPublisher,
         trace_manager: Optional[TraceQueueManager] = None
     ) -> Generator[StreamResponse, None, None]:
         """
@@ -170,6 +225,8 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
         :return:
         """
         for message in self._queue_manager.listen():
+            if publisher:
+                publisher.publish(message=message)
             event = message.event
 
             if isinstance(event, QueueErrorEvent):
@@ -251,6 +308,10 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
             else:
                 continue
 
+        if publisher:
+            publisher.publish(None)
+
+
     def _save_workflow_app_log(self, workflow_run: WorkflowRun) -> None:
         """
         Save workflow app log.

api/core/app/entities/task_entities.py

@@ -69,6 +69,7 @@ class WorkflowTaskState(TaskState):
 
     iteration_nested_node_ids: list[str] = None
 
+
 class AdvancedChatTaskState(WorkflowTaskState):
     """
     AdvancedChatTaskState entity
@@ -86,6 +87,8 @@ class StreamEvent(Enum):
     ERROR = "error"
     MESSAGE = "message"
     MESSAGE_END = "message_end"
+    TTS_MESSAGE = "tts_message"
+    TTS_MESSAGE_END = "tts_message_end"
     MESSAGE_FILE = "message_file"
     MESSAGE_REPLACE = "message_replace"
     AGENT_THOUGHT = "agent_thought"
@@ -130,6 +133,22 @@ class MessageStreamResponse(StreamResponse):
     answer: str
 
 
+class MessageAudioStreamResponse(StreamResponse):
+    """
+    MessageStreamResponse entity
+    """
+    event: StreamEvent = StreamEvent.TTS_MESSAGE
+    audio: str
+
+
+class MessageAudioEndStreamResponse(StreamResponse):
+    """
+    MessageStreamResponse entity
+    """
+    event: StreamEvent = StreamEvent.TTS_MESSAGE_END
+    audio: str
+
+
 class MessageEndStreamResponse(StreamResponse):
     """
     MessageEndStreamResponse entity
@@ -186,6 +205,7 @@ class WorkflowStartStreamResponse(StreamResponse):
     """
     WorkflowStartStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -205,6 +225,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
     """
     WorkflowFinishStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -232,6 +253,7 @@ class NodeStartStreamResponse(StreamResponse):
     """
     NodeStartStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -273,6 +295,7 @@ class NodeFinishStreamResponse(StreamResponse):
     """
     NodeFinishStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -323,10 +346,12 @@ class NodeFinishStreamResponse(StreamResponse):
             }
         }
 
+
 class IterationNodeStartStreamResponse(StreamResponse):
     """
     NodeStartStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -344,10 +369,12 @@ class IterationNodeStartStreamResponse(StreamResponse):
     workflow_run_id: str
     data: Data
 
+
 class IterationNodeNextStreamResponse(StreamResponse):
     """
     NodeStartStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -365,10 +392,12 @@ class IterationNodeNextStreamResponse(StreamResponse):
     workflow_run_id: str
     data: Data
 
+
 class IterationNodeCompletedStreamResponse(StreamResponse):
     """
     NodeCompletedStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -393,10 +422,12 @@ class IterationNodeCompletedStreamResponse(StreamResponse):
     workflow_run_id: str
     data: Data
 
+
 class TextChunkStreamResponse(StreamResponse):
     """
     TextChunkStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -411,6 +442,7 @@ class TextReplaceStreamResponse(StreamResponse):
     """
     TextReplaceStreamResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -473,6 +505,7 @@ class ChatbotAppBlockingResponse(AppBlockingResponse):
     """
     ChatbotAppBlockingResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -492,6 +525,7 @@ class CompletionAppBlockingResponse(AppBlockingResponse):
     """
     CompletionAppBlockingResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -510,6 +544,7 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
     """
     WorkflowAppBlockingResponse entity
     """
+
     class Data(BaseModel):
         """
         Data entity
@@ -528,10 +563,12 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
     workflow_run_id: str
     data: Data
 
+
 class WorkflowIterationState(BaseModel):
     """
     WorkflowIterationState entity
     """
+
     class Data(BaseModel):
         """
         Data entity

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -4,6 +4,8 @@ import time
 from collections.abc import Generator
 from typing import Optional, Union, cast
 
+from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
+from core.app.apps.advanced_chat.app_generator_tts_publisher import AppGeneratorTTSPublisher, AudioTrunk
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import (
     AgentChatAppGenerateEntity,
@@ -32,6 +34,8 @@ from core.app.entities.task_entities import (
     CompletionAppStreamResponse,
     EasyUITaskState,
     ErrorStreamResponse,
+    MessageAudioEndStreamResponse,
+    MessageAudioStreamResponse,
     MessageEndStreamResponse,
     StreamResponse,
 )
@@ -87,6 +91,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
         """
         super().__init__(application_generate_entity, queue_manager, user, stream)
         self._model_config = application_generate_entity.model_conf
+        self._app_config = application_generate_entity.app_config
         self._conversation = conversation
         self._message = message
 
@@ -102,7 +107,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
         self._conversation_name_generate_thread = None
 
     def process(
-        self,
+            self,
     ) -> Union[
         ChatbotAppBlockingResponse,
         CompletionAppBlockingResponse,
@@ -123,7 +128,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
                 self._application_generate_entity.query
             )
 
-        generator = self._process_stream_response(
+        generator = self._wrapper_process_stream_response(
             trace_manager=self._application_generate_entity.trace_manager
         )
         if self._stream:
@@ -202,14 +207,64 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
                     stream_response=stream_response
                 )
 
+    def _listenAudioMsg(self, publisher, task_id: str):
+        if publisher is None:
+            return None
+        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        if audio_msg and audio_msg.status != "finish":
+            # audio_str = audio_msg.audio.decode('utf-8', errors='ignore')
+            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
+        return None
+
+    def _wrapper_process_stream_response(self, trace_manager: Optional[TraceQueueManager] = None) -> \
+            Generator[StreamResponse, None, None]:
+
+        tenant_id = self._application_generate_entity.app_config.tenant_id
+        task_id = self._application_generate_entity.task_id
+        publisher = None
+        text_to_speech_dict = self._app_config.app_model_config_dict.get('text_to_speech')
+        if text_to_speech_dict and text_to_speech_dict.get('autoPlay') == 'enabled' and text_to_speech_dict.get('enabled'):
+            publisher = AppGeneratorTTSPublisher(tenant_id, text_to_speech_dict.get('voice', None))
+        for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
+            while True:
+                audio_response = self._listenAudioMsg(publisher, task_id)
+                if audio_response:
+                    yield audio_response
+                else:
+                    break
+            yield response
+
+        start_listener_time = time.time()
+        # timeout
+        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
+            if publisher is None:
+                break
+            audio = publisher.checkAndGetAudio()
+            if audio 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)
+                continue
+            if audio.status == "finish":
+                break
+            else:
+                start_listener_time = time.time()
+                yield MessageAudioStreamResponse(audio=audio.audio,
+                                                 task_id=task_id)
+        yield MessageAudioEndStreamResponse(audio='', task_id=task_id)
+
     def _process_stream_response(
-        self, trace_manager: Optional[TraceQueueManager] = None
+            self,
+            publisher: AppGeneratorTTSPublisher,
+            trace_manager: Optional[TraceQueueManager] = None
     ) -> Generator[StreamResponse, None, None]:
         """
         Process stream response.
         :return:
         """
         for message in self._queue_manager.listen():
+            if publisher:
+                publisher.publish(message)
             event = message.event
 
             if isinstance(event, QueueErrorEvent):
@@ -272,12 +327,13 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
                 yield self._ping_stream_response()
             else:
                 continue
-
+        if publisher:
+            publisher.publish(None)
         if self._conversation_name_generate_thread:
             self._conversation_name_generate_thread.join()
 
     def _save_message(
-        self, trace_manager: Optional[TraceQueueManager] = None
+            self, trace_manager: Optional[TraceQueueManager] = None
     ) -> None:
         """
         Save message.