Integrate WebSocket support: add `/api/v1/ws` endpoint, enable real-time message handling via `websocket_message_queue`, and refactor services and models to support WebSocket-based question and answer updates.

src/VoiceDialogue/api/app.py

@@ -11,7 +11,7 @@ from .core.config import AppConfig
 from .core.lifespan import lifespan
 from .middleware.logging import LoggingMiddleware
 from .middleware.rate_limit import RateLimitMiddleware
-from .routes import tts_routes, asr_routes, system_routes
+from .routes import tts_routes, asr_routes, system_routes, websocket_routes
 
 # 配置日志
 logging.basicConfig(
@@ -63,9 +63,10 @@ def _register_routes(app: FastAPI):
     v1_router.include_router(tts_routes.router, prefix="/tts", tags=["TTS模型管理"])
     v1_router.include_router(asr_routes.router, prefix="/asr", tags=["ASR模型管理"])
     v1_router.include_router(system_routes.router, prefix="/system", tags=["系统管理"])
-
     app.include_router(v1_router)
 
+    app.add_websocket_route("/api/v1/ws", websocket_routes.ws)
+
     # 根路径和健康检查
     _register_health_routes(app)
 

src/VoiceDialogue/api/core/service_factories.py

@@ -1,6 +1,6 @@
 from core.constants import (
     transcribed_text_queue, text_input_queue, audio_output_queue,
-    audio_frames_queue, user_voice_queue
+    audio_frames_queue, user_voice_queue, websocket_message_queue
 )
 from services.audio import EchoCancellingAudioCapture, TTSAudioGenerator, AudioStreamPlayer
 from services.audio.generators import BaseTTSConfig, tts_config_registry
@@ -41,7 +41,8 @@ class ServiceFactories:
         """创建LLM文本生成服务"""
         return LLMResponseGenerator(
             user_question_queue=transcribed_text_queue,
-            generated_answer_queue=text_input_queue
+            generated_answer_queue=text_input_queue,
+            websocket_message_queue=websocket_message_queue,
         )
 
     @staticmethod
@@ -59,7 +60,10 @@ class ServiceFactories:
     @staticmethod
     def create_audio_player() -> AudioStreamPlayer:
         """创建音频播放服务"""
-        return AudioStreamPlayer(audio_playing_queue=audio_output_queue)
+        return AudioStreamPlayer(
+            audio_playing_queue=audio_output_queue,
+            websocket_message_queue=websocket_message_queue
+        )
 
 
 def get_core_voice_service_definitions(system_language: str, tts_config: BaseTTSConfig = None) -> list:

src/VoiceDialogue/api/routes/__init__.py

@@ -1,3 +1,3 @@
-from . import tts_routes, asr_routes, system_routes
+from . import tts_routes, asr_routes, system_routes, websocket_routes
 
-__all__ = ["tts_routes", "asr_routes", "system_routes"]
+__all__ = ["tts_routes", "asr_routes", "system_routes", "websocket_routes"]

src/VoiceDialogue/api/routes/websocket_routes.py

@@ -0,0 +1,34 @@
+import logging
+from queue import Empty
+
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from core.constants import websocket_message_queue, session_manager
+
+ws = APIRouter()
+logger = logging.getLogger(__name__)
+
+
+@ws.websocket("/api/v1/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    """WebSocket连接端点"""
+    try:
+        # 建立连接
+        await websocket.accept()
+        # 保持连接活跃
+        while True:
+            try:
+                message = await websocket_message_queue.get()
+            except Empty:
+                continue
+
+            if message.session_id != session_manager.current_id:
+                continue
+
+            await websocket.send_json(message.model_dump())
+
+    except WebSocketDisconnect:
+        pass
+    except Exception as e:
+        logger.error(f"WebSocket连接异常: {e}")

src/VoiceDialogue/core/constants.py

@@ -1,3 +1,4 @@
+import asyncio
 import multiprocessing
 import threading
 from collections import OrderedDict
@@ -36,6 +37,7 @@ user_voice_queue = multiprocessing.Queue()
 transcribed_text_queue = multiprocessing.Queue()
 text_input_queue = multiprocessing.Queue()
 audio_output_queue = multiprocessing.Queue()
+websocket_message_queue = asyncio.Queue()
 
 # ======================= 全局状态实例 =======================
 

src/VoiceDialogue/models/__init__.py

@@ -1 +1,7 @@
-from .voice_task import VoiceTask
+from .voice_task import VoiceTask, QuestionDisplayMessage, AnswerDisplayMessage
+
+__all__ = (
+    'VoiceTask',
+    'QuestionDisplayMessage',
+    'AnswerDisplayMessage'
+)

src/VoiceDialogue/models/voice_task.py

@@ -1,3 +1,5 @@
+from enum import Enum
+
 import numpy as np
 from pydantic import BaseModel, Field
 
@@ -30,3 +32,25 @@ class VoiceTask(BaseModel):
 
     class Config:
         arbitrary_types_allowed = True
+
+
+class DisplayMessageType(str, Enum):
+    QUESTION = 'question'
+    ANSWER = 'answer'
+
+
+class BaseDisplayMessage(BaseModel):
+    message_type: DisplayMessageType
+    session_id: str
+    task_id: str
+
+
+class QuestionDisplayMessage(BaseDisplayMessage):
+    message_type: DisplayMessageType = DisplayMessageType.QUESTION
+    question: str
+
+
+class AnswerDisplayMessage(BaseDisplayMessage):
+    message_type: DisplayMessageType = DisplayMessageType.ANSWER
+    answer_index: int
+    answer: str

src/VoiceDialogue/services/audio/generator.py

@@ -19,8 +19,12 @@ class TTSAudioGenerator(BaseThread):
     4. 将生成的音频任务放入音频队列中
     """
 
-    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
-                 text_input_queue: Queue, audio_output_queue: Queue, tts_config: BaseTTSConfig):
+    def __init__(
+            self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
+            text_input_queue: Queue,
+            audio_output_queue: Queue,
+            tts_config: BaseTTSConfig,
+    ):
         """
         初始化TTS音频生成器
 

src/VoiceDialogue/services/audio/player.py

@@ -11,16 +11,20 @@ from core.constants import (
     user_still_speaking_event, voice_state_manager, dropped_audio_cache, chat_history_cache,
     silence_over_threshold_event
 )
-from models.voice_task import VoiceTask
+from models.voice_task import VoiceTask, AnswerDisplayMessage
 
 
 class AudioStreamPlayer(BaseThread):
     """音频流播放器 - 负责播放生成的音频并管理播放状态"""
 
-    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
-                 audio_playing_queue):
+    def __init__(
+            self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
+            audio_playing_queue: Queue,
+            websocket_message_queue: Queue,
+    ):
         super().__init__(group, target, name, args, kwargs, daemon=daemon)
         self.audio_playing_queue: Queue = audio_playing_queue
+        self.websocket_message_queue: Queue = websocket_message_queue
 
     def run(self):
         self.is_ready = True
@@ -56,6 +60,15 @@ class AudioStreamPlayer(BaseThread):
                     if answer_id not in voice_state_manager.waiting_second_answer_mapping:
                         continue
 
+                if self.websocket_message_queue:
+                    self.websocket_message_queue.put_nowait(
+                        AnswerDisplayMessage(
+                            session_id=voice_task.session_id,
+                            task_id=task_id,
+                            answer_index=voice_task.answer_index,
+                            answer=voice_task.answer_sentence,
+                        )
+                    )
                 # now = time.time()
                 # print(
                 #     f'整体耗时: {(now - voice_task.send_time):.2f}\n'

src/VoiceDialogue/services/text/generator.py

@@ -8,7 +8,7 @@ from langchain_core.chat_history import InMemoryChatMessageHistory
 from config import paths
 from core.base import BaseThread
 from core.constants import chat_history_cache
-from models.voice_task import VoiceTask
+from models.voice_task import VoiceTask, QuestionDisplayMessage
 from services.text.processor import preprocess_sentence_text, \
     create_langchain_chat_llamacpp_instance, create_langchain_pipeline, warmup_langchain_pipeline
 
@@ -26,14 +26,17 @@ ENGLISH_SYSTEM_PROMPT = ("You are an AI assistant skilled at simulating authenti
 class LLMResponseGenerator(BaseThread):
     """LLM 回答生成器 - 负责使用语言模型生成回答文本"""
 
-    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
-                 user_question_queue: Queue,
-                 generated_answer_queue: Queue
-                 ):
+    def __init__(
+            self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None,
+            user_question_queue: Queue,
+            generated_answer_queue: Queue,
+            websocket_message_queue: Queue = None
+    ):
         super().__init__(group, target, name, args, kwargs, daemon=daemon)
 
         self.user_question_queue = user_question_queue
         self.generated_answer_queue = generated_answer_queue
+        self.websocket_message_queue = websocket_message_queue
 
         self.english_sentence_end_marks = {'!', '?', '.', ',', ':', ';'}
         self.chinese_sentence_end_marks = {'，', '。', '！', '？', '：', '；', '、'}
@@ -121,6 +124,15 @@ class LLMResponseGenerator(BaseThread):
 
         user_question = voice_task.transcribed_text
         print(f'用户问题: {user_question}')
+        if self.websocket_message_queue:
+            self.websocket_message_queue.put_nowait(
+                QuestionDisplayMessage(
+                    session_id=voice_task.session_id,
+                    question=user_question,
+                    task_id=voice_task.id,
+                )
+            )
+
         voice_task.llm_start_time = time.time()
 
         system_prompt = self._get_prompt_by_language(voice_task.language)