Introduce core module for API lifecycle management: add configuration, service factories, service manager, and lifespan handlers to streamline application startup, shutdown, and service orchestration.

src/VoiceDialogue/api/app.py

@@ -1,17 +1,11 @@
 import logging
-from contextlib import asynccontextmanager
 from typing import Dict, Any
 
 from fastapi import FastAPI, HTTPException, APIRouter
 from fastapi.middleware.cors import CORSMiddleware
 
-from services.audio.audio_player import AudioStreamPlayer
-from services.core.constants import (
-    transcribed_text_queue, text_input_queue, audio_output_queue, audio_frames_queue, user_voice_queue
-)
-from services.speech import SpeechStateMonitor, ASRWorker
-from services.text.text_generator import LLMResponseGenerator
-from utils import get_system_language
+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
@@ -23,138 +17,99 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
-# 全局状态存储
-app_state: Dict[str, Any] = {}
 
+def create_app() -> FastAPI:
+    """创建并配置FastAPI应用"""
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    """应用启动和关闭的生命周期管理"""
-    # 启动时的初始化
-    logger.info("正在启动VoiceDialogue API服务...")
+    # 应用配置
+    config = AppConfig()
 
-    system_default_language = get_system_language()
+    # 创建FastAPI应用
+    app = FastAPI(
+        title=config.title,
+        description=config.description,
+        version=config.version,
+        docs_url=config.docs_url,
+        redoc_url=config.redoc_url,
+        lifespan=lifespan,
+    )
 
-    # 初始化TTS配置注册表
-    try:
-        from services.audio.audio_generator import tts_config_registry
-        logger.info(f"已加载 {len(tts_config_registry.get_all_configs())} 个TTS配置")
-        app_state["tts_configs_loaded"] = True
-    except Exception as e:
-        logger.error(f"TTS配置加载失败: {e}")
-        app_state["tts_configs_loaded"] = False
+    # 添加CORS中间件
+    app.add_middleware(CORSMiddleware, **config.get_cors_config())
 
-    speech_monitor = SpeechStateMonitor(
-        audio_frame_queue=audio_frames_queue,
-        user_voice_queue=user_voice_queue,
-    )
-    speech_monitor.start()
-    app_state['speech_monitor'] = speech_monitor
+    # 添加自定义中间件
+    app.add_middleware(LoggingMiddleware)
+    app.add_middleware(RateLimitMiddleware)
 
-    asr_worker = ASRWorker(
-        user_voice_queue=user_voice_queue, transcribed_text_queue=transcribed_text_queue,
-        language=system_default_language
-    )
-    asr_worker.start()
-    app_state['asr_worker'] = asr_worker
+    # 注册路由
+    _register_routes(app)
 
-    llm_generator = LLMResponseGenerator(
-        user_question_queue=transcribed_text_queue, generated_answer_queue=text_input_queue
-    )
-    llm_generator.start()
-    app_state['llm_generator'] = llm_generator
-
-    audio_player = AudioStreamPlayer(audio_playing_queue=audio_output_queue)
-    audio_player.start()
-    app_state['audio_player'] = audio_player
-
-    app_state["system_running"] = True
-    logger.info("VoiceDialogue API服务启动完成")
-    yield
-
-    # 关闭时的清理
-    logger.info("正在关闭VoiceDialogue API服务...")
-    app_state["system_running"] = False
-    logger.info("VoiceDialogue API服务已关闭")
-
-
-# 创建FastAPI应用
-app = FastAPI(
-    title="VoiceDialogue API",
-    description="""
-    语音对话系统的HTTP API接口
-    
-    ## 功能特性
-    
-    * **TTS模型管理**: 查看、加载、删除TTS模型
-    * **模型状态监控**: 实时监控模型下载和加载状态
-    * **RESTful API**: 标准的REST接口设计
-    * **自动文档**: 自动生成的API文档和测试界面
-    
-    ## 使用方法
-    
-    1. 查看所有可用的TTS模型: `GET /api/v1/tts/models`
-    2. 加载指定模型: `POST /api/v1/tts/models/load`
-    3. 查看模型状态: `GET /api/v1/tts/models/{model_id}/status`
-    4. 删除模型: `DELETE /api/v1/tts/models/{model_id}`
-    """,
-    version="1.0.0",
-    docs_url="/docs",
-    redoc_url="/redoc",
-    lifespan=lifespan,
-)
+    # 注册异常处理器
+    _register_exception_handlers(app)
 
-# 添加CORS中间件
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # 生产环境中应该设置具体的域名
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
+    return app
 
-# 添加自定义中间件
-app.add_middleware(LoggingMiddleware)
-app.add_middleware(RateLimitMiddleware)
-
-# 注册路由
-v1_router = APIRouter(prefix="/api/v1")
-# v1_router.include_router(voice_routes.router, prefix="/voice", tags=["语音处理"])
-# v1_router.include_router(system_routes.router, prefix="/system", tags=["系统控制"])
-v1_router.include_router(tts_routes.router, prefix="/tts", tags=["TTS模型管理"])
-v1_router.include_router(asr_routes.router, prefix="/asr", tags=["ASR模型管理"])
-
-app.include_router(v1_router)
-
-
-@app.get("/")
-async def root():
-    """根路径健康检查"""
-    return {
-        "message": "欢迎使用VoiceDialogue API",
-        "status": "running",
-        "version": "1.0.0",
-        "docs_url": "/docs",
-        "redoc_url": "/redoc"
-    }
-
-
-@app.get("/health")
-async def health_check():
-    """健康检查端点"""
-    return {
-        "status": "healthy",
-        "tts_configs_loaded": app_state.get("tts_configs_loaded", False),
-        "system_running": app_state.get("system_running", False),
-        "available_models": len(app_state.get("available_models", []))
-    }
-
-
-# 全局异常处理器
-@app.exception_handler(Exception)
-async def global_exception_handler(request, exc):
-    logger.error(f"未处理的异常: {exc}", exc_info=True)
-    return HTTPException(
-        status_code=500,
-        detail="内部服务器错误"
-    )
+
+def _register_routes(app: FastAPI):
+    """注册所有路由"""
+    # API路由
+    v1_router = APIRouter(prefix="/api/v1")
+    v1_router.include_router(tts_routes.router, prefix="/tts", tags=["TTS模型管理"])
+    v1_router.include_router(asr_routes.router, prefix="/asr", tags=["ASR模型管理"])
+
+    app.include_router(v1_router)
+
+    # 根路径和健康检查
+    _register_health_routes(app)
+
+
+def _register_health_routes(app: FastAPI):
+    """注册健康检查路由"""
+
+    @app.get("/")
+    async def root():
+        """根路径健康检查"""
+        return {
+            "message": "欢迎使用VoiceDialogue API",
+            "status": "running",
+            "version": "1.0.0",
+            "docs_url": "/docs",
+            "redoc_url": "/redoc"
+        }
+
+    @app.get("/health")
+    async def health_check():
+        """健康检查端点"""
+        app_state = getattr(app, 'state', {})
+
+        return {
+            "status": "healthy",
+            "tts_configs_loaded": app_state.get("tts_configs_loaded", False),
+            "system_running": app_state.get("system_running", False),
+            "tts_config_count": app_state.get("tts_config_count", 0),
+            "services": _get_service_status(app_state)
+        }
+
+
+def _get_service_status(app_state: Dict[str, Any]) -> dict:
+    """获取服务状态信息"""
+    service_manager = app_state.get("service_manager")
+    if service_manager:
+        return service_manager.get_service_status()
+    return {"total_services": 0, "services": {}}
+
+
+def _register_exception_handlers(app: FastAPI):
+    """注册全局异常处理器"""
+
+    @app.exception_handler(Exception)
+    async def global_exception_handler(request, exc):
+        logger.error(f"未处理的异常: {exc}", exc_info=True)
+        return HTTPException(
+            status_code=500,
+            detail="内部服务器错误"
+        )
+
+
+# 创建应用实例
+app = create_app()

src/VoiceDialogue/api/core/__init__.py

@@ -0,0 +1,15 @@
+from .config import AppConfig, TTSConfigInitializer
+from .lifespan import lifespan, LifespanManager
+from .service_factories import ServiceFactories, get_service_definitions
+from .service_manager import ServiceManager, ServiceDefinition
+
+__all__ = [
+    'ServiceManager',
+    'ServiceDefinition',
+    'ServiceFactories',
+    'get_service_definitions',
+    'AppConfig',
+    'TTSConfigInitializer',
+    'lifespan',
+    'LifespanManager'
+]

src/VoiceDialogue/api/core/config.py

@@ -0,0 +1,79 @@
+import logging
+from typing import Dict, Any
+
+logger = logging.getLogger(__name__)
+
+
+class TTSConfigInitializer:
+    """TTS配置初始化器"""
+
+    @staticmethod
+    def initialize() -> Dict[str, Any]:
+        """初始化TTS配置"""
+        result = {
+            "tts_configs_loaded": False,
+            "tts_config_count": 0,
+            "tts_config_errors": []
+        }
+
+        try:
+            from services.audio.audio_generator import tts_config_registry
+            config_count = len(tts_config_registry.get_all_configs())
+
+            result.update({
+                "tts_configs_loaded": True,
+                "tts_config_count": config_count
+            })
+
+            logger.info(f"已加载 {config_count} 个TTS配置")
+
+        except ImportError as e:
+            error_msg = f"TTS模块导入失败: {e}"
+            logger.error(error_msg)
+            result["tts_config_errors"].append(error_msg)
+
+        except Exception as e:
+            error_msg = f"TTS配置加载失败: {e}"
+            logger.error(error_msg, exc_info=True)
+            result["tts_config_errors"].append(error_msg)
+
+        return result
+
+
+class AppConfig:
+    """应用配置类"""
+
+    def __init__(self):
+        self.title = "VoiceDialogue API"
+        self.version = "1.0.0"
+        self.description = self._get_description()
+        self.docs_url = "/docs"
+        self.redoc_url = "/redoc"
+
+    def _get_description(self) -> str:
+        return """
+        语音对话系统的HTTP API接口
+        
+        ## 功能特性
+        
+        * **TTS模型管理**: 查看、加载、删除TTS模型
+        * **模型状态监控**: 实时监控模型下载和加载状态
+        * **RESTful API**: 标准的REST接口设计
+        * **自动文档**: 自动生成的API文档和测试界面
+        
+        ## 使用方法
+        
+        1. 查看所有可用的TTS模型: `GET /api/v1/tts/models`
+        2. 加载指定模型: `POST /api/v1/tts/models/load`
+        3. 查看模型状态: `GET /api/v1/tts/models/{model_id}/status`
+        4. 删除模型: `DELETE /api/v1/tts/models/{model_id}`
+        """
+
+    def get_cors_config(self) -> dict:
+        """获取CORS配置"""
+        return {
+            "allow_origins": ["*"],  # 生产环境中应该设置具体的域名
+            "allow_credentials": True,
+            "allow_methods": ["*"],
+            "allow_headers": ["*"],
+        }

src/VoiceDialogue/api/core/lifespan.py

@@ -0,0 +1,102 @@
+import logging
+import time
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from utils import get_system_language
+from .config import TTSConfigInitializer
+from .service_factories import get_service_definitions
+from .service_manager import ServiceManager
+
+logger = logging.getLogger(__name__)
+
+
+class LifespanManager:
+    """应用生命周期管理器"""
+
+    def __init__(self, app: FastAPI):
+        self.app = app
+        self.service_manager = ServiceManager()
+
+    async def startup(self):
+        """应用启动逻辑"""
+        logger.info("正在启动VoiceDialogue API服务...")
+        startup_start_time = time.time()
+
+        try:
+            # 初始化系统语言
+            system_language = get_system_language()
+            logger.info(f"系统默认语言: {system_language}")
+
+            # 初始化TTS配置
+            tts_config = TTSConfigInitializer.initialize()
+            self._update_app_state(tts_config)
+
+            # 获取服务定义
+            service_definitions = get_service_definitions(system_language)
+
+            # 启动所有服务
+            await self._start_all_services(service_definitions)
+
+            # 更新应用状态
+            self._update_app_state({
+                "service_manager": self.service_manager,
+                "system_running": True,
+                "system_language": system_language
+            })
+
+            # 记录启动信息
+            startup_duration = time.time() - startup_start_time
+            service_status = self.service_manager.get_service_status()
+
+            logger.info(f"VoiceDialogue API服务启动完成")
+            logger.info(f"启动总耗时: {startup_duration:.2f}秒")
+            logger.info(f"启动的服务数量: {service_status['total_services']}")
+
+            if service_status['startup_errors']:
+                logger.warning(f"启动时发生 {service_status['startup_errors']} 个错误")
+
+        except Exception as e:
+            logger.error(f"服务启动失败: {e}", exc_info=True)
+            await self.shutdown()
+            raise
+
+    def _update_app_state(self, state_updates: dict):
+        """更新应用状态"""
+        for key, value in state_updates.items():
+            setattr(self.app.state, key, value)
+
+    async def _start_all_services(self, service_definitions):
+        """启动所有服务"""
+        for service_def in service_definitions:
+            success = self.service_manager.start_service(service_def)
+            if not success and service_def.required:
+                raise RuntimeError(f"必需服务 {service_def.name} 启动失败")
+
+    async def shutdown(self):
+        """应用关闭逻辑"""
+        logger.info("正在关闭VoiceDialogue API服务...")
+
+        # 更新状态
+        setattr(self.app.state, "system_running", False)
+
+        # 停止所有服务
+        self.service_manager.stop_all_services()
+
+        logger.info("VoiceDialogue API服务已关闭")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """FastAPI生命周期管理"""
+    # 创建生命周期管理器
+    lifespan_manager = LifespanManager(app)
+
+    try:
+        # 启动
+        await lifespan_manager.startup()
+        yield
+    finally:
+        # 关闭
+        await lifespan_manager.shutdown()

src/VoiceDialogue/api/core/service_factories.py

@@ -0,0 +1,120 @@
+from typing import Any
+
+from services.audio import EchoCancellingAudioCapture, TTSAudioGenerator, AudioStreamPlayer
+from services.audio.audio_generator import BaseTTSConfig
+from services.core.constants import (
+    transcribed_text_queue, text_input_queue, audio_output_queue,
+    audio_frames_queue, user_voice_queue
+)
+from services.speech import SpeechStateMonitor, ASRWorker
+from services.text.text_generator import LLMResponseGenerator
+from .service_manager import ServiceDefinition
+
+
+class ServiceFactories:
+    """服务工厂类，封装所有服务的创建逻辑"""
+
+    @staticmethod
+    def create_audio_capture() -> EchoCancellingAudioCapture:
+        """创建音频捕获服务"""
+        return EchoCancellingAudioCapture(
+            audio_frames_queue=audio_frames_queue
+        )
+
+    @staticmethod
+    def create_speech_monitor() -> SpeechStateMonitor:
+        """创建语音监控服务"""
+        return SpeechStateMonitor(
+            audio_frame_queue=audio_frames_queue,
+            user_voice_queue=user_voice_queue,
+        )
+
+    @staticmethod
+    def create_asr_worker(language: str) -> ASRWorker:
+        """创建ASR服务"""
+        return ASRWorker(
+            user_voice_queue=user_voice_queue,
+            transcribed_text_queue=transcribed_text_queue,
+            language=language
+        )
+
+    @staticmethod
+    def create_llm_generator() -> LLMResponseGenerator:
+        """创建LLM文本生成服务"""
+        return LLMResponseGenerator(
+            user_question_queue=transcribed_text_queue,
+            generated_answer_queue=text_input_queue
+        )
+
+    @staticmethod
+    def create_tts_audio_generator(tts_speaker_config: BaseTTSConfig) -> TTSAudioGenerator:
+        """创建TTS音频生成服务"""
+        return TTSAudioGenerator(
+            text_input_queue=text_input_queue,
+            audio_output_queue=audio_output_queue,
+            tts_config=tts_speaker_config
+        )
+
+    @staticmethod
+    def create_audio_player() -> AudioStreamPlayer:
+        """创建音频播放服务"""
+        return AudioStreamPlayer(audio_playing_queue=audio_output_queue)
+
+    @staticmethod
+    def create_tts_config_loader() -> Any:
+        """创建TTS配置加载器的虚拟服务"""
+
+        class TTSConfigLoader:
+            def __init__(self):
+                self.is_ready = False
+                self._running = False
+
+            def start(self):
+                self._running = True
+                self.is_ready = True
+
+            def stop(self):
+                self._running = False
+
+            def is_alive(self):
+                return self._running
+
+        return TTSConfigLoader()
+
+
+def get_service_definitions(system_language: str) -> list:
+    """获取服务定义配置"""
+
+    return [
+        ServiceDefinition(
+            name="tts_config_loader",
+            factory=ServiceFactories.create_tts_config_loader,
+            required=False,
+            startup_timeout=10
+        ),
+
+        ServiceDefinition(
+            name="speech_monitor",
+            factory=ServiceFactories.create_speech_monitor,
+            dependencies=[],
+            health_check=lambda service: hasattr(service, 'is_ready') and service.is_ready
+        ),
+
+        ServiceDefinition(
+            name="asr_worker",
+            factory=lambda: ServiceFactories.create_asr_worker(system_language),
+            dependencies=["speech_monitor"]
+        ),
+
+        ServiceDefinition(
+            name="llm_generator",
+            factory=ServiceFactories.create_llm_generator,
+            dependencies=["asr_worker"]
+        ),
+
+        ServiceDefinition(
+            name="audio_player",
+            factory=ServiceFactories.create_audio_player,
+            dependencies=["llm_generator"]
+        )
+    ]

src/VoiceDialogue/api/core/service_manager.py

@@ -0,0 +1,156 @@
+import logging
+import time
+from dataclasses import dataclass
+from typing import Dict, List, Callable, Any, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ServiceDefinition:
+    """服务定义类，用于配置化管理服务"""
+    name: str
+    factory: Callable[[], Any]
+    dependencies: List[str] = None
+    required: bool = True
+    startup_timeout: int = 30  # 启动超时时间（秒）
+    health_check: Optional[Callable[[Any], bool]] = None
+
+    def __post_init__(self):
+        if self.dependencies is None:
+            self.dependencies = []
+
+
+class ServiceManager:
+    """服务管理器，负责服务的统一启动、停止和监控"""
+
+    def __init__(self):
+        self.services: Dict[str, Any] = {}
+        self.startup_errors: Dict[str, str] = {}
+        self.startup_times: Dict[str, float] = {}
+        self._shutdown_hooks: List[Callable] = []
+
+    def add_shutdown_hook(self, hook: Callable):
+        """添加关闭钩子函数"""
+        self._shutdown_hooks.append(hook)
+
+    def start_service(self, service_def: ServiceDefinition) -> bool:
+        """启动单个服务"""
+        start_time = time.time()
+        try:
+            logger.info(f"正在启动服务: {service_def.name}")
+
+            # 检查依赖服务
+            if not self._check_dependencies(service_def.dependencies):
+                raise RuntimeError(f"服务 {service_def.name} 的依赖服务未就绪")
+
+            service = service_def.factory()
+            service.start()
+
+            # 等待服务就绪
+            if not self._wait_for_service_ready(service, service_def):
+                raise TimeoutError(f"服务 {service_def.name} 启动超时")
+
+            # 健康检查
+            if service_def.health_check and not service_def.health_check(service):
+                raise RuntimeError(f"服务 {service_def.name} 健康检查失败")
+
+            self.services[service_def.name] = service
+            self.startup_times[service_def.name] = time.time() - start_time
+
+            logger.info(f"服务 {service_def.name} 启动成功，耗时: {self.startup_times[service_def.name]:.2f}秒")
+            return True
+
+        except Exception as e:
+            error_msg = f"服务 {service_def.name} 启动失败: {e}"
+            logger.error(error_msg, exc_info=True)
+            self.startup_errors[service_def.name] = error_msg
+
+            if service_def.required:
+                raise RuntimeError(error_msg)
+            return False
+
+    def _check_dependencies(self, dependencies: List[str]) -> bool:
+        """检查依赖服务是否已启动并就绪"""
+        for dep in dependencies:
+            if dep not in self.services:
+                logger.error(f"依赖服务 {dep} 未启动")
+                return False
+            if not self.services[dep].is_ready:
+                logger.error(f"依赖服务 {dep} 未就绪")
+                return False
+        return True
+
+    def _wait_for_service_ready(self, service: Any, service_def: ServiceDefinition) -> bool:
+        """等待服务就绪"""
+        timeout = service_def.startup_timeout
+        start_time = time.time()
+
+        while not service.is_ready and (time.time() - start_time) < timeout:
+            time.sleep(0.1)
+
+        return service.is_ready
+
+    def stop_all_services(self):
+        """停止所有服务"""
+        logger.info("正在停止所有服务...")
+
+        # 执行关闭钩子
+        for hook in self._shutdown_hooks:
+            try:
+                hook()
+            except Exception as e:
+                logger.error(f"执行关闭钩子时发生错误: {e}", exc_info=True)
+
+        # 按启动顺序的逆序停止服务
+        for service_name in reversed(list(self.services.keys())):
+            self._stop_single_service(service_name)
+
+        self.services.clear()
+
+    def _stop_single_service(self, service_name: str):
+        """停止单个服务"""
+        try:
+            service = self.services[service_name]
+            logger.info(f"正在停止服务: {service_name}")
+
+            service.stop()
+
+            # 等待服务停止（最多等待5秒）
+            timeout = 5
+            start_time = time.time()
+            while service.is_alive() and (time.time() - start_time) < timeout:
+                time.sleep(0.1)
+
+            if service.is_alive():
+                logger.warning(f"服务 {service_name} 未能在超时时间内停止")
+            else:
+                logger.info(f"服务 {service_name} 已停止")
+
+        except Exception as e:
+            logger.error(f"停止服务 {service_name} 时发生错误: {e}", exc_info=True)
+
+    def get_service_status(self) -> dict:
+        """获取所有服务状态"""
+        return {
+            "total_services": len(self.services),
+            "startup_errors": len(self.startup_errors),
+            "startup_times": self.startup_times.copy(),
+            "errors": self.startup_errors.copy(),
+            "services": {
+                name: {
+                    "running": service.is_alive(),
+                    "ready": service.is_ready
+                }
+                for name, service in self.services.items()
+            }
+        }
+
+    def get_service(self, name: str) -> Optional[Any]:
+        """获取指定服务实例"""
+        return self.services.get(name)
+
+    def is_service_running(self, name: str) -> bool:
+        """检查服务是否正在运行"""
+        service = self.services.get(name)
+        return service is not None and service.is_alive()

src/VoiceDialogue/services/audio/__init__.py

@@ -0,0 +1,9 @@
+from .aec_audio_capture import EchoCancellingAudioCapture
+from .audio_answer import TTSAudioGenerator
+from .audio_player import AudioStreamPlayer
+
+__all__ = (
+    "EchoCancellingAudioCapture",
+    "TTSAudioGenerator",
+    "AudioStreamPlayer",
+)