diff --git a/app/agent/llm/capability.py b/app/agent/llm/capability.py index d9ebae80..5d81cf46 100644 --- a/app/agent/llm/capability.py +++ b/app/agent/llm/capability.py @@ -670,6 +670,11 @@ class AgentCapabilityManager: def _normalize_provider_name(provider: Optional[str]) -> str: return (provider or "openai").strip().lower() + @staticmethod + def _get_provider_log_name(provider: AudioCapabilityProvider) -> str: + provider_name = getattr(provider, "name", None) + return provider_name if isinstance(provider_name, str) else provider.__class__.__name__ + @classmethod def get_audio_provider(cls, mode: str) -> Optional[AudioCapabilityProvider]: provider_name = cls._normalize_provider_name( @@ -714,17 +719,45 @@ class AgentCapabilityManager: @classmethod def transcribe_audio(cls, content: bytes, filename: str = "input.ogg") -> Optional[str]: + """将语音文件内容转写为文字,并记录能力调用日志。""" provider = cls.get_audio_provider("input") if not provider or not cls.is_audio_input_available(): + logger.info("语音转文字跳过:音频输入能力未启用或 provider 不可用") return None - return provider.transcribe_audio(content=content, filename=filename) + provider_name = cls._get_provider_log_name(provider) + logger.info( + f"语音转文字开始:provider={provider_name}, filename={filename}, " + f"bytes={len(content) if content else 0}" + ) + transcript = provider.transcribe_audio(content=content, filename=filename) + if transcript: + logger.info( + f"语音转文字完成:provider={provider_name}, filename={filename}, " + f"text_len={len(transcript)}" + ) + else: + logger.info( + f"语音转文字无结果:provider={provider_name}, filename={filename}" + ) + return transcript @classmethod def synthesize_speech(cls, text: str) -> Optional[Path]: + """将文字合成为语音文件,并记录能力调用日志。""" provider = cls.get_audio_provider("output") if not provider or not cls.is_audio_output_available(): + logger.info("文字转语音跳过:音频输出能力未启用或 provider 不可用") return None - return provider.synthesize_speech(text=text) + provider_name = cls._get_provider_log_name(provider) + logger.info( + f"文字转语音开始:provider={provider_name}, text_len={len(text) if text else 0}" + ) + output_path = provider.synthesize_speech(text=text) + if output_path: + logger.info(f"文字转语音完成:provider={provider_name}, path={output_path}") + else: + logger.info(f"文字转语音无结果:provider={provider_name}") + return output_path @classmethod def resolve_reply_mode(cls, channel: Optional[str], source: Optional[str]) -> str: diff --git a/tests/test_agent_llm_capability.py b/tests/test_agent_llm_capability.py index 11e4eb5d..63642495 100644 --- a/tests/test_agent_llm_capability.py +++ b/tests/test_agent_llm_capability.py @@ -135,29 +135,43 @@ class AgentCapabilityManagerTest(unittest.TestCase): def test_transcribe_audio_routes_to_input_provider(self): provider = Mock() + provider.name = "mock_audio" provider.is_available_for_audio_input.return_value = True provider.transcribe_audio.return_value = "你好" with patch.object(settings, "LLM_SUPPORT_AUDIO_INPUT", True), patch.object( AgentCapabilityManager, "get_audio_provider", return_value=provider - ): + ), patch.object(capability_module.logger, "info") as log_info: result = AgentCapabilityManager.transcribe_audio(b"audio") self.assertEqual(result, "你好") provider.transcribe_audio.assert_called_once() + self.assertTrue( + any("语音转文字开始" in call.args[0] for call in log_info.call_args_list) + ) + self.assertTrue( + any("语音转文字完成" in call.args[0] for call in log_info.call_args_list) + ) def test_synthesize_speech_routes_to_output_provider(self): provider = Mock() + provider.name = "mock_audio" provider.is_available_for_audio_output.return_value = True provider.synthesize_speech.return_value = Path("/tmp/reply.opus") with patch.object(settings, "LLM_SUPPORT_AUDIO_OUTPUT", True), patch.object( AgentCapabilityManager, "get_audio_provider", return_value=provider - ): + ), patch.object(capability_module.logger, "info") as log_info: result = AgentCapabilityManager.synthesize_speech("你好") self.assertEqual(result, Path("/tmp/reply.opus")) provider.synthesize_speech.assert_called_once_with(text="你好") + self.assertTrue( + any("文字转语音开始" in call.args[0] for call in log_info.call_args_list) + ) + self.assertTrue( + any("文字转语音完成" in call.args[0] for call in log_info.call_args_list) + ) def test_native_voice_reply_supports_channels_with_audio_output(self): """校验 Agent 语音回复渠道支持判断覆盖常见渠道写法。"""