Merge pull request #62 from wugren/MVP

Add story maker

rootfs/agents/fairy_tale_writer/agent.toml

@@ -0,0 +1,8 @@
+instance_id = "fairy_tale_writer"
+fullname = "tracy wang"
+llm_model_name = "gpt-3.5-turbo-16k-0613"
+enable_function = []
+
+[[prompt]]
+role = "system"
+content = "你是一个童话做作家，能够写出各种有趣的童话。"

rootfs/agents/manager/agent.toml

@@ -1,5 +1,7 @@
 instance_id = "agent:xxxxxxabcde"
 fullname = "musk"
+enable_function = []
+
 [[prompt]]
 role = "system"
 content = "你有丰富的管理技能，擅长将复杂工作拆解成简单的任务，让团队成员高效协作。"

rootfs/agents/speecher/agent.toml

@@ -0,0 +1,8 @@
+instance_id = "speecher"
+fullname = "tracy wang"
+llm_model_name = "gpt-3.5-turbo-16k-0613"
+enable_function = ["text_to_speech"]
+
+[[prompt]]
+role = "system"
+content = "你是一个故事播音员，可以将故事演播成音频，演播前需要将故事改编成播音剧本，提取旁白和角色台词，以及每个角色需要有性别、年龄、以及每句台词的语气等。如果生成了音频文件则告知你的用户。"

rootfs/workflows/story_maker/workflow.toml

@@ -0,0 +1,42 @@
+name = "story_maker"
+
+
+[filter]
+"*" = "manager"
+
+[roles.manager]
+name = "manager"
+fullname = "总导演"
+agent="manager"
+enable_function = []
+
+[[roles.manager.prompt]]
+role="system"
+content="""
+你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,speecher。一个故事制作分成两个阶段：让writer写出故事，再交由speecher演播故事生成音频文件。你的基本工作模式是：
+1. 收到客户的明确的指令后,让writer写出故事
+2. 将writer写出的故事交给speecher演播
+3. 获得音频文件之后，将音频文件的存放路径交给客户
+4. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息
+```
+##/send_msg 成员名称
+内容
+```
+"""
+
+[roles.writer]
+name = "writer"
+agent = "fairy_tale_writer"
+fullname = "作家"
+enable_function = []
+[[roles.writer.prompt]]
+role="system"
+content=""
+
+[roles.speecher]
+name = "speecher"
+agent = "speecher"
+enable_function = ["text_to_speech"]
+[[roles.speecher.prompt]]
+role="system"
+content=""

src/aios_kernel/__init__.py

@@ -19,6 +19,7 @@ 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 .text_to_speech_function import TextToSpeechFunction
 from .workspace_env import WorkspaceEnvironment
 
 AIOS_Version = "0.5.1, build 2023-9-17"

src/aios_kernel/agent.py

@@ -26,7 +26,7 @@ class AgentPrompt:
         self.system_message = None
 
     def as_str(self)->str:
-        result_str = "" 
+        result_str = ""
         if self.system_message:
             result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n"
         if self.messages:
@@ -34,18 +34,18 @@ class AgentPrompt:
                 result_str += msg.get("role") + ":" + msg.get("content") + "\n"
 
         return result_str
-    
+
     def to_message_list(self):
         result = []
         if self.system_message:
             result.append(self.system_message)
         result.extend(self.messages)
         return result
-    
+
     def append(self,prompt):
         if prompt is None:
             return
-        
+
         if prompt.system_message is not None:
             if self.system_message is None:
                 self.system_message = prompt.system_message
@@ -99,9 +99,9 @@ class AIAgentTemplete:
                 logger.error("load prompt from config failed!")
                 return False
 
-        
+
         return True
-    
+
 
 class AIAgent:
     def __init__(self) -> None:
@@ -111,7 +111,7 @@ class AIAgent:
         self.agent_id:str = None
         self.template_id:str = None
         self.fullname:str = None
-        self.powerby = None  
+        self.powerby = None
         self.enable = True
         self.enable_kb = False
         self.enable_timestamp = False
@@ -124,7 +124,7 @@ class AIAgent:
         self.owner_env : Environment = None
         self.owenr_bus = None
         self.enable_function_list = []
-        
+
     @classmethod
     def create_from_templete(cls,templete:AIAgentTemplete, fullname:str):
         # Agent just inherit from templete on craete,if template changed,agent will not change
@@ -137,7 +137,7 @@ class AIAgent:
         result_agent.powerby = templete.author
         result_agent.prompt = templete.prompt
         return result_agent
-    
+
     def load_from_config(self,config:dict) -> bool:
         if config.get("instance_id") is None:
             logger.error("agent instance_id is None!")
@@ -190,7 +190,7 @@ class AIAgent:
         if llm_result_str == "ignore":
             r.state = "ignore"
             return r
-        
+
         lines = llm_result_str.splitlines()
         is_need_wait = False
 
@@ -207,7 +207,7 @@ class AIAgent:
 
                     r.send_msgs.append(new_msg)
                     is_need_wait = True
-                    
+
                 case "post_msg":# postmsg($target_id,$msg_content)
                     if len(func_args) != 1:
                         logger.error(f"parse postmsg failed! {func_call}")
@@ -217,22 +217,22 @@ class AIAgent:
                     msg_content = func_item.body
                     new_msg.set(self.agent_id,target_id,msg_content)
                     r.post_msgs.append(new_msg)
-                    
+
                 case "call":# call($func_name,$args_str)
                     r.calls.append(func_item)
                     is_need_wait = True
                     return True
                 case "post_call": # post_call($func_name,$args_str)
                     r.post_calls.append(func_item)
-                    return True    
+                    return True
-                
+
         current_func : FunctionItem = None
         for line in lines:
             if line.startswith("##/"):
                 if current_func:
                     if check_args(current_func) is False:
                         r.resp += current_func.dumps()
-               
+
                 func_name,func_args = AgentMsg.parse_function_call(line[3:])
                 current_func = FunctionItem(func_name,func_args)
             else:
@@ -240,11 +240,11 @@ class AIAgent:
                     current_func.append_body(line + "\n")
                 else:
                     r.resp += line + "\n"
-        
+
         if current_func:
             if check_args(current_func) is False:
                 r.resp += current_func.dumps()
-                
+
         if len(r.send_msgs) > 0 or len(r.calls) > 0:
             r.state = "waiting"
         else:
@@ -281,21 +281,22 @@ class AIAgent:
     def _get_inner_functions(self) -> dict:
         if self.owner_env is None:
             return None
-        
+
         all_inner_function = self.owner_env.get_all_ai_functions()
         if all_inner_function is None:
             return None
-        
+
         result_func = []
         result_len = 0
         for inner_func in all_inner_function:
             func_name = inner_func.get_name()
-            if self.enable_function_list:
+            if self.enable_function_list is not None:
                 if len(self.enable_function_list) > 0:
                     if func_name not in self.enable_function_list:
                         logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}")
                         continue
-
+                else:
+                    continue
             this_func = {}
             this_func["name"] = func_name
             this_func["description"] = inner_func.get_description()
@@ -315,28 +316,29 @@ class AIAgent:
         func_node : AIFunction = self.owner_env.get_ai_function(func_name)
         if func_node is None:
             return "execute failed,function not found"
-        
+
         ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target)
         try:
             result_str:str = await func_node.execute(**arguments)
         except Exception as e:
-            result_str = "call error:" + str(e)  
+            result_str = "call error:" + str(e)
             logger.error(f"llm execute inner func:{func_name} error:{e}")
-            
 
+
+        logger.info("llm execute inner func result:" + result_str)
         inner_functions,inner_function_len = self._get_inner_functions()
         prompt.messages.append({"role":"function","content":result_str,"name":func_name})
         task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions)
-        
+
         ineternal_call_record.result_str = task_result.result_str
         ineternal_call_record.done_time = time.time()
         org_msg.inner_call_chain.append(ineternal_call_record)
 
         if stack_limit > 0:
             inner_func_call_node = task_result.result_message.get("function_call")
-            
+
         if inner_func_call_node:
-            return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1)      
+            return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1)
         else:
             return task_result.result_str
 
@@ -346,7 +348,7 @@ class AIAgent:
     def _format_msg_by_env_value(self,prompt:AgentPrompt):
         if self.owner_env is None:
             return
-        
+
         for msg in prompt.messages:
             old_content = msg.get("content")
             msg["content"] = old_content.format_map(self.owner_env)
@@ -382,7 +384,7 @@ class AIAgent:
        
             system_prompt_len = prompt.get_prompt_token_len()
             input_len = len(msg.body)
-            
+
             history_prmpt,history_token_len = await self._get_prompt_from_session(chatsession,system_prompt_len + function_token_len,input_len)
             prompt.append(history_prmpt) # chat context
             
@@ -399,7 +401,7 @@ class AIAgent:
             if inner_func_call_node:
                 #TODO to save more token ,can i use msg_prompt?
                 final_result = await self._execute_func(inner_func_call_node,prompt,msg)
-            
+
             llm_result : LLMResult = self._get_llm_result_type(final_result)
             is_ignore = False
             result_prompt_str = ""
@@ -417,21 +419,21 @@ class AIAgent:
                             agent_sesion = AIChatSession.get_session(self.agent_id,f"{sendmsg.target}#{sendmsg.topic}",self.chat_db)
                             agent_sesion.append(sendmsg)
                             agent_sesion.append(send_resp)
-                    
+
                     final_result = llm_result.resp + result_prompt_str
 
             if is_ignore is not True:
                 resp_msg = msg.create_resp_msg(final_result)
                 chatsession.append(msg)
                 chatsession.append(resp_msg)
-                
+
                 return resp_msg
-            
+
             return None
-        
+
     def get_id(self) -> str:
         return self.agent_id
-    
+
     def get_fullname(self) -> str:
         return self.fullname
 
@@ -440,14 +442,14 @@ class AIAgent:
 
     def get_llm_model_name(self) -> str:
         return self.llm_model_name
-    
+
     def get_max_token_size(self) -> int:
         return self.max_token_size
-    
+
     async def _get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len,is_groupchat=False) -> AgentPrompt:
         # TODO: get prompt from group chat is different from single chat
         history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len
-        messages = chatsession.read_history() # read 
+        messages = chatsession.read_history() # read
         result_token_len = 0
         result_prompt = AgentPrompt()
         read_history_msg = 0

src/aios_kernel/bus.py

@@ -19,14 +19,14 @@ class AIBusHandler:
     async def handle_message(self,msg:AgentMsg) -> Any:
         if self.handler is None:
             return None
-        
+
         resp_msg = await self.handler(msg)
         if self.enable_defualt_proc:
             if resp_msg is not None:
                 await self.owner_bus.post_message(resp_msg,False)
 
         return resp_msg
-            
+
 class AIBus:
     _instance = None
     @classmethod
@@ -48,16 +48,16 @@ class AIBus:
             if msg.rely_msg_id is not None:
                 handler.results[msg.rely_msg_id] = msg
                 return None
-            
+
             handler.queue.put_nowait(msg)
             self.start_process(target_id)
             return True
-        
+
         if use_unhandle:
             if self.unhandle_handler is not None:
                 if await self.unhandle_handler(self,msg):
                     return await self.post_message(msg,False)
-      
+
         logger.warn(f"post message to {msg.target} failed!,target not found")
         return False
 
@@ -71,7 +71,7 @@ class AIBus:
         if sender_handler is None:
             logger.warn(f"sender {sender_id} not register on AI_BUS!")
             return None
-        
+
         post_result = await self.post_message(msg)
         if post_result is False:
             return None
@@ -84,37 +84,39 @@ class AIBus:
                 msg.status = AgentMsgStatus.RESPONSED
                 del sender_handler.results[msg.msg_id]
                 return resp
-            
+
             await asyncio.sleep(0.2)
             retry_times += 1
-            if retry_times > 5*120: # default timeout is 120 sec
+            if retry_times > 5*240: # default timeout is 240 sec
                 msg.status = AgentMsgStatus.ERROR
                 return None
-            
+
         return None
-    
+
     def register_unhandle_message_handler(self,handler:Any) -> Queue:
         self.unhandle_handler = handler
 
     # means sub
-    def register_message_handler(self,handler_name:str,handler:Any) -> Queue:        
+    def register_message_handler(self,handler_name:str,handler:Any) -> Queue:
-        handler_node =  AIBusHandler(handler,self) 
+        handler_node =  AIBusHandler(handler,self)
         self.handlers[handler_name] = handler_node
         return handler_node.queue
-    
+
     async def process_queue(self, handler:AIBusHandler):
         while True:
             # Wait for a message
             message = await handler.queue.get()
 
-            #try:
+            try:
                 # Try to handle the message
-            await handler.handle_message(message)
+                await handler.handle_message(message)
-            #except Exception as e:
+            except Exception as e:
                 # If an error occurs, put the message back into the queue
-            #    logger.error(f"handle message {message.msg_id} failed! {e}")
+                logger.error(f"handle message {message.msg_id} failed! {e}")
+                logger.exception(e)
+                raise e
                 #self.queues[name].put_nowait(message)
-        
+
         return
 
     def start_process(self,target_name):
@@ -122,12 +124,12 @@ class AIBus:
         if handler is None:
             logger.error(f"handler {target_name} not found!")
             return
-        
+
         if handler.handler is None:
             return
 
         if handler.working_task is not None:
             logger.warn(f"handler {target_name} is already working!")
             return
-        
+
-        handler.working_task = asyncio.create_task(self.process_queue(handler))
+        handler.working_task = asyncio.create_task(self.process_queue(handler))

src/aios_kernel/chatsession.py

@@ -35,10 +35,10 @@ class ChatSessionDB:
             self._create_table(conn)
 
         return conn
-    
+
     def close(self):
         if not hasattr(self.local, 'conn'):
-            return 
+            return
         self.local.conn.close()
 
     def _create_table(self, conn):
@@ -56,7 +56,7 @@ class ChatSessionDB:
 
             # create messages table
             # reciver_id could be None
-        
+
             conn.execute("""
                 CREATE TABLE IF NOT EXISTS Messages (
                     MessageID TEXT PRIMARY KEY,
@@ -142,7 +142,7 @@ class ChatSessionDB:
         except Error as e:
             logging.error("Error occurred while inserting message: %s", e)
             return -1  # return -1 if an error occurs
-    
+
     def get_chatsession_by_id(self, session_id):
         """Get a message by its ID"""
         conn = self._get_conn()
@@ -150,7 +150,7 @@ class ChatSessionDB:
         c.execute("SELECT * FROM ChatSessions WHERE SessionID = ?", (session_id,))
         chatsession = c.fetchone()
         return chatsession
-    
+
     def get_chatsession_by_owner_topic(self, owner_id, topic):
         """Get a chatsession by its owner and topic"""
         conn = self._get_conn()
@@ -175,7 +175,7 @@ class ChatSessionDB:
         except Error as e:
             logging.error("Error occurred while getting sessions: %s", e)
             return -1, None  # return -1 and None if an error occurs
-        
+
     def get_message_by_id(self, message_id):
         """Get a message by its ID"""
         conn =self._get_conn()
@@ -216,7 +216,7 @@ class ChatSessionDB:
         except Error as e:
             logging.error("Error occurred while updating message status: %s", e)
             return -1  # return -1 if an error occurs
-        
+
 
 # chat session store the chat history between owner and agent
 # chat session might be large, so can read / write at stream mode.
@@ -230,7 +230,7 @@ class AIChatSession:
     #        cls._dbs[db_path] = db
     #    db.get_chatsession_by_id(session_id)
     #    #result = AIChatSession()
-    
+
     @classmethod
     def get_session(cls,owner_id:str,session_topic:str,db_path:str,auto_create = True) -> str:
         db = cls._dbs.get(db_path)
@@ -249,21 +249,22 @@ class AIChatSession:
             result = AIChatSession(owner_id,session[0],db)
             result.topic = session_topic
 
-        return result          
+        return result
-    
+
 
     def __init__(self,owner_id:str, session_id:str, db:ChatSessionDB) -> None:
         self.owner_id :str = owner_id
         self.session_id : str = session_id
         self.db : ChatSessionDB = db
-        
+
         self.topic : str = None
         self.start_time : str = None
 
     def get_owner_id(self) -> str:
         return self.owner_id
-    
+
     def read_history(self, number:int=10,offset=0) -> [AgentMsg]:
+        return []
         msgs = self.db.get_messages(self.session_id, number, offset)
         result = []
         for msg in msgs:
@@ -298,4 +299,4 @@ class AIChatSession:
     #    """chat session changed event handler"""
     #    pass
 
-    #TODO : add iterator interface for read chat history 
+    #TODO : add iterator interface for read chat history

src/aios_kernel/compute_kernel.py

@@ -7,7 +7,7 @@ from asyncio import Queue
 
 from .agent import AgentPrompt
 from .compute_node import ComputeNode
-from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskResult
+from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskResult, ComputeTaskType
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +23,7 @@ class ComputeKernel:
         if cls._instance is None:
             cls._instance = ComputeKernel()
         return cls._instance
-    
+
     def __init__(self) -> None:
         self.is_start = False
         self.task_queue = Queue()
@@ -73,7 +73,7 @@ class ComputeKernel:
         hit_pos = random.randint(0, total_weights - 1)
         for i in range(min(len(support_nodes) - 1, hit_pos), -1, -1):
             if support_nodes[i]["pos"] <= hit_pos:
-                return node
+                return support_nodes[i]["node"]
 
         logger.warning(
             f"task {task.display()} is not support by any compute node")
@@ -118,7 +118,7 @@ class ComputeKernel:
                 if task_req.state == ComputeTaskState.ERROR:
                     break
 
-                if check_times >= 20:
+                if check_times >= 120:
                     task_req.state = ComputeTaskState.ERROR
                     break
 
@@ -129,7 +129,7 @@ class ComputeKernel:
         if task_req.state == ComputeTaskState.DONE:
             return task_req.result
 
-        return "error!"
+        raise Exception("error!")
 
     
     def text_embedding(self,input:str,model_name:Optional[str] = None):
@@ -137,7 +137,7 @@ class ComputeKernel:
         task_req.set_text_embedding_params(input,model_name)
         self.run(task_req)
         return task_req
-    
+
     async def do_text_embedding(self,input:str,model_name:Optional[str] = None) -> [float]:
         task_req = self.text_embedding(input,model_name)
         async def check_timer():
@@ -155,11 +155,45 @@ class ComputeKernel:
 
                 await asyncio.sleep(0.5)
                 check_times += 1
-            
+
         await asyncio.create_task(check_timer())
         if task_req.state == ComputeTaskState.DONE:
             return task_req.result.result
-              
+
-        return "error!"
+        return "error!"
-    
+
+    async def do_text_to_speech(self,
+                       input:str,
+                       language_code:Optional[str] = None,
+                       gender: Optional[str] = None,
+                       age: Optional[str] = None,
+                       voice_name: Optional[str] = None,
+                       tone: Optional[str] = None):
+        task_req = ComputeTask()
+        task_req.params["text"] = input
+        task_req.params["language_code"] = language_code
+        task_req.params["gender"] = gender
+        task_req.params["age"] = age
+        task_req.params["voice_name"] = voice_name
+        task_req.params["tone"] = tone
+        task_req.task_type = ComputeTaskType.TEXT_2_VOICE
+        self.run(task_req)
+
+        check_times = 0
+        while True:
+            if task_req.state == ComputeTaskState.DONE:
+                break
+            if task_req.state == ComputeTaskState.ERROR:
+                break
+            if check_times >= 60:
+                task_req.state = ComputeTaskState.ERROR
+                break
+            await asyncio.sleep(0.5)
+            check_times += 1
+
+        if task_req.state == ComputeTaskState.DONE:
+            return task_req.result.result
+        else:
+            raise Exception("do_text_to_speech failed!")
+
 

src/aios_kernel/google_text_to_speech_node.py

@@ -3,9 +3,11 @@ import os
 import asyncio
 from asyncio import Queue
 import logging
+from typing import Optional
 
 from google.cloud import texttospeech
 
+from .storage import AIStorage
 from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
 from .compute_node import ComputeNode
 
@@ -21,26 +23,78 @@ see:https://cloud.google.com/text-to-speech/docs/before-you-begin
 class GoogleTextToSpeechNode(ComputeNode):
     _instance = None
 
-    def __new__(cls, *args, **kwargs):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super(GoogleTextToSpeechNode, cls).__new__(cls)
+            cls._instance = 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: Optional[texttospeech.TextToSpeechClient] = None
 
-        self.client = texttospeech.TextToSpeechClient()
+        self.language_list = {
-
+            "cnm-CN": {
+                "female": ["cmn-CN-Standard-A",
+                           "cmn-CN-Standard-D",
+                           "cmn-CN-Wavenet-A",
+                           "cmn-CN-Wavenet-D",
+                           "cmn-TW-Standard-A",
+                           "cmn-TW-Wavenet-A"],
+                "man": ["cmn-CN-Standard-B",
+                        "cmn-CN-Standard-C",
+                        "cmn-CN-Wavenet-B",
+                        "cmn-CN-Wavenet-C",
+                        "cmn-TW-Standard-B",
+                        "cmn-TW-Standard-C",
+                        "cmn-TW-Wavenet-B",
+                        "cmn-TW-Wavenet-C"]
+            },
+            "en-US": {
+                "female": ["en-US-Neural2-C",
+                           "en-US-Neural2-E",
+                           "en-US-Neural2-F",
+                           "en-US-Neural2-G",
+                           "en-US-Neural2-H",
+                           "en-US-News-K",
+                           "en-US-News-L",
+                           "en-US-Standard-C",
+                           "en-US-Standard-E",
+                           "en-US-Standard-F",
+                           "en-US-Standard-G",
+                           "en-US-Standard-H",
+                           "en-US-Studio-O",
+                           "en-US-Wavenet-C",
+                           "en-US-Wavenet-E",
+                           "en-US-Wavenet-F",
+                           "en-US-Wavenet-G",
+                           "en-US-Wavenet-H"],
+                "man": ["en-US-Polyglot-1",
+                        "en-US-Standard-A",
+                        "en-US-Standard-B",
+                        "en-US-Standard-D",
+                        "en-US-Standard-I",
+                        "en-US-Standard-J",
+                        "en-US-Studio-M",
+                        "en-US-Wavenet-A",
+                        "en-US-Wavenet-B",
+                        "en-US-Wavenet-D",
+                        "en-US-Wavenet-I",
+                        "en-US-Wavenet-J"]
+            }
+        }
         self.start()
 
+    def init(self):
+        user_config = AIStorage.get_instance().get_user_config()
+        google_application_credentials = user_config.get_value("google_application_credentials")
+        if google_application_credentials is None:
+            raise Exception("google_application_credentials is None!")
+        os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = google_application_credentials
+        self.client = texttospeech.TextToSpeechClient()
+
     def start(self):
         async def _run_task_loop():
             while True:
@@ -64,10 +118,24 @@ class GoogleTextToSpeechNode(ComputeNode):
         task.state = ComputeTaskState.RUNNING
         language_code = task.params["language_code"]
         text = task.params["text"]
+        voice_name = task.params["voice_name"]
+        gender = task.params["gender"]
+        age = task.params["age"]
+
+        if language_code == "zh":
+            language_code = "cnm-CN"
+        elif language_code == "en":
+            language_code = "en-US"
+        else:
+            raise Exception(f"language_code {language_code} not support")
+
+        lang_list = self.language_list[language_code][gender]
+        voice = lang_list[hash(voice_name) % len(lang_list)]
 
         synthesis_input = texttospeech.SynthesisInput(text=text)
         voice = texttospeech.VoiceSelectionParams(language_code=language_code,
-                                                  ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL)
+                                                  ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL,
+                                                  name=voice)
 
         audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3)
 
@@ -95,10 +163,18 @@ class GoogleTextToSpeechNode(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.TEXT_2_VOICE:
+        if task.task_type == ComputeTaskType.TEXT_2_VOICE:
             return True
         return False
 
     def is_local(self) -> bool:
         return False
+
+    def declare_user_config(self):
+        if os.getenv("GOOGLE_APPLICATION_CREDENTIALS") is None:
+            user_config = AIStorage.get_instance().get_user_config()
+            user_config.add_user_config("google_application_credentials",
+                                        "google application credentials, please visit:https://cloud.google.com/text-to-speech/docs/before-you-begin",
+                                        False,
+                                        None)

src/aios_kernel/open_ai_node.py

@@ -20,7 +20,7 @@ class OpenAI_ComputeNode(ComputeNode):
         if cls._instance is None:
             cls._instance = OpenAI_ComputeNode()
         return cls._instance
-    
+
     @classmethod
     def declare_user_config(cls):
         if os.getenv("OPENAI_API_KEY_") is None:
@@ -46,7 +46,7 @@ class OpenAI_ComputeNode(ComputeNode):
         if self.openai_api_key is None:
             logger.error("openai_api_key is None!")
             return False
-        
+
         openai.api_key = self.openai_api_key
         self.start()
         return True
@@ -68,7 +68,7 @@ class OpenAI_ComputeNode(ComputeNode):
 
                 resp = openai.Embedding.create(model=model_name,
                                                 input=input)
-                
+
                 # resp = {
                 # "object": "list",
                 # "data": [
@@ -86,7 +86,7 @@ class OpenAI_ComputeNode(ComputeNode):
 
                 logger.info(f"openai response: {resp}")
 
-                result = ComputeTaskResult()    
+                result = ComputeTaskResult()
                 result.set_from_task(task)
                 result.worker_id = self.node_id
                 result.result = resp["data"][0]["embedding"]
@@ -100,23 +100,23 @@ class OpenAI_ComputeNode(ComputeNode):
                 if max_token_size is None:
                     max_token_size = 4000
 
-                result_token = int(max_token_size * 0.4)
+                result_token = max_token_size
-                    
-                logger.info(f"call openai {mode_name} prompts: {prompts}")
 
                 if llm_inner_functions is None:
+                    logger.info(f"call openai {mode_name} prompts: {prompts}")
                     resp = openai.ChatCompletion.create(model=mode_name,
                                                     messages=prompts,
                                                     max_tokens=result_token,
                                                     temperature=0.7)
                 else:
+                    logger.info(f"call openai {mode_name} prompts: {prompts} functions: {json.dumps(llm_inner_functions)}")
                     resp = openai.ChatCompletion.create(model=mode_name,
                                                         messages=prompts,
                                                         functions=llm_inner_functions,
                                                         max_tokens=result_token,
                                                         temperature=0.7) # TODO: add temperature to task params?
 
-            
+
                 logger.info(f"openai response: {json.dumps(resp, indent=4)}")
 
                 result = ComputeTaskResult()
@@ -139,6 +139,7 @@ class OpenAI_ComputeNode(ComputeNode):
                 result.result_message = resp["choices"][0]["message"]
                 if token_usage:
                     result.result_refers["token_usage"] = token_usage
+                logger.info(f"openai success response: {result.result_str}")
                 return result
             case _:
                 task.state = ComputeTaskState.ERROR
@@ -148,7 +149,7 @@ class OpenAI_ComputeNode(ComputeNode):
         if self.is_start is True:
             return
         self.is_start = True
-        
+
         async def _run_task_loop():
             while True:
                 task = await self.task_queue.get()
@@ -171,13 +172,13 @@ class OpenAI_ComputeNode(ComputeNode):
 
 
     def is_support(self, task: ComputeTask) -> bool:
-        if task.task_type == ComputeTaskType.LLM_COMPLETION: 
+        if task.task_type == ComputeTaskType.LLM_COMPLETION:
             if not task.params["model_name"]:
                 return True
             model_name : str = task.params["model_name"]
             if model_name.startswith("gpt-"):
                 return True
-            
+
         if task.task_type == ComputeTaskType.TEXT_EMBEDDING:
             if task.params["model_name"] == "text-embedding-ada-002":
                 return True

src/aios_kernel/text_to_speech_function.py

@@ -0,0 +1,101 @@
+import io
+import logging
+import os
+import random
+from typing import Dict
+
+from aios_kernel import ComputeKernel
+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 = "根据输入的剧本生成音频文件，成功时会返回音频文件路径"
+
+    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"]},
+                "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"
+        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)
+                    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))))
+            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/workflow.py

@@ -2,7 +2,7 @@ import logging
 import asyncio
 import json
 import os
-import time 
+import time
 from asyncio import Queue
 from typing import Optional,Tuple,List
 from abc import ABC, abstractmethod
@@ -32,7 +32,7 @@ class MessageFilter:
 
         # TODO: add more filter
         return None
-            
+
     def load_from_config(self,config:dict) -> bool:
         self.filters = config
         return True
@@ -68,8 +68,8 @@ class Workflow:
     def load_from_config(self,config:dict) -> bool:
         if config is None:
             return False
-        
+
-        if config.get("name") is None: 
+        if config.get("name") is None:
             logger.error("workflow config must have name")
             return False
         self.workflow_name = config.get("name")
@@ -84,7 +84,7 @@ class Workflow:
             if self.rule_prompt.load_from_config(config.get("prompt")) is False:
                 logger.error("Workflow load prompt failed")
                 return False
-            
+
         if config.get("roles") is None:
             logger.error("workflow config must have roles")
             return False
@@ -106,13 +106,13 @@ class Workflow:
             self.env_db_file = self.owner_workflow.env_db_file
         self.workflow_env = WorkflowEnvironment(self.workflow_id,self.env_db_file)
 
-        env_ndoe = config.get("enviroment") 
+        env_ndoe = config.get("enviroment")
         if  env_ndoe is not None:
             if self._load_env_from_config(env_ndoe) is False:
                 logger.error("Workflow load env failed")
                 return False
 
-        connected_env_ndoe = config.get("connected_env") 
+        connected_env_ndoe = config.get("connected_env")
         if  connected_env_ndoe is not None:
            for _node in connected_env_ndoe:
                 env_id = _node.get("env_id")
@@ -124,13 +124,13 @@ class Workflow:
                      logger.error(f"Workflow load connected_env failed, env {env_id} not found!")
                      return False
                 self.connect_to_environment(remote_env,_node.get("event2msg"))
-            
+
         sub_workflows = config.get("sub_workflows")
         if sub_workflows is not None:
             if self._load_sub_workflows(sub_workflows) is False:
                 logger.error("Workflow load sub workflows failed")
                 return False
-            
+
         return True
 
     def _load_env_from_config(self,config:dict) -> bool:
@@ -147,7 +147,7 @@ class Workflow:
                 return False
             self.sub_workflows[k] = sub_workflow
         return True
-    
+
     def _parse_msg_target(self,s:str)->list[str]:
         return s.split(".")
 
@@ -170,12 +170,12 @@ class Workflow:
                 if current_workflow is None:
                     logger.error(f"sub workflow {inner_obj_id[i]} not found!")
                     return None
-            
+
             i += 1
-        
+
         logger.error(f"{msg.target} not found! forword message failed!")
         return None
-        
+
     def get_workflow_id_from_target(self,target:str) -> str:
         target_list = target.split(".")
         if len(target_list) == 0:
@@ -203,11 +203,11 @@ class Workflow:
 
         #1. workflow start process message
         final_result = None
-        
+
         # this is workflow's group_chat session
         session_topic = msg.sender + "#" + msg.topic
         chatsesssion = AIChatSession.get_session(self.workflow_id,session_topic,self.db_file)
-        
+
         #2. find role by msg.mentions or workflow's selector logic
         if msg.mentions is not None:
             if not self.workflow_id in msg.mentions:
@@ -219,20 +219,20 @@ class Workflow:
                 this_role = self.role_group.get(mention)
                 if this_role is not None:
                     return await self.role_process_msg(msg,this_role,chatsesssion)
-                
+
         if self.input_filter is not None:
             select_role_id = self.input_filter.select(msg)
-            if select_role_id is not None: 
+            if select_role_id is not None:
                 select_role = self.role_group.get(select_role_id)
                 if select_role is None:
                     logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group")
                     return None
-                
+
                 return await self.role_process_msg(msg,select_role,chatsesssion)
             else:
                 logger.error(f"input_filter return None for :{msg.body}")
                 return None
-                
+
         logger.error(f"{self.workflow_id}:no role can process this msg:{msg.body}")
         return final_result
 
@@ -245,7 +245,7 @@ class Workflow:
         if llm_result_str == "ignore":
             r.state = "ignore"
             return r
-        
+
         lines = llm_result_str.splitlines()
         is_need_wait = False
 
@@ -262,7 +262,7 @@ class Workflow:
 
                     r.send_msgs.append(new_msg)
                     is_need_wait = True
-                    
+
                 case "post_msg":# postmsg($target_id,$msg_content)
                     if len(func_args) != 1:
                         logger.error(f"parse postmsg failed! {func_call}")
@@ -272,22 +272,22 @@ class Workflow:
                     msg_content = func_item.body
                     new_msg.set("_",target_id,msg_content)
                     r.post_msgs.append(new_msg)
-                    
+
                 case "call":# call($func_name,$args_str)
                     r.calls.append(func_item)
                     is_need_wait = True
                     return True
                 case "post_call": # post_call($func_name,$args_str)
                     r.post_calls.append(func_item)
-                    return True    
+                    return True
-                
+
         current_func : FunctionItem = None
         for line in lines:
             if line.startswith("##/"):
                 if current_func:
                     if check_args(current_func) is False:
                         r.resp += current_func.dumps()
-               
+
                 func_name,func_args = AgentMsg.parse_function_call(line[3:])
                 current_func = FunctionItem(func_name,func_args)
             else:
@@ -295,7 +295,7 @@ class Workflow:
                     current_func.append_body(line + "\n")
                 else:
                     r.resp += line + "\n"
-        
+
         if current_func:
             if check_args(current_func) is False:
                 r.resp += current_func.dumps()
@@ -309,14 +309,14 @@ class Workflow:
 
     async def role_post_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession):
         msg.sender = the_role.get_role_id()
-        
+
         target_role = self.role_group.get(msg.target)
         if target_role:
             msg.target = target_role.get_role_id()
             logger.info(f"{msg.sender} post message {msg.msg_id} to inner role: {msg.target}")
             asyncio.create_task(self.role_process_msg(msg,target_role,workflow_chat_session))
             return
-        
+
         target_workflow = self.sub_workflows.get(msg.target)
         if target_workflow:
             msg.target = target_workflow.workflow_id
@@ -341,7 +341,7 @@ class Workflow:
             # msg.target = target_workflow.workflow_id
             logger.info(f"{msg.sender} send message {msg.msg_id} to sub workflow: {msg.target}")
             return await target_workflow._process_msg(msg)
-        
+
         logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}")
         return await self.get_bus().send_message(msg)
 
@@ -352,8 +352,8 @@ class Workflow:
         func_node : AIFunction = self.workflow_env.get_ai_function(func_item.name)
         if func_node is None:
             return "execute failed,function not found"
-    
+
-        result_str:str = await func_node.execute(**arguments)        
+        result_str:str = await func_node.execute(**arguments)
         return result_str
 
     async def role_post_call(self,func_item:FunctionItem,the_role:AIRole):
@@ -363,7 +363,7 @@ class Workflow:
     def _format_msg_by_env_value(self,prompt:AgentPrompt):
         if self.workflow_env is None:
             return
-        
+
         for msg in prompt.messages:
             old_content = msg.get("content")
             msg["content"] = old_content.format_map(self.workflow_env)
@@ -372,15 +372,17 @@ class Workflow:
         all_inner_function = self.workflow_env.get_all_ai_functions()
         if all_inner_function is None:
             return None
-        
+
         result_func = []
         for inner_func in all_inner_function:
             func_name = inner_func.get_name()
-            if the_role.enable_function_list:
+            if the_role.enable_function_list is not None:
                 if len(the_role.enable_function_list) > 0:
                     if func_name not in the_role.enable_function_list:
-                        logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}")
+                        logger.debug(f"agent {self.agent_id} ignore inner func:{func_name}")
                         continue
+                else:
+                    continue
             this_func = {}
             this_func["name"] = func_name
             this_func["description"] = inner_func.get_description()
@@ -399,17 +401,17 @@ class Workflow:
         func_node : AIFunction = self.workflow_env.get_ai_function(func_name)
         if func_node is None:
             return "execute failed,function not found"
-        
+
         ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target)
 
         result_str:str = await func_node.execute(**arguments)
-        
+
-        inner_functions = self._get_inner_functions()
+        inner_functions = self._get_inner_functions(the_role)
         prompt.messages.append({"role":"function","content":result_str,"name":func_name})
         task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,
                                                                                 the_role.agent.llm_model_name,the_role.agent.max_token_size,
                                                                                 inner_functions)
-        
+
         ineternal_call_record.result_str = task_result.result_str
         ineternal_call_record.done_time = time.time()
         org_msg.inner_call_chain.append(ineternal_call_record)
@@ -419,13 +421,13 @@ class Workflow:
             return await self._role_execute_func(the_role,inner_func_call_node,prompt,org_msg,stack_limit-1)      
         else:
             return task_result.result_str
-    
+
     def _is_in_same_workflow(self,msg) -> bool:
         pass
 
-    async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession):        
+    async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession):
         msg.target = the_role.get_role_id()
-  
+
 
         prompt = AgentPrompt()
         prompt.append(the_role.agent.prompt)
@@ -433,7 +435,7 @@ class Workflow:
         prompt.append(the_role.get_prompt())
         # prompt.append(self._get_function_prompt(the_role.get_name()))
         # prompt.append(self._get_knowlege_prompt(the_role.get_name()))
-        
+
         #support group chat, user content include sender name!
         prompt.append(await self._get_prompt_from_session(workflow_chat_session))
 
@@ -443,15 +445,15 @@ class Workflow:
 
         self._format_msg_by_env_value(prompt)
         inner_functions = self._get_inner_functions(the_role)
-        
+
         async def _do_process_msg():
             #TODO: send msg to agent might be better?
             task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size(),inner_functions)
             result_str = task_result.result_str
             logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}")
-            
+
             inner_func_call_node = task_result.result_message.get("function_call")
-            
+
             if inner_func_call_node:
                 #TODO to save more token ,can i use msg_prompt?
                 result_str = await self._role_execute_func(the_role,inner_func_call_node,prompt,msg)
@@ -461,7 +463,7 @@ class Workflow:
                 postmsg.prev_msg_id = msg.get_msg_id()
                 # might be craete a new msg.topic for this postmsg
                 postmsg.topic = msg.topic
-                
+
                 await self.role_post_msg(postmsg,the_role,workflow_chat_session)
                 if not self._is_in_same_workflow(postmsg):
                     role_sesion = AIChatSession.get_session(the_role.get_role_id(),f"{postmsg.target}#{msg.topic}",self.db_file)
@@ -469,14 +471,14 @@ class Workflow:
                 else:
                     # message will be saved in role.process_message
                     pass
-  
+
-                
+
             for post_call in result.post_calls:
                 action_msg = msg.create_action_msg(post_call[0],post_call[1],the_role.get_role_id())
                 workflow_chat_session.append(action_msg)
                 await self.role_post_call(post_call,the_role)
                 #save post_call
-                
+
             result_prompt_str = ""
             match result.state:
                 case "ignore":
@@ -506,11 +508,11 @@ class Workflow:
                         else:
                              # message will be saved in role.process_message
                             pass
-                       
+
                     for call in result.calls:
                         action_msg = msg.create_action_msg(call[0],call[1],call_result,the_role.get_role_id)
                         call_result = await self.role_call(call,the_role)
-          
+
                         if call_result is not None:
                             result_prompt_str += f"\ncall {call[0]} result is :{call_result}"
                             #save action
@@ -522,7 +524,7 @@ class Workflow:
                     prompt.append(result_prompt)
                     r = await _do_process_msg()
                     return r
-        
+
         return await _do_process_msg()
 
     async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt:
@@ -533,9 +535,9 @@ class Workflow:
                 result_prompt.messages.append({"role":"assistant","content":msg.body})
             else:
                 result_prompt.messages.append({"role":"user","content":f"{msg.body}"})
-        
+
         return result_prompt
-    
+
     def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt:
         pass
 
@@ -557,7 +559,7 @@ class Workflow:
             #        if k == "role":
             #            continue
             #        else:
-            #            
+            #
             #            def _env_msg_handler(env_event:EnvironmentEvent) -> None:
             #                the_msg:AgentMsg= self._env_event_to_msg(env_event)
             #                self.role_post_msg

src/aios_kernel/workflow_env.py

@@ -7,6 +7,8 @@ from sqlite3 import Error
 import threading
 import logging
 from typing import Optional
+
+from .text_to_speech_function import TextToSpeechFunction
 from .environment import Environment,EnvironmentEvent
 from .ai_function import SimpleAIFunction
 from .storage import AIStorage
@@ -23,8 +25,8 @@ class CalenderEvent(EnvironmentEvent):
         self.data = data
 
     def display(self) -> str:
-        return f"#event timer:{self.data}"    
+        return f"#event timer:{self.data}"
-    
+
 # AI Calender GOAL: Let user use "create notify after 2 days" to create a timer event
 class CalenderEnvironment(Environment):
     def __init__(self, env_id: str) -> None:
@@ -39,7 +41,7 @@ class CalenderEnvironment(Environment):
         #self.add_ai_function(SimpleAIFunction("serach_events",
         #                                "search events in calender",
         #                                self._search_events))
-        
+
         get_param = {
             "start_time": "start time (UTC) of event",
             "end_time": "end time (UTC) of event"
@@ -59,14 +61,14 @@ class CalenderEnvironment(Environment):
         self.add_ai_function(SimpleAIFunction("add_event",
                                         "add event to calender",
                                         self._add_event,add_param))
-        
+
         delete_param = {
             "event_id": "id of event"
         }
         self.add_ai_function(SimpleAIFunction("delete_event",
                                         "delete event from calender",
                                         self._delete_event,delete_param))
-        
+
         update_param = {
             "event_id": "id of event",
             "new_title": "new title of event",
@@ -79,7 +81,7 @@ class CalenderEnvironment(Environment):
         self.add_ai_function(SimpleAIFunction("update_event",
                                         "update event in calender",
                                         self._update_event,update_param))
-        
+
         #self.add_ai_function(SimpleAIFunction("user_confirm",
         #                                      "user confirm",
         #                                      self._user_confirm))
@@ -98,7 +100,7 @@ class CalenderEnvironment(Environment):
                 );
             """)
             await db.commit()
-        
+
     async def _add_event(self,title, start_time, end_time, participants=None, location=None, details=None):
         async with aiosqlite.connect(self.db_file) as db:
             await db.execute("""
@@ -127,7 +129,7 @@ class CalenderEnvironment(Environment):
                 _event["details"] = row[6]
                 result[row[0]] = _event
             return json.dumps(result, indent=4, sort_keys=True)
-        
+
     async def _get_events_by_time_range(self,start_time, end_time):
         async with aiosqlite.connect(self.db_file) as db:
             cursor = await db.execute("""
@@ -153,7 +155,7 @@ class CalenderEnvironment(Environment):
                 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):
         fields_to_update = []
         values = []
@@ -191,8 +193,8 @@ class CalenderEnvironment(Environment):
             WHERE id = ?;
         """
 
-        values.append(event_id)       
+        values.append(event_id)
-       
+
         async with aiosqlite.connect(self.db_file) as db:
             await db.execute(sql_update_query, values)
             await db.commit()
@@ -212,16 +214,16 @@ class CalenderEnvironment(Environment):
 
     async def start(self) -> None:
         if self.is_run:
-            return 
+            return
         self.is_run = True
         await self.init_db()
-        
+
         self.register_get_handler("now",self.get_now)
         async def timer_loop():
             while True:
                 if self.is_run == False:
                     break
-                
+
                 await asyncio.sleep(1.0)
                 now = datetime.now()
                 formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
@@ -238,13 +240,13 @@ class CalenderEnvironment(Environment):
     def get_now(self)->str:
         now = datetime.now()
         formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
-        return formatted_time     
+        return formatted_time
 
     async def _get_now(self) -> str:
         now = datetime.now()
         formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
         return formatted_time
-    
+
 # Default Workflow Environment(Context)
 class WorkflowEnvironment(Environment):
     def __init__(self, env_id: str,db_file:str) -> None:
@@ -252,6 +254,7 @@ class WorkflowEnvironment(Environment):
         self.db_file = db_file
         self.local = threading.local()
         self.table_name = "WorkflowEnv_" + env_id
+        self.add_ai_function(TextToSpeechFunction())
 
 
     def _get_conn(self):
@@ -259,7 +262,7 @@ class WorkflowEnvironment(Environment):
         if not hasattr(self.local, 'conn'):
             self.local.conn = self._create_connection()
         return self.local.conn
-    
+
     def _create_connection(self):
         """ create a database connection to a SQLite database """
         conn = None
@@ -273,10 +276,10 @@ class WorkflowEnvironment(Environment):
             self._create_table(conn)
 
         return conn
-    
+
     def close(self):
         if not hasattr(self.local, 'conn'):
-            return 
+            return
         self.local.conn.close()
 
     def _create_table(self, conn):
@@ -293,7 +296,7 @@ class WorkflowEnvironment(Environment):
             conn.commit()
         except Error as e:
             logging.error("Error occurred while creating tables: %s", e)
-    
+
     def _do_get_value(self, key: str) -> str | None:
         try:
             conn = self._get_conn()
@@ -311,7 +314,7 @@ class WorkflowEnvironment(Environment):
         super().set_value(key,str_value)
         if is_storage is False:
             return
-        
+
         try:
             conn = self._get_conn()
             conn.execute("""

src/service/aios_shell/aios_shell.py

@@ -59,6 +59,9 @@ 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.declare_user_config()
+
 
     async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool:
         target_id = msg.target.split(".")[0]
@@ -66,14 +69,14 @@ class AIOS_Shell:
         if agent is not None:
             bus.register_message_handler(target_id,agent._process_msg)
             return True
-        
+
         a_workflow = await WorkflowManager.get_instance().get_workflow(target_id)
         if a_workflow is not None:
             bus.register_message_handler(target_id,a_workflow._process_msg)
             return True
-        
+
         return False
-    
+
     async def is_agent(self,target_id:str) -> bool:
         agent : AIAgent = await AgentManager.get_instance().get(target_id)
         if agent is not None:
@@ -98,7 +101,15 @@ class AIOS_Shell:
             logger.error("openai node initial failed!")
             return False
         ComputeKernel.get_instance().add_compute_node(open_ai_node)
-        
+
+        try:
+            google_text_to_speech_node = GoogleTextToSpeechNode.get_instance()
+            google_text_to_speech_node.init()
+            ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node)
+        except Exception as e:
+            logger.error(f"google text to speech node initial failed! {e}")
+            return False
+
         llama_ai_node = LocalLlama_ComputeNode()
         await llama_ai_node.start()
         # ComputeKernel.get_instance().add_compute_node(llama_ai_node)
@@ -115,19 +126,19 @@ class AIOS_Shell:
         contact_config_path =os.path.abspath(f"{user_data_dir}/contacts.toml")
         cm = ContactManager.get_instance(contact_config_path)
         cm.load_data()
-        
+
         tunnels_config_path = os.path.abspath(f"{user_data_dir}/etc/tunnels.cfg.toml")
         tunnel_config = None
-        try: 
+        try:
             tunnel_config = toml.load(tunnels_config_path)
             if tunnel_config is not None:
                 await AgentTunnel.load_all_tunnels_from_config(tunnel_config)
         except Exception as e:
             logger.warning(f"load tunnels config from {tunnels_config_path} failed!")
-        
+
         KnowledgePipline.get_instance().initial()
-        return True 
+        return True
-        
+
 
     def get_version(self) -> str:
         return "0.5.1"
@@ -158,25 +169,25 @@ class AIOS_Shell:
             case "email":
                 tunnel_config["type"] = "EmailTunnel"
             case _:
-                error_text = FormattedText([("class:error", f"tunnel type {tunnel_type}not support!")])    
+                error_text = FormattedText([("class:error", f"tunnel type {tunnel_type}not support!")])
                 print_formatted_text(error_text,style=shell_style)
                 return None
 
-        intro_text = FormattedText([("class:prompt", tunnel_introduce)])    
+        intro_text = FormattedText([("class:prompt", tunnel_introduce)])
         print_formatted_text(intro_text,style=shell_style)
         for key,item in intpu_table.items():
             user_input = await try_get_input(f"{key} : {item.desc}")
             if user_input is None:
                 return None
-            
-            tunnel_config[key] = user_input   
 
-        return tunnel_config        
+            tunnel_config[key] = user_input
+
+        return tunnel_config
 
     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")
-        try: 
+        try:
             all_tunnels = toml.load(tunnels_config_path)
             if all_tunnels is not None:
                 all_tunnels[tunnel_config["tunnel_id"]] = tunnel_config
@@ -234,7 +245,7 @@ class AIOS_Shell:
             prompt.messages.append({"role": "user", "content":" ".join(args[1:])})
             result = await KnowledgeBase().query_prompt(prompt)
             print_formatted_text(result.as_str())
-            
+
     async def call_func(self,func_name, args):
         match func_name:
             case 'send':
@@ -251,13 +262,13 @@ class AIOS_Shell:
                     key = args[0]
                     config_item = AIStorage.get_instance().get_user_config().get_config_item(key)
                     old_value = AIStorage.get_instance().get_user_config().get_value(key)
-                    
+
                     if config_item is not None:
                         value = await session.prompt_async(f"{key} : {config_item.desc} \nCurrent : {old_value}\nPlease input new value:",style=shell_style)
                         AIStorage.get_instance().get_user_config().set_value(key,value)
                         await AIStorage.get_instance().get_user_config().save_to_user_config()
                         show_text = FormattedText([("class:title", f"set {key} to {value} success!")])
-                
+
                 return show_text
             case 'connect':
                 show_text = FormattedText([("class:title", "args error, /connect $target telegram | email")])
@@ -295,8 +306,8 @@ class AIOS_Shell:
                 if len(args) >= 1:
                     self.username = args[0]
                 AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
-                
+
-                return self.username + " login success!"    
+                return self.username + " login success!"
             case 'history':
                 num = 10
                 offset = 0
@@ -326,9 +337,9 @@ class AIOS_Shell:
                 return FormattedText([("class:title", f"help~~~")])
 
 
-##########################################################################################################################    
+##########################################################################################################################
 history = FileHistory('aios_shell_history.txt')
-session = PromptSession(history=history) 
+session = PromptSession(history=history)
 
 def parse_function_call(func_string):
     if len(func_string) > 2:
@@ -339,7 +350,7 @@ def parse_function_call(func_string):
             return func_name, params
     else:
         return None
-    
+
 async def try_get_input(desc:str,check_func:callable = None) -> str:
     user_input = await session.prompt_async(f"{desc} \nType /exit to abort. \nPlease input:",style=shell_style)
     err_str = ""
@@ -349,13 +360,13 @@ async def try_get_input(desc:str,check_func:callable = None) -> str:
                 return user_input
             else:
                 return None
-        
+
     else:
         is_ok,err_str = check_func(user_input)
         if is_ok:
             return user_input
-    
+
-    error_text = FormattedText([("class:error", err_str)])    
+    error_text = FormattedText([("class:error", err_str)])
     print_formatted_text(error_text,style=shell_style)
     return await try_get_input(desc,check_func)
 
@@ -375,7 +386,7 @@ async def main_daemon_loop(shell:AIOS_Shell):
     return 0
 
 def print_welcome_screen():
-    print("\033[1;31m")  
+    print("\033[1;31m")
     logo = """
 \t   _______                    ____________________   __
 \t   __  __ \______________________  __ \__    |__  | / /
@@ -386,9 +397,9 @@ def print_welcome_screen():
 
     """
     print(logo)
-    print("\033[0m")  
+    print("\033[0m")
 
-    print("\033[1;32m \t\tWelcome to OpenDAN - Your Personal AI OS\033[0m\n")  
+    print("\033[1;32m \t\tWelcome to OpenDAN - Your Personal AI OS\033[0m\n")
 
     introduce = """
 \tThe core goal of version 0.5.1 is to turn the concept of AIOS into code and get it up and running as quickly as possible. 
@@ -398,12 +409,12 @@ def print_welcome_screen():
 \twe intend to strengthen some components. This document will explain these changes and provide an update 
 \ton the current development progress of MVP(0.5.1,0.5.2)
 
-""" 
+"""
     print(introduce)
 
-    print(f"\033[1;34m \t\tVersion: {AIOS_Version}\n\033")  
+    print(f"\033[1;34m \t\tVersion: {AIOS_Version}\n\033")
-    print("\033[1;33m \tOpenDAN is an open-source project, let's define the future of Humans and AI together.\033[0m")  
+    print("\033[1;33m \tOpenDAN is an open-source project, let's define the future of Humans and AI together.\033[0m")
-    print("\033[1;33m \tGithub\t: https://github.com/fiatrete/OpenDAN-Personal-AI-OS\033[0m")  
+    print("\033[1;33m \tGithub\t: https://github.com/fiatrete/OpenDAN-Personal-AI-OS\033[0m")
     print("\033[1;33m \tWebsite\t: https://www.opendan.ai\033[0m")
     print("\n\n")
 
@@ -414,19 +425,19 @@ async def main():
     logging.basicConfig(filename="aios_shell.log",filemode="w",encoding='utf-8',force=True,
                         level=logging.INFO,
                         format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s')
-    
+
     if os.path.isdir(f"{directory}/../../../rootfs"):
         AIStorage.get_instance().is_dev_mode = True
     else:
-        AIStorage.get_instance().is_dev_mode = False    
+        AIStorage.get_instance().is_dev_mode = False
 
     is_daemon = False
     if os.name != 'nt':
         if os.getppid() == 1:
             is_daemon = True
 
-    shell = AIOS_Shell("user")   
+    shell = AIOS_Shell("user")
-    shell.declare_all_user_config() 
+    shell.declare_all_user_config()
     await AIStorage.get_instance().initial()
     check_result = AIStorage.get_instance().get_user_config().check_config()
     if check_result is not None:
@@ -437,7 +448,7 @@ async def main():
             #Remind users to enter necessary configurations.
             if await get_user_config_from_input(check_result) is False:
                 return 1
-    
+
     init_result = await shell.initial()
     if init_result is False:
         if is_daemon:
@@ -453,18 +464,18 @@ async def main():
     proxy.apply_storage()
 
     #TODO: read last input config
-    completer = WordCompleter(['/send $target $msg $topic', 
+    completer = WordCompleter(['/send $target $msg $topic',
-                               '/open $target $topic', 
+                               '/open $target $topic',
                                '/history $num $offset',
                                '/login $username',
                                '/connect $target',
-                               '/knowledge add email | dir', 
+                               '/knowledge add email | dir',
                                '/knowledge journal [$topn]',
                                '/knowledge query $query' 
                                '/set_config $key',
                                '/list_config',
                                '/show',
-                               '/exit', 
+                               '/exit',
                                '/help'], ignore_case=True)
 
     current = AIStorage.get_instance().get_user_config().get_value("shell.current")
@@ -472,7 +483,7 @@ async def main():
     shell.current_target = current[1]
     shell.current_topic = current[0]
 
-    await asyncio.sleep(0.2) 
+    await asyncio.sleep(0.2)
     while True:
         user_input = await session.prompt_async(f"{shell.username}<->{shell.current_topic}@{shell.current_target}$ ",completer=completer,style=shell_style)
         if len(user_input) <= 1:
@@ -493,6 +504,6 @@ async def main():
         #print_formatted_text(f"{shell.username}<->{shell.current_topic}@{shell.current_target} >>> {resp}",style=shell_style)
 
 
-if __name__ == "__main__":    
+if __name__ == "__main__":
     asyncio.run(main())