From cbfff71d6bffc663f61d626879704066d358c7b6 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 15 Sep 2023 00:33:18 +0800 Subject: [PATCH] add whisper speech to text node and google text to speech node --- src/aios_kernel/__init__.py | 6 +- src/aios_kernel/google_text_to_speech_node.py | 104 ++++++++++++++++ src/aios_kernel/whisper_node.py | 111 ++++++++++++++++++ src/requirements.txt | 1 + 4 files changed, 220 insertions(+), 2 deletions(-) create mode 100644 src/aios_kernel/google_text_to_speech_node.py create mode 100644 src/aios_kernel/whisper_node.py diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index bef117f..2a87b5c 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -4,8 +4,10 @@ from .chatsession import AIChatSession from .agent import AIAgent,AIAgentTemplete,AgentPrompt from .compute_kernel import ComputeKernel,ComputeTask from .compute_node import ComputeNode,LocalComputeNode -from .open_ai_node import OpenAI_ComputeNode +from .open_ai_node import OpenAI_ComputeNode from .role import AIRole,AIRoleGroup from .workflow import Workflow from .bus import AIBus -from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent \ No newline at end of file +from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent +from .whisper_node import WhisperComputeNode +from .google_text_to_speech_node import GoogleTextToSpeechNode diff --git a/src/aios_kernel/google_text_to_speech_node.py b/src/aios_kernel/google_text_to_speech_node.py new file mode 100644 index 0000000..7709eef --- /dev/null +++ b/src/aios_kernel/google_text_to_speech_node.py @@ -0,0 +1,104 @@ + +import os +import asyncio +from asyncio import Queue +import logging + +from google.cloud import texttospeech + +from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType +from .compute_node import ComputeNode + +logger = logging.getLogger(__name__) + + +""" +You need to set the GOOGLE_APPLICATION_CREDENTIALS environment variable when using it. +see:https://cloud.google.com/text-to-speech/docs/before-you-begin +""" + + +class GoogleTextToSpeechNode(ComputeNode): + _instance = None + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + cls._instance = super(GoogleTextToSpeechNode, cls).__new__(cls) + cls._instance.is_start = False + return cls._instance + + def __init__(self): + super().__init__() + if self.is_start is True: + logger.warn("GoogleTextToSpeechNode is already start") + return + + self.is_start = True + self.node_id = "google_text_to_speech_node" + self.task_queue = Queue() + + self.client = texttospeech.TextToSpeechClient() + + self.start() + + def start(self): + async def _run_task_loop(): + while True: + task = await self.task_queue.get() + try: + result = self._run_task(task) + if result is not None: + task.state = ComputeTaskState.DONE + task.result = result + except Exception as e: + logger.error(f"google_text_to_speech_node run task error: {e}") + task.state = ComputeTaskState.ERROR + task.result = ComputeTaskResult() + task.result.set_from_task(task) + task.result.worker_id = self.node_id + task.result.result_str = str(e) + + asyncio.create_task(_run_task_loop()) + + def _run_task(self, task: ComputeTask): + task.state = ComputeTaskState.RUNNING + language_code = task.params["language_code"] + text = task.params["text"] + + synthesis_input = texttospeech.SynthesisInput(text=text) + voice = texttospeech.VoiceSelectionParams(language_code=language_code, + ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL) + + audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3) + + response = self.client.synthesize_speech(input=synthesis_input, voice=voice, audio_config=audio_config) + + result = ComputeTaskResult() + result.set_from_task(task) + result.worker_id = self.node_id + result.result = response.audio_content + return result + + async def push_task(self, task: ComputeTask, proiority: int = 0): + logger.info(f"google_text_to_speech_node push task: {task.display()}") + self.task_queue.put_nowait(task) + + async def remove_task(self, task_id: str): + pass + + def get_task_state(self, task_id: str): + pass + + def display(self) -> str: + return f"GoogleTextToSpeechNode: {self.node_id}" + + def get_capacity(self): + return 0 + + def is_support(self, task_type: ComputeTaskType) -> bool: + if task_type == ComputeTaskType.TEXT_2_VOICE: + return True + return False + + def is_local(self) -> bool: + return False diff --git a/src/aios_kernel/whisper_node.py b/src/aios_kernel/whisper_node.py new file mode 100644 index 0000000..b036ad5 --- /dev/null +++ b/src/aios_kernel/whisper_node.py @@ -0,0 +1,111 @@ +from asyncio import Queue +import asyncio +import openai +import os +import logging + +from .compute_node import ComputeNode +from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType + +logger = logging.getLogger(__name__) + + +class WhisperComputeNode(ComputeNode): + _instance = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance.is_start = False + return cls._instance + + def __init__(self) -> None: + super().__init__() + if self.is_start is True: + logger.warn("WhisperComputeNode is already start") + return + + self.is_start = True + self.node_id = "whisper_node" + self.enable = True + self.task_queue = Queue() + self.open_api_key = None + + if self.open_api_key is None and os.getenv("OPENAI_API_KEY") is not None: + self.open_api_key = os.getenv("OPENAI_API_KEY") + + if self.open_api_key is None: + raise Exception("WhisperComputeNode open_api_key is None") + + self.start() + + def start(self): + async def _run_task_loop(): + while True: + task = await self.task_queue.get() + try: + result = self._run_task(task) + if result is not None: + task.state = ComputeTaskState.DONE + task.result = result + except Exception as e: + logger.error(f"whisper_node run task error: {e}") + task.state = ComputeTaskState.ERROR + task.result = ComputeTaskResult() + task.result.set_from_task(task) + task.result.worker_id = self.node_id + task.result.result_str = str(e) + + asyncio.create_task(_run_task_loop()) + + def _run_task(self, task: ComputeTask): + task.state = ComputeTaskState.RUNNING + prompt = task.params["prompt"] + response_format = None + if "response_format" in task.params: + response_format = task.params["response_format"] + temperature = None + if "temperature" in task.params: + temperature = task.params["temperature"] + language = None + if "language" in task.params: + language = task.params["language"] + file = task.params["file"] + + resp = openai.Audio.transcribe("whisper-1", + file, + self.open_api_key, + prompt=prompt, + response_format=response_format, + temperature=temperature, + language=language) + result = ComputeTaskResult() + result.set_from_task(task) + result.worker_id = self.node_id + result.result_str = resp["text"] + result.result = resp + return result + + async def push_task(self, task: ComputeTask, proiority: int = 0): + logger.info(f"whisper_node push task: {task.display()}") + self.task_queue.put_nowait(task) + + async def remove_task(self, task_id: str): + pass + + def get_task_state(self, task_id: str): + pass + + def display(self) -> str: + return f"WhisperComputeNode: {self.node_id}" + + def get_capacity(self): + return 0 + + def is_support(self, task_type: ComputeTaskType) -> bool: + if task_type == ComputeTaskType.VOICE_2_TEXT: + return True + return False + + def is_local(self) -> bool: + return False diff --git a/src/requirements.txt b/src/requirements.txt index 605dc72..7de369e 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -6,3 +6,4 @@ Pillow==10.0.0 prompt_toolkit==3.0.39 stability_sdk==0.8.4 toml==0.10.2 +google-cloud-texttospeech==2.14.1