#!/usr/bin/env python3 """ Agent Worker - 主控制器 协调 STT、LLM、TTS、Memory 等组件，实现完整的语音对话流程。 """ import asyncio import logging import time from typing import Optional, Dict, Any from collections import deque from event_bus import EventBus, EventType, EventBusEvent from components.base import BaseComponent, ComponentConfig, ComponentState, STTComponent, LLMComponent, TTSComponent, MemoryComponent from components.stt import create_stt_component, STTComponent from components.llm import create_llm_component, LLMComponent from components.tts import create_tts_component, TTSComponent from components.memory import MemoryComponentImpl from components.audio_play_queue import create_audio_play_queue, AudioPlayQueue from components.vad import create_vad_component, TENVADComponent logger = logging.getLogger(__name__) class AgentWorker(BaseComponent): """ Agent Worker - 主控制器 协调各组件，实现语音对话流程。 """ def __init__(self, config: Dict, event_bus: EventBus): # 先调用父类初始化 super().__init__( config=ComponentConfig( name="agent_worker", enabled=True, priority=0, # 最高优先级 dependencies=["stt", "llm", "tts", "memory"] ), event_bus=event_bus ) # 保存组件配置（与父类 config 区分） self._component_config = config # 组件实例 self.stt: Optional[STTComponent] = None self.llm: Optional[LLMComponent] = None self.tts: Optional[TTSComponent] = None self.memory: Optional[MemoryComponentImpl] = None self.audio_queue: Optional[AudioPlayQueue] = None self.vad: Optional[TENVADComponent] = None # 打断控制 self._is_processing = False self._interrupt_requested = False # 统计 self._total_turns = 0 self._total_errors = 0 async def _do_initialize(self): """初始化 Agent Worker""" logger.info("🚀 初始化 Agent Worker") # 1. 创建各组件 self.stt = create_stt_component(self._component_config, self.event_bus) self.llm = create_llm_component(self._component_config, self.event_bus) self.tts = create_tts_component(self._component_config, self.event_bus) self.memory = MemoryComponentImpl(self._component_config, self.event_bus) # 2. 初始化各组件 await self.stt.initialize() await self.llm.initialize() await self.tts.initialize() await self.memory.initialize() self.audio_queue = create_audio_play_queue(self._component_config, self.event_bus) await self.audio_queue.initialize() self.vad = create_vad_component(self._component_config, self.event_bus) await self.vad.initialize() # 3. 订阅事件 self.subscribe(EventType.USER_SPEECH, self._on_user_speech) self.subscribe(EventType.ASSISTANT_SPEECH, self._on_assistant_speech) self.subscribe(EventType.INTERRUPT, self._on_interrupt) logger.info("✅ Agent Worker 初始化完成") async def _run_loop(self): """主循环 - 启动所有组件""" logger.info("▶️ 启动 Agent Worker") # 启动各组件 await self.stt.start() await self.llm.start() await self.tts.start() await self.memory.start() await self.audio_queue.start() await self.vad.start() self.state = ComponentState.RUNNING logger.info("🎉 Agent Worker 运行中") # 主循环保持运行 while self.state == ComponentState.RUNNING: try: await asyncio.sleep(1) except asyncio.CancelledError: break except Exception as e: logger.error(f"❌ Agent Worker 主循环错误: {e}") await asyncio.sleep(1) async def _on_user_speech(self, event: EventBusEvent): """处理用户语音事件""" # 检查是否需要打断 if self._is_processing and self._interrupt_requested: logger.info("⚡ 用户打断，停止当前回复") await self.audio_queue.interrupt() if self._is_processing: logger.debug("⏸️ 正在处理中，等待打断...") return self._is_processing = True self._interrupt_requested = False text = event.data.get("text", "") logger.info(f"👤 用户: {text[:100]}...") try: # 1. 记录对话历史 self._dialogue_history.append({ "role": "user", "text": text, "timestamp": time.time() }) # 2. 获取记忆上下文 context = await self._get_memory_context() # 3. LLM 生成回复 reply = await self.llm.generate(text, context) # 4. 记录对话历史 self._dialogue_history.append({ "role": "assistant", "text": reply, "timestamp": time.time() }) # 5. 发布回复事件 self.event_bus.publish(EventType.ASSISTANT_SPEECH, { "text": reply, "speaker": "assistant" }, source="agent_worker") # 6. 更新统计 self._total_turns += 1 except Exception as e: logger.error(f"❌ 处理用户语音失败: {e}") self._total_errors += 1 self.event_bus.publish(EventType.SYSTEM_ERROR, { "component": "agent", "error": str(e) }, source="agent_worker") finally: self._is_processing = False async def _on_assistant_speech(self, event: EventBusEvent): """处理助手回复事件 - 执行 TTS 并播放""" text = event.data.get("text", "") logger.info(f"🔊 TTS 合成中: {text[:50]}...") try: # 1. TTS 合成 audio = await self.tts.speak(text) if audio: # 2. 添加到播放队列 await self.audio_queue.play(audio) self.event_bus.publish(EventType.AUDIO_PLAYING, { "text": text, "duration": len(audio) / (16000 * 2) }, source="agent_worker") except Exception as e: logger.error(f"❌ TTS 失败: {e}") self.event_bus.publish(EventType.SYSTEM_ERROR, { "component": "tts", "error": str(e) }, source="agent_worker") async def _on_interrupt(self, event: EventBusEvent): """处理打断事件""" logger.info("⚡ 收到打断请求") self._interrupt_requested = True await self.audio_queue.interrupt() self.event_bus.publish(EventType.AUDIO_STOPPED, { "reason": "interrupt" }, source="agent_worker") async def _get_memory_context(self) -> Dict: """获取记忆上下文""" context = {} try: # 获取最近的人物上下文 ctx = await self.memory.get_context( identity="user", hours=24 ) context["profile"] = ctx.get("profile", {}) context["recent_facts"] = ctx.get("current_facts", [])[:5] except Exception as e: logger.error(f"❌ 获取记忆上下文失败: {e}") return context def get_dialogue_history(self) -> list: """获取对话历史""" return list(self._dialogue_history) def get_stats(self) -> Dict: """获取统计信息""" return { "total_turns": self._total_turns, "total_errors": self._total_errors, "memory_stats": self.memory.get_stats() if self.memory else {} } async def stop(self) -> bool: """停止 Agent Worker""" logger.info("⏹️ 停止 Agent Worker") # 停止各组件 if self.stt: await self.stt.stop() if self.llm: await self.llm.stop() if self.tts: await self.tts.stop() if self.memory: await self.memory.stop() if self.audio_queue: await self.audio_queue.stop() if self.vad: await self.vad.stop() return await super().stop() def is_ready(self) -> bool: """检查是否就绪""" return self.state == ComponentState.RUNNING def health_check(self) -> Dict[str, Any]: """健康检查""" return { "status": "healthy" if self.is_ready() else "stopped", "stt": self.stt.health_check() if self.stt else {"status": "uninitialized"}, "llm": self.llm.health_check() if self.llm else {"status": "uninitialized"}, "tts": self.tts.health_check() if self.tts else {"status": "uninitialized"}, "memory": self.memory.health_check() if self.memory else {"status": "uninitialized"}, "audio_queue": self.audio_queue.health_check() if self.audio_queue else {"status": "uninitialized"}, "vad": self.vad.health_check() if self.vad else {"status": "uninitialized"} } # ========== 入口点 ========== async def main(): """测试入口""" # 配置 config = { "agent": { "stt": {"backend": "funasr", "enabled": True}, "llm": {"backend": "agnes", "enabled": True}, "tts": {"backend": "edge_tts", "enabled": True}, "memory": {"enabled": True}, "audio": {"enabled": True}, "vad": {"enabled": True} } } # 创建事件总线 event_bus = EventBus() # 创建 Agent Worker agent = AgentWorker(config, event_bus) # 启动 await agent.initialize() await agent.start() logger.info("=" * 50) logger.info("📊 Agent Worker 健康状态:") logger.info("=" * 50) for comp, status in agent.health_check().items(): if isinstance(status, dict): logger.info(f" {comp}: {status.get('status', 'unknown')}") else: logger.info(f" {comp}: {status}") logger.info("=" * 50) # 测试对话 await asyncio.sleep(1) # 发布测试事件 event_bus.publish(EventType.USER_SPEECH, { "text": "你好，我叫张三", "speaker": "user" }, source="test") await asyncio.sleep(2) # 停止 await agent.stop() await event_bus.stop() print(f"\n📊 统计: {agent.get_stats()}") if __name__ == "__main__": asyncio.run(main())