add whisper speech to text node and google text to speech node

src/aios_kernel/__init__.py

@@ -9,3 +9,5 @@ from .role import AIRole,AIRoleGroup
 from .workflow import Workflow
 from .bus import AIBus
 from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent
+from .whisper_node import WhisperComputeNode
+from .google_text_to_speech_node import GoogleTextToSpeechNode

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

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

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