Merge pull request #101 from wugren/MVP

TTS and ASR function implemented based on openai api

src/aios_kernel/__init__.py

@@ -6,6 +6,7 @@ from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeT
 from .compute_node import ComputeNode,LocalComputeNode
 from .open_ai_node import OpenAI_ComputeNode
 from .role import AIRole,AIRoleGroup
+from .storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem
 from .workflow import Workflow
 from .bus import AIBus
 from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent,PaintEnvironment
@@ -15,8 +16,7 @@ from .google_text_to_speech_node import GoogleTextToSpeechNode
 from .tunnel import AgentTunnel
 from .tg_tunnel import TelegramTunnel
 from .email_tunnel import EmailTunnel
-from .storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem
+from .contact_manager import ContactManager,Contact,FamilyMember
-from .contact_manager import ContactManager,Contact,FamilyMember  
 from .text_to_speech_function import TextToSpeechFunction
 from .image_2_text_function import Image2TextFunction
 from .workspace_env import ShellEnvironment

src/aios_kernel/asr_function.py

@@ -0,0 +1,55 @@
+import logging
+from typing import Dict
+
+from aios_kernel import ComputeKernel
+from aios_kernel.ai_function import AIFunction
+
+logger = logging.getLogger(__name__)
+
+
+class AsrFunction(AIFunction):
+    def __init__(self):
+        self.func_id = "speech_to_text"
+        self.description = "语音识别，将语音转换为文字"
+
+    def get_name(self) -> str:
+        return self.func_id
+
+    def get_description(self) -> str:
+        return self.description
+
+    def get_parameters(self) -> Dict:
+        return {
+            "type": "object",
+            "properties": {
+                "audio_file": {"type": "string", "description": "音频文件路径"},
+                "model": {"type": "string", "description": "识别模型", "enum": ["openai-whisper"]},
+                "prompt": {"type": "string", "description": "提示语句,可以为None"},
+                "response_format": {"type": "string", "description": "返回格式", "enum": ["text", "json", "srt", "verbose_json", "vtt"]},
+            }
+        }
+
+    async def execute(self, **kwargs) -> str:
+        logger.info(f"execute asr function: {kwargs}")
+
+        audio_file = kwargs.get("audio_file")
+        model = kwargs.get("model")
+        prompt = kwargs.get("prompt")
+        response_format = kwargs.get("response_format")
+        if response_format is None:
+            response_format = "text"
+
+        result = await ComputeKernel.get_instance().do_speech_to_text(audio_file, model, prompt, response_format)
+        if result is not None:
+            return f"exec speech_to_text Ok. {response_format} is\n```\n{result.result_str}\n```"
+        else:
+            return "exec speech_to_text failed"
+
+    def is_local(self) -> bool:
+        return True
+
+    def is_in_zone(self) -> bool:
+        return True
+
+    def is_ready_only(self) -> bool:
+        return False

src/aios_kernel/compute_kernel.py

@@ -77,7 +77,7 @@ class ComputeKernel:
         if len(support_nodes) < 1:
             logger.warning(f"task {task.display()} is not support by any compute node")
             return None
-        
+
         # hit a random node with weight
         hit_pos = random.randint(0, total_weights - 1)
         for i in range(min(len(support_nodes) - 1, hit_pos), -1, -1):
@@ -126,8 +126,8 @@ class ComputeKernel:
         task_req.set_llm_params(prompt,resp_mode,mode_name, max_token,inner_functions)
         self.run(task_req)
         return task_req
-    
+
-    async def _wait_task(self,task_req:ComputeTask, timeout=60)->ComputeTaskResult:        
+    async def _wait_task(self,task_req:ComputeTask, timeout=60)->ComputeTaskResult:
         async def check_timer():
             check_times = 0
             while True:
@@ -136,7 +136,7 @@ class ComputeKernel:
 
                 if task_req.state == ComputeTaskState.ERROR:
                     break
-                
+
                 if timeout is not None and check_times >= timeout*2:
                     task_req.state = ComputeTaskState.ERROR
                     break
@@ -181,7 +181,7 @@ class ComputeKernel:
         task_req.set_image_embedding_params(input,model_name)
         self.run(task_req)
         return task_req
-    
+
     async def do_image_embedding(self,input:ObjectID,model_name:Optional[str] = None) -> [float]:
         task_req = self.image_embedding(input,model_name)
         task_result = await self._wait_task(task_req)
@@ -197,7 +197,8 @@ class ComputeKernel:
                        gender: Optional[str] = None,
                        age: Optional[str] = None,
                        voice_name: Optional[str] = None,
-                       tone: Optional[str] = None):
+                       tone: Optional[str] = None,
+                       model_name: Optional[str] = None):
         task_req = ComputeTask()
         task_req.params["text"] = input
         task_req.params["language_code"] = language_code
@@ -205,6 +206,7 @@ class ComputeKernel:
         task_req.params["age"] = age
         task_req.params["voice_name"] = voice_name
         task_req.params["tone"] = tone
+        task_req.params["model_name"] = model_name
         task_req.task_type = ComputeTaskType.TEXT_2_VOICE
         self.run(task_req)
 
@@ -213,6 +215,24 @@ class ComputeKernel:
         if task_req.state == ComputeTaskState.DONE:
             return task_result.result
 
+    async def do_speech_to_text(self,
+                                audio: str,
+                                model: str,
+                                prompt: Optional[str],
+                                response_format: Optional[str]):
+        task_req = ComputeTask()
+        task_req.params["file"] = audio
+        task_req.params["model_name"] = model
+        task_req.params["prompt"] = prompt
+        task_req.params["response_format"] = response_format
+        task_req.task_type = ComputeTaskType.VOICE_2_TEXT
+
+        self.run(task_req)
+
+        task_result = await self._wait_task(task_req)
+
+        if task_req.state == ComputeTaskState.DONE:
+            return task_result
 
     def text_2_image(self, prompt:str, model_name:Optional[str] = None, negative_prompt = None):
         task = ComputeTask()

src/aios_kernel/google_text_to_speech_node.py

@@ -117,9 +117,18 @@ class GoogleTextToSpeechNode(ComputeNode):
     def _run_task(self, task: ComputeTask):
         task.state = ComputeTaskState.RUNNING
         language_code = task.params["language_code"]
+        if language_code is None:
+            language_code = "en"
+
         text = task.params["text"]
         voice_name = task.params["voice_name"]
+        if voice_name is None:
+            voice_name = "default"
+
         gender = task.params["gender"]
+        if gender is None:
+            gender = "female"
+
         age = task.params["age"]
 
         if language_code == "zh":

src/aios_kernel/openai_tts_node.py

@@ -0,0 +1,120 @@
+import asyncio
+import io
+import logging
+import os
+from asyncio import Queue
+
+from openai import AsyncOpenAI
+
+from aios_kernel import ComputeNode, ComputeTask, ComputeTaskState, ComputeTaskResult, ComputeTaskType, AIStorage
+
+logger = logging.getLogger(__name__)
+
+
+class OpenAITTSComputeNode(ComputeNode):
+    _instance = None
+
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            cls._instance = cls()
+        return cls._instance
+
+    def __init__(self):
+        super().__init__()
+        self.is_start = False
+        self.node_id = "openai_tts_node"
+        self.task_queue = Queue()
+        self.voice_list = {
+            "female": ["nova", "shimmer"],
+            "man": ["alloy", "echo", "fable", "onyx"]
+        }
+        if os.getenv("OPENAI_API_KEY") is not None:
+            self.openai_api_key = os.getenv("OPENAI_API_KEY")
+        else:
+            self.openai_api_key = AIStorage.get_instance().get_user_config().get_value("openai_api_key")
+
+        self.start()
+
+    def start(self):
+        if self.is_start is True:
+            logger.warn("OpenAITTSComputeNode is already start")
+            return
+        self.is_start = True
+
+        async def _run_task_loop():
+            while True:
+                task = await self.task_queue.get()
+                try:
+                    result = await self._run_task(task)
+                    if result is not None:
+                        task.state = ComputeTaskState.DONE
+                        task.result = result
+                except Exception as e:
+                    logger.error(f"openai_tts_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())
+
+    async def _run_task(self,task: ComputeTask):
+        task.state = ComputeTaskState.RUNNING
+        text = task.params["text"]
+        voice_name = task.params["voice_name"]
+        if voice_name is None:
+            voice_name = "default"
+        gender = task.params["gender"]
+        if gender is None:
+            gender = "female"
+
+        voice_list = self.voice_list[gender]
+        voice = voice_list[hash(voice_name)%len(voice_list)]
+
+        model_name = task.params['model_name']
+        if model_name is None:
+            model_name = 'tts-1'
+
+        client = AsyncOpenAI(api_key=self.openai_api_key)
+
+        response = await client.audio.speech.create(model=model_name, voice=voice, input=text)
+
+        cache = io.BytesIO()
+        async for data in await response.aiter_bytes():
+            cache.write(data)
+
+        cache.seek(0)
+
+        result = ComputeTaskResult()
+        result.set_from_task(task)
+        result.worker_id = self.node_id
+        result.result = cache.read()
+        return result
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
+        logger.info(f"openai_tts_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"OpenAITTSComputeNode: {self.node_id}"
+
+    def get_capacity(self):
+        return 0
+
+    def is_support(self, task: ComputeTask) -> bool:
+        if task.task_type == ComputeTaskType.TEXT_2_VOICE:
+            if task.params['model_name'] is None or task.params['model_name'] == 'tts-1' or task.params['model_name'] == 'tts-1-hd':
+                return True
+        return False
+
+
+    def is_local(self) -> bool:
+        return False

src/aios_kernel/script_to_speech_function.py

@@ -0,0 +1,107 @@
+import io
+import logging
+import os
+import random
+from pathlib import Path
+from typing import Dict
+
+from aios_kernel import ComputeKernel, AIStorage
+from aios_kernel.ai_function import AIFunction
+
+from pydub import AudioSegment
+
+logger = logging.getLogger(__name__)
+
+
+class ScriptToSpeechFunction(AIFunction):
+    def __init__(self):
+        self.func_id = "script_to_speech"
+        self.description = "根据输入的剧本生成音频文件，成功时会返回音频文件路径"
+        self.speech_path = os.path.join(AIStorage.get_instance().get_myai_dir(), "tts")
+        Path(self.speech_path).mkdir(exist_ok=True)
+
+    def get_name(self) -> str:
+        return self.func_id
+
+    def get_description(self) -> str:
+        return self.description
+
+    def get_parameters(self) -> Dict:
+        return {
+            "type": "object",
+            "properties": {
+                "language": {"type": "string", "description": "演播语言", "enum": ["zh", "en"]},
+                "model": {"type": "string", "description": "演播模型", "enum": ["tts-1", "tts-1-hd"]},
+                "roles": {"type": "array", "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string", "description": "角色名字"},
+                        "gender": {"type": "string", "description": "角色性别", "enum": ["man", "female"]},
+                        "age": {"type": "string", "description": "年龄", "enum": ["child", "adult"]},
+                    }}},
+                "lines": {"type": "array", "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string", "description": "角色名字"},
+                        "tone": {"type": "string", "description": "演播情感",
+                                 "enum": ["happy", "sad", "angry", "fear", "disgust", "surprise", "neutral"]},
+                        "text": {"type": "string", "description": "台词"},
+                    }
+                }}
+            }
+        }
+
+    async def execute(self, **kwargs) -> str:
+        logger.info(f"execute text_to_speech function: {kwargs}")
+
+        language = kwargs.get("language")
+        if language is None:
+            language = "zh"
+        model = kwargs.get("model")
+        roles = kwargs.get("roles")
+        lines = kwargs.get("lines")
+
+        audio = None
+        for line in lines:
+            name = line.get("name")
+            tone = line.get("tone")
+            text = line.get("text")
+            gender = None
+            age = None
+            for role in roles:
+                role_name = role.get("name")
+                if role_name == name:
+                    gender = role.get("gender")
+                    age = role.get("age")
+                    break
+            i = 0
+            while i < 3:
+                try:
+                    data = await ComputeKernel.get_instance().do_text_to_speech(text, language, gender, age, name, tone, model_name=model)
+                    if audio is None:
+                        audio = AudioSegment.from_mp3(io.BytesIO(data))
+                    else:
+                        audio = audio + AudioSegment.from_mp3(io.BytesIO(data))
+                    break
+                except Exception as e:
+                    logger.error(f"do_text_to_speech failed: {e}")
+                    i += 1
+                    continue
+
+        if audio is not None:
+            path = os.path.join(self.speech_path, "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))))
+            audio.export(path, format="mp3")
+            return "exec text_to_speech OK，speech file store at ```{}```".format(path)
+        else:
+            return "exec text_to_speech failed"
+
+    def is_local(self) -> bool:
+        return True
+
+    def is_in_zone(self) -> bool:
+        return True
+
+    def is_ready_only(self) -> bool:
+        return False
+
+

src/aios_kernel/text_to_speech_function.py

@@ -2,19 +2,23 @@ import io
 import logging
 import os
 import random
+from pathlib import Path
 from typing import Dict
 
-from aios_kernel import ComputeKernel
+from aios_kernel import ComputeKernel, AIStorage
 from aios_kernel.ai_function import AIFunction
 
 from pydub import AudioSegment
 
 logger = logging.getLogger(__name__)
 
+
 class TextToSpeechFunction(AIFunction):
     def __init__(self):
         self.func_id = "text_to_speech"
-        self.description = "根据输入的剧本生成音频文件，成功时会返回音频文件路径"
+        self.description = "根据输入的文本生成音频文件，成功时会返回音频文件路径"
+        self.speech_path = os.path.join(AIStorage.get_instance().get_myai_dir(), "tts")
+        Path(self.speech_path).mkdir(exist_ok=True)
 
     def get_name(self) -> str:
         return self.func_id
@@ -27,22 +31,8 @@ class TextToSpeechFunction(AIFunction):
             "type": "object",
             "properties": {
                 "language": {"type": "string", "description": "演播语言", "enum": ["zh", "en"]},
-                "roles": {"type": "array", "items": {
+                "model": {"type": "string", "description": "演播模型", "enum": ["tts-1", "tts-1-hd"]},
-                    "type": "object",
+                "text": {"type": "string", "description": "文本内容"}
-                    "properties": {
-                        "name": {"type": "string", "description": "角色名字"},
-                        "gender": {"type": "string", "description": "角色性别", "enum": ["man", "female"]},
-                        "age": {"type": "string", "description": "年龄", "enum": ["child", "adult"]},
-                    }}},
-                "lines": {"type": "array", "items": {
-                    "type": "object",
-                    "properties": {
-                        "name": {"type": "string", "description": "角色名字"},
-                        "tone": {"type": "string", "description": "演播情感",
-                                 "enum": ["happy", "sad", "angry", "fear", "disgust", "surprise", "neutral"]},
-                        "text": {"type": "string", "description": "台词"},
-                    }
-                }}
             }
         }
 
@@ -51,41 +41,27 @@ class TextToSpeechFunction(AIFunction):
 
         language = kwargs.get("language")
         if language is None:
-            language = "zh"
+            language = "en"
-        roles = kwargs.get("roles")
+        model = kwargs.get("model")
-        lines = kwargs.get("lines")
+        text = kwargs.get("text")
 
-        audio = None
+        i = 0
-        for line in lines:
+        while i < 3:
-            name = line.get("name")
+            try:
-            tone = line.get("tone")
+                data = await ComputeKernel.get_instance().do_text_to_speech(text, language, None, None, None, None,
-            text = line.get("text")
+                                                                            model_name=model)
-            gender = None
+                if data is not None:
-            age = None
+                    audio = AudioSegment.from_mp3(io.BytesIO(data))
-            for role in roles:
+                break
-                role_name = role.get("name")
+            except Exception as e:
-                if role_name == name:
+                logger.error(f"do_text_to_speech failed: {e}")
-                    gender = role.get("gender")
+                i += 1
-                    age = role.get("age")
+                continue
-                    break
-            i = 0
-            while i < 3:
-                try:
-                    data = await ComputeKernel.get_instance().do_text_to_speech(text, language, gender, age, name, tone)
-                    if audio is None:
-                        audio = AudioSegment.from_mp3(io.BytesIO(data))
-                    else:
-                        audio = audio + AudioSegment.from_mp3(io.BytesIO(data))
-                    break
-                except Exception as e:
-                    logger.error(f"do_text_to_speech failed: {e}")
-                    i += 1
-                    continue
 
         if audio is not None:
-            path = os.path.join(os.path.realpath(os.curdir), "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))))
+            path = os.path.join(self.speech_path, "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))))
             audio.export(path, format="mp3")
-            return "exec text_to_speech OK，speech file store at {}".format(path)
+            return "exec text_to_speech OK，speech file store at ```{}```".format(path)
         else:
             return "exec text_to_speech failed"
 

src/aios_kernel/whisper_node.py

@@ -1,55 +1,77 @@
+import io
+import json
 from asyncio import Queue
 import asyncio
 import openai
 import os
 import logging
+import srt
+import webvtt
 
+from openai import AsyncOpenAI
+from openai.cli._progress import BufferReader
+from pydub import AudioSegment
+from datetime import timedelta
+
+from . import AIStorage
 from .compute_node import ComputeNode
 from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
 
 logger = logging.getLogger(__name__)
 
+SECONDS_IN_HOUR = 3600
+SECONDS_IN_MINUTE = 60
+HOURS_IN_DAY = 24
+MICROSECONDS_IN_MILLISECOND = 1000
+
+
+def timedelta_to_vtt_timestamp(timedelta_timestamp):
+    hrs, secs_remainder = divmod(timedelta_timestamp.seconds, SECONDS_IN_HOUR)
+    hrs += timedelta_timestamp.days * HOURS_IN_DAY
+    mins, secs = divmod(secs_remainder, SECONDS_IN_MINUTE)
+    msecs = timedelta_timestamp.microseconds // MICROSECONDS_IN_MILLISECOND
+    return "%02d:%02d:%02d.%03d" % (hrs, mins, secs, msecs)
+
 
 class WhisperComputeNode(ComputeNode):
     _instance = None
 
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super().__new__(cls)
+            cls._instance = cls()
-            cls._instance.is_start = False
         return cls._instance
 
     def __init__(self) -> None:
         super().__init__()
-        if self.is_start is True:
+        self.is_start = False
-            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:
+        if os.getenv("OPENAI_API_KEY") is not None:
-            self.open_api_key = os.getenv("OPENAI_API_KEY")
+            self.openai_api_key = os.getenv("OPENAI_API_KEY")
-
+        else:
-        if self.open_api_key is None:
+            self.openai_api_key = AIStorage.get_instance().get_user_config().get_value("openai_api_key")
-            raise Exception("WhisperComputeNode open_api_key is None")
 
         self.start()
 
     def start(self):
+        if self.is_start is True:
+            logger.warn("WhisperComputeNode is already start")
+            return
+        self.is_start = True
         async def _run_task_loop():
             while True:
                 task = await self.task_queue.get()
                 try:
-                    result = self._run_task(task)
+                    result = await 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}")
+                    logger.exception(e)
                     task.state = ComputeTaskState.ERROR
                     task.result = ComputeTaskResult()
                     task.result.set_from_task(task)
@@ -58,7 +80,7 @@ class WhisperComputeNode(ComputeNode):
 
         asyncio.create_task(_run_task_loop())
 
-    def _run_task(self, task: ComputeTask):
+    async def _run_task(self, task: ComputeTask):
         task.state = ComputeTaskState.RUNNING
         prompt = task.params["prompt"]
         response_format = None
@@ -72,19 +94,114 @@ class WhisperComputeNode(ComputeNode):
             language = task.params["language"]
         file = task.params["file"]
 
-        resp = openai.Audio.transcribe("whisper-1",
+        client = AsyncOpenAI(api_key=self.openai_api_key)
-                                       file,
+
-                                       self.open_api_key,
+        if os.path.getsize(file) > 25 * 1024 * 1024:
-                                       prompt=prompt,
+            audio = AudioSegment.from_file(file)
-                                       response_format=response_format,
+            text = ""
-                                       temperature=temperature,
+            results = []
-                                       language=language)
+            latest_resp = None
-        result = ComputeTaskResult()
+            step = 10 * 60 * 1000
-        result.set_from_task(task)
+            for i in range(0, len(audio), step):
-        result.worker_id = self.node_id
+                if i + step < len(audio):
-        result.result_str = resp["text"]
+                    chunk = audio[i:i + step]
-        result.result = resp
+                else:
-        return result
+                    chunk = audio[i:]
+                seg_file = io.BytesIO()
+                chunk.export(seg_file, format="mp3")
+                seg_file.seek(0)
+
+                resp = await client.audio.transcriptions.create(model="whisper-1",
+                                                   file = ("test.mp3", seg_file),
+                                                   language=language,
+                                                   temperature=temperature,
+                                                   prompt=prompt,
+                                                   response_format=response_format)
+                if response_format == "json":
+                    if text == "":
+                        text = resp.text
+                    else:
+                        text += "," + resp.text
+                elif response_format == "text":
+                    if text == "":
+                        text = resp
+                    else:
+                        text += "," + resp
+                elif response_format == "verbose_json":
+                    if text == "":
+                        text = resp.text
+                    else:
+                        text += "," + resp.text
+                    results.extend(resp.segments)
+                elif response_format == "srt":
+                    srt_list = list(srt.parse(resp))
+                    for item in srt_list:
+                        item.start += timedelta(milliseconds=i)
+                        item.end += timedelta(milliseconds=i)
+                        results.append(item)
+                elif response_format == "vtt":
+                    vtt = webvtt.read_buffer(io.StringIO(resp))
+                    for caption in vtt.captions:
+                        start = timedelta_to_vtt_timestamp(
+                            srt.srt_timestamp_to_timedelta(caption.start) + timedelta(milliseconds=i))
+                        end = timedelta_to_vtt_timestamp(
+                            srt.srt_timestamp_to_timedelta(caption.end) + timedelta(milliseconds=i))
+                        results.append(webvtt.Caption(start, end, caption.text))
+                else:
+                    raise Exception(f"not support response_format: {response_format}")
+
+                latest_resp = resp
+
+            result = ComputeTaskResult()
+            result.set_from_task(task)
+            result.worker_id = self.node_id
+            if response_format == "text":
+                result.result_str = text
+                result.result = text
+            elif response_format == "json":
+                result.result_str = json.dumps({"text": text})
+                resp.text = text
+                result.result = resp
+            elif response_format == "verbose_json":
+                result.result_str = json.dumps({"text": text, "segments": results})
+                latest_resp.text = text
+                latest_resp.segments = results
+                result.result = latest_resp
+            elif response_format == "srt":
+                result.result_str = srt.compose(results)
+                result.result = result.result_str
+            elif response_format == "vtt":
+                vtt = webvtt.WebVTT()
+                vtt.captions.extend(results)
+                f = io.StringIO()
+                vtt.write(f)
+                f.seek(0)
+                result.result_str = f.read()
+                result.result = result.result_str
+            return result
+        else:
+            with open(file, "rb") as file_reader:
+                buffer_reader = BufferReader(file_reader.read(), desc="Upload progress")
+
+            resp = await client.audio.transcriptions.create(model="whisper-1",
+                                               file = (file, buffer_reader),
+                                               language=language,
+                                               temperature=temperature,
+                                               prompt=prompt,
+                                               response_format=response_format)
+            result = ComputeTaskResult()
+            result.set_from_task(task)
+            result.worker_id = self.node_id
+            if response_format == "json":
+                result.result_str = json.dumps({"text": resp.text})
+            elif response_format == "verbose_json":
+                result.result_str = json.dumps({"text": resp.text, "segments": resp.segments})
+            elif response_format == "srt" or response_format == "vtt" or response_format == "text":
+                result.result_str = resp
+            else:
+                raise Exception(f"not support response_format: {response_format}")
+            result.result = resp
+            return result
 
     async def push_task(self, task: ComputeTask, proiority: int = 0):
         logger.info(f"whisper_node push task: {task.display()}")
@@ -102,9 +219,10 @@ class WhisperComputeNode(ComputeNode):
     def get_capacity(self):
         return 0
 
-    def is_support(self, task_type: ComputeTaskType) -> bool:
+    def is_support(self, task: ComputeTask) -> bool:
-        if task_type == ComputeTaskType.VOICE_2_TEXT:
+        if task.task_type == ComputeTaskType.VOICE_2_TEXT:
-            return True
+            if task.params['model_name'] is None or task.params['model_name'] == 'openai-whisper':
+                return True
         return False
 
     def is_local(self) -> bool:

src/aios_kernel/workflow_env.py

@@ -8,7 +8,7 @@ import threading
 import logging
 from typing import Optional
 
-from .text_to_speech_function import TextToSpeechFunction
+from .script_to_speech_function import ScriptToSpeechFunction
 from .image_2_text_function import Image2TextFunction
 from .compute_kernel import ComputeKernel, ComputeTaskResultCode
 from .environment import Environment,EnvironmentEvent
@@ -80,7 +80,7 @@ class CalenderEnvironment(Environment):
                                         "update event in calender",
                                         self._update_event,update_param))
 
-        
+
         #maybe this function should be in other env?
         paint_param = {
             "prompt": "A description of the content of the painting",
@@ -89,18 +89,18 @@ class CalenderEnvironment(Environment):
         self.add_ai_function(SimpleAIFunction("paint",
                                         "Draw a picture according to the description",
                                         self._paint,paint_param))
-        
+
         self.add_ai_function(SimpleAIFunction("get_contact",
                                         "get contact info",
                                         self._get_contact,{"name":"name of contact"}))
-        
+
         self.add_ai_function(SimpleAIFunction("set_contact",
                                         "set contact info",
                                         self._set_contact,{"name":"name of contact","contact_info":"A json to descrpit contact"}))
-                                        
+
-        
+
-        
+
-        
+
         #self.add_ai_function(SimpleAIFunction("user_confirm",
         #                                      "user confirm",
         #                                      self._user_confirm))
@@ -169,10 +169,10 @@ class CalenderEnvironment(Environment):
                 _event["location"] = row[5]
                 _event["details"] = row[6]
                 result[row[0]] = _event
-            
+
             if not have_result:
                 return "No event."
-            
+
             return json.dumps(result, indent=4, sort_keys=True)
 
     async def _update_event(self,event_id, new_title=None, new_participants=None, new_location=None, new_details=None ,start_time=None, end_time=None):
@@ -230,7 +230,7 @@ class CalenderEnvironment(Environment):
 
     def _do_get_value(self,key:str) -> Optional[str]:
         return None
-    
+
     async def _get_contact(self,name:str) -> str:
         cm = ContactManager.get_instance()
         contact : Contact = cm.find_contact_by_name(name)
@@ -302,7 +302,7 @@ class CalenderEnvironment(Environment):
         formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
         return formatted_time
 
-    
+
     async def _paint(self, prompt, model_name = None) -> str:
         result = await ComputeKernel.get_instance().do_text_2_image(prompt, model_name)
         if result.result_code == ComputeTaskResultCode.ERROR:
@@ -344,7 +344,7 @@ class WorkflowEnvironment(Environment):
         self.db_file = db_file
         self.local = threading.local()
         self.table_name = "WorkflowEnv_" + env_id
-        self.add_ai_function(TextToSpeechFunction())
+        self.add_ai_function(ScriptToSpeechFunction())
         self.add_ai_function(Image2TextFunction())
 
 
@@ -418,4 +418,4 @@ class WorkflowEnvironment(Environment):
             logging.error(f"Error occurred while update env{self.env_id}.{key} ,error:{e}")
 
     def get_functions(self):
-        pass
+        pass

src/requirements.txt

@@ -139,4 +139,6 @@ stability_sdk
 sentence-transformers==2.2.2
 tiktoken
 markdown
-PyPDF2
+PyPDF2
+srt==3.5.3
+webvtt-py==0.4.6

src/service/aios_shell/aios_shell.py

@@ -20,6 +20,8 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import WordCompleter
 from prompt_toolkit.styles import Style
 
+from aios_kernel.openai_tts_node import OpenAITTSComputeNode
+
 directory = os.path.dirname(__file__)
 sys.path.append(directory + '/../../')
 
@@ -74,8 +76,8 @@ class AIOS_Shell:
         user_config.add_user_config("shell.current","last opened target and topic",True,"default@Jarvis")
         proxy.declare_user_config()
 
-        google_text_to_speech = GoogleTextToSpeechNode.get_instance()
+        # google_text_to_speech = GoogleTextToSpeechNode.get_instance()
-        google_text_to_speech.declare_user_config()
+        # google_text_to_speech.declare_user_config()
 
         Local_Stability_ComputeNode.declare_user_config()
 
@@ -93,8 +95,8 @@ class AIOS_Shell:
         if a_workflow is not None:
             bus.register_message_handler(target_id,a_workflow._process_msg)
             return True
-        
+
-        a_contact = await ContactManager.get_instance().get_contact(target_id)
+        a_contact = ContactManager.get_instance().find_contact_by_name(target_id)
         if a_contact is not None:
             bus.register_message_handler(target_id,a_contact._process_msg)
             return True
@@ -143,6 +145,12 @@ class AIOS_Shell:
             return False
         ComputeKernel.get_instance().add_compute_node(open_ai_node)
 
+        whisper_node = WhisperComputeNode.get_instance()
+        ComputeKernel.get_instance().add_compute_node(whisper_node);
+
+        openai_tts_node = OpenAITTSComputeNode.get_instance()
+        ComputeKernel.get_instance().add_compute_node(openai_tts_node)
+
         llama_nodes = ComputeNodeConfig.get_instance().initial()
         for llama_node in llama_nodes:
             llama_node.start()
@@ -158,41 +166,41 @@ class AIOS_Shell:
                 await AIStorage.get_instance().set_feature_init_result("llama",False)
 
 
-        if await AIStorage.get_instance().is_feature_enable("aigc"):
+        # if await AIStorage.get_instance().is_feature_enable("aigc"):
-            try:
+            # try:
-                google_text_to_speech_node = GoogleTextToSpeechNode.get_instance()
+            #     google_text_to_speech_node = GoogleTextToSpeechNode.get_instance()
-                google_text_to_speech_node.init()
+            #     google_text_to_speech_node.init()
-                ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node)
+            #     ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node)
-            except Exception as e:
+            # except Exception as e:
-                logger.error(f"google text to speech node initial failed! {e}")
+            #     logger.error(f"google text to speech node initial failed! {e}")
-                await AIStorage.get_instance.set_feature_init_result("aigc",False)
+            #     await AIStorage.get_instance.set_feature_init_result("aigc",False)
 
             # stability_api_node = Stability_ComputeNode()
             # if await stability_api_node.initial() is not True:
             #     logger.error("stability api node initial failed!")
             # ComputeKernel.get_instance().add_compute_node(stability_api_node)
 
-        
+
-        
+
         local_st_text_compute_node = LocalSentenceTransformer_Text_ComputeNode()
         if local_st_text_compute_node.initial() is not True:
             logger.error("local sentence transformer text embedding node initial failed!")
         else:
             ComputeKernel.get_instance().add_compute_node(local_st_text_compute_node)
-            
+
         local_st_image_compute_node = LocalSentenceTransformer_Image_ComputeNode()
         if local_st_image_compute_node.initial() is not True:
             logger.error("local sentence transformer image embedding node initial failed!")
         else:
             ComputeKernel.get_instance().add_compute_node(local_st_image_compute_node)
-       
+
 
         await ComputeKernel.get_instance().start()
 
         AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg)
         #AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
-        
+
-        
+
         pipelines = KnowledgePipelineManager.initial(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines"))
         pipelines.load_dir(os.path.join(AIStorage().get_instance().get_system_app_dir(), "knowledge_pipelines"))
         pipelines.load_dir(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge_pipelines"))
@@ -220,7 +228,7 @@ class AIOS_Shell:
     async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str:
         if sender == self.username:
             AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
-        
+
         agent_msg = AgentMsg()
         agent_msg.set(sender,target_id,msg)
         agent_msg.topic = topic
@@ -235,7 +243,7 @@ class AIOS_Shell:
 
     async def _user_process_msg(self,msg:AgentMsg) -> AgentMsg:
         pass
-    
+
 
     async def get_tunnel_config_from_input(self,tunnel_target,tunnel_type):
         tunnel_config = {}
@@ -274,7 +282,7 @@ class AIOS_Shell:
     async def append_tunnel_config(self,tunnel_config):
         user_data_dir = AIStorage.get_instance().get_myai_dir()
         tunnels_config_path = os.path.abspath(f"{user_data_dir}/etc/tunnels.cfg.toml")
-        all_tunnels = None 
+        all_tunnels = None
         try:
             all_tunnels = toml.load(tunnels_config_path)
         except Exception as e:
@@ -327,12 +335,12 @@ class AIOS_Shell:
         if contact_telegram is None:
             return None
         contact.telegram = contact_telegram
-        
+
         contact_email = await try_get_input(f"Input {contact_name}'s email:")
         if contact_email is None:
             return None
         contact.email = contact_email
-        
+
         contact_phone = await try_get_input(f"Input {contact_name}'s phone (optional):")
         if contact_phone is not None:
             contact.phone = contact_phone
@@ -340,13 +348,13 @@ class AIOS_Shell:
         contact_note = await try_get_input(f"Input {contact_name}'s note (optional):")
         if contact_note is not None:
             contact.note = contact_note
-        
+
         contact.added_by = self.username
         if is_update:
             cm.set_contact(contact_name,contact)
         else:
             cm.add_contact(contact_name,contact)
-    
+
     async def handle_knowledge_commands(self, args):
         show_text = FormattedText([("class:title", "sub command not support!\n"
                               "/knowledge pipelines\n" 
@@ -393,7 +401,7 @@ class AIOS_Shell:
             if args[1] == "llama":
                 if len(args) < 4:
                     return show_text
-                
+
                 model_name = args[2]
                 url = args[3]
                 ComputeNodeConfig.get_instance().add_node("llama", url, model_name)
@@ -409,7 +417,7 @@ class AIOS_Shell:
             if args[1] == "llama":
                 if len(args) < 4:
                     return show_text
-                
+
                 model_name = args[2]
                 url = args[3]
                 ComputeNodeConfig.get_instance().remove_node("llama", url, model_name)
@@ -495,7 +503,7 @@ class AIOS_Shell:
                 else:
                     show_text = FormattedText([("class:error", "/open Need Target Agent/Workflow ID! like /open Jarvis default")])
                     return show_text
-                
+
                 if len(args) >= 2:
                     topic = args[1]
                 else:
@@ -506,7 +514,7 @@ class AIOS_Shell:
                     target_exist = True
                 if await WorkflowManager.get_instance().is_exist(target_id):
                     target_exist = True
-                
+
                 if target_exist is False:
                     show_text = FormattedText([("class:error", f"Target {target_id} not exist!")])
                     return show_text
@@ -537,11 +545,11 @@ class AIOS_Shell:
                 else:
                     show_text = FormattedText([("class:error", "/disable Need Feature Name! like /disable llama")])
                     return show_text
-                
+
                 if not await AIStorage.get_instance().is_feature_enable(feature):
                     show_text = FormattedText([("class:title", f"Feature {feature} already disabled!")])
                     return show_text
-                
+
                 await AIStorage.get_instance().disable_feature(feature)
                 show_text = FormattedText([("class:title", f"Feature {feature} disabled!")])
                 return show_text
@@ -717,8 +725,8 @@ async def main():
 
         logging.basicConfig(handlers=[handler],
                             level=logging.INFO,
-                            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')    
+                            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-    
+
     is_daemon = False
     logger.info(f"Check Host OS :{os.name}")
     if os.name != 'nt':
@@ -739,7 +747,7 @@ async def main():
     shell.username = AIStorage.get_instance().get_user_config().get_value("username")
     init_result = await shell.initial()
     proxy.apply_storage()
-    
+
     if init_result is False:
         if is_daemon:
             logger.error("aios shell initial failed!")
@@ -759,7 +767,7 @@ async def main():
                                '/connect $target',
                                '/contact $name',
                                '/knowledge pipelines',
-                               '/knowledge journal $pipeline [$topn]', 
+                               '/knowledge journal $pipeline [$topn]',
                                '/knowledge query $object_id',
                                '/set_config $key',
                                '/enable $feature',