Merge branch 'MVP' into MVP

src/aios_kernel/__init__.py

@@ -1,5 +1,5 @@
 from .environment import Environment,EnvironmentEvent
-from .agent_message import AgentMsg,AgentMsgState
+from .agent_message import AgentMsg,AgentMsgStatus
 from .chatsession import AIChatSession
 from .agent import AIAgent,AIAgentTemplete,AgentPrompt
 from .compute_kernel import ComputeKernel,ComputeTask
@@ -8,4 +8,15 @@ from .open_ai_node import OpenAI_ComputeNode
 from .knowledge_base import KnowledgeBase 
 from .role import AIRole,AIRoleGroup
 from .workflow import Workflow
-from .bus import AIBus
+from .bus import AIBus
+from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent
+from .local_llama_compute_node import LocalLlama_ComputeNode
+from .whisper_node import WhisperComputeNode
+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
+
+AIOS_Version = "0.5.1, build 2023-9-17"
+

src/aios_kernel/agent.py

@@ -5,9 +5,13 @@ import asyncio
 import logging
 import uuid
 import time
+import json
 
-from .agent_message import AgentMsg
+from .agent_message import AgentMsg, AgentMsgStatus, AgentMsgType
 from .chatsession import AIChatSession
+from .compute_task import ComputeTaskResult
+from .ai_function import AIFunction
+from .environment import Environment
 
 logger = logging.getLogger(__name__)
 
@@ -69,7 +73,7 @@ class AIAgent:
         self.prompt:AgentPrompt = None
         self.llm_model_name:str = None
         self.max_token_size:int = 3600
-        self.instance_id:str = None
+        self.agent_id:str = None
         self.template_id:str = None
         self.fullname:str = None
         self.powerby = None  
@@ -77,6 +81,8 @@ class AIAgent:
 
         self.chat_db = None
         self.unread_msg = Queue() # msg from other agent
+        self.owner_env : Environment = None
+        self.owenr_bus = None
         
     @classmethod
     def create_from_templete(cls,templete:AIAgentTemplete, fullname:str):
@@ -85,7 +91,7 @@ class AIAgent:
         result_agent.llm_model_name = templete.llm_model_name
         result_agent.max_token_size = templete.max_token_size
         result_agent.template_id = templete.template_id
-        result_agent.instance_id = "agent#" + uuid.uuid4().hex
+        result_agent.agent_id = "agent#" + uuid.uuid4().hex
         result_agent.fullname = fullname
         result_agent.powerby = templete.author
         result_agent.prompt = templete.prompt
@@ -95,10 +101,10 @@ class AIAgent:
         if config.get("instance_id") is None:
             logger.error("agent instance_id is None!")
             return False
-        self.instance_id = config["instance_id"]
+        self.agent_id = config["instance_id"]
 
         if config.get("fullname") is None:
-            logger.error(f"agent {self.instance_id} fullname is None!")
+            logger.error(f"agent {self.agent_id} fullname is None!")
             return False
         self.fullname = config["fullname"]
 
@@ -123,83 +129,101 @@ class AIAgent:
             return "ignore"
         
         return "text"
+    
+    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 = []
+        for inner_func in all_inner_function:
+            this_func = {}
+            this_func["name"] = inner_func.get_name()
+            this_func["description"] = inner_func.get_description()
+            this_func["parameters"] = inner_func.get_parameters()
+            result_func.append(this_func)
+
+        return result_func 
+
+    async def _execute_func(self,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg) -> str:
+        from .compute_kernel import ComputeKernel
+
+        func_name = inenr_func_call_node.get("name")
+        arguments = json.loads(inenr_func_call_node.get("arguments"))
+
+        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)
+
+        result_str:str = await func_node.execute(**arguments)
+        
+        inner_functions = 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)
+
+        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)      
+        else:
+            return task_result.result_str
 
     async def _process_msg(self,msg:AgentMsg) -> AgentMsg:
             from .compute_kernel import ComputeKernel
+
             session_topic = msg.get_sender() + "#" + msg.topic
-            chatsession = AIChatSession.get_session(self.instance_id,session_topic,self.chat_db)
+            chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db)
+            if msg.mentions is not None:
+                if not self.agent_id in msg.mentions:
+                    chatsession.append(msg)
+                    logger.info(f"agent {self.agent_id} recv a group chat message from {msg.sender},but is not mentioned,ignore!")
+                    return None
+            
             prompt = AgentPrompt()
             prompt.append(self.prompt)
- 
-            # prompt.append(self._get_function_prompt(the_role.get_name()))
             # prompt.append(self._get_knowlege_prompt(the_role.get_name()))
             prompt.append(await self._get_prompt_from_session(chatsession)) # chat context
             
             msg_prompt = AgentPrompt()
             msg_prompt.messages = [{"role":"user","content":msg.body}]
             prompt.append(msg_prompt)
-    
-            result = await ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size)
-            final_result = result
-            result_type : str = self._get_llm_result_type(result)
+
+            inner_functions = self._get_inner_functions()
+
+            task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions)
+            final_result = task_result.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?
+                final_result = await self._execute_func(inner_func_call_node,prompt,msg)
+            
+            result_type : str = self._get_llm_result_type(final_result)
             is_ignore = False
 
             match result_type:
-                # case "function":
-                #    callchain:CallChain = self._parse_function_call_chain(result)
-                #    resp = await callchain.exec()
-                #    if callchain.have_result():
-                #        # generator proc resp prompt with WAITING state
-                #        proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession)
-                #        final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
-                #        return final_result
-
-                
-                # case "send_message":
-                #    # send message to other / sub workflow
-                #    next_msg:AgentMsg = self._parse_to_msg(result)
-                #    if next_msg is not None:
-                #        # TODO: Next Target can be another role in workflow
-                #        next_workflow:Workflow = self.get_workflow(next_msg.get_target())
-                #        inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id())
-
-                #        inner_chat_session.append_post(next_msg)
-                #        resp = await next_workflow.send_msg(next_msg)
-                #        inner_chat_session.append_recv(resp)
-                #        # generator proc resp prompt with WAITING state
-                #        proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession)
-                #        final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
-                        
-                #        return final_result
-                    
-                #case "post_message":
-                #    # post message to other / sub workflow
-                #    next_msg:AgentMsg = self._parse_to_msg(result)
-                #    if next_msg is not None:
-                #        next_workflow:Workflow = self.get_workflow(next_msg.get_target())
-                #        inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id())
-                #        inner_chat_session.append_post(next_msg)
-                #        next_workflow.post_msg(next_msg)
-
                 case "ignore":
                     is_ignore = True
                     
             if is_ignore is not True:
-                # TODO : how to get inner chat session?
-                resp_msg = AgentMsg()
-                resp_msg.set(self.instance_id,msg.sender,final_result)
-                resp_msg.topic = msg.topic
-
-                if chatsession is not None:
-                    chatsession.append_recv(msg)
-                    chatsession.append_post(resp_msg)
+                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.instance_id
+        return self.agent_id
     
     def get_fullname(self) -> str:
         return self.fullname
@@ -213,13 +237,14 @@ class AIAgent:
     def get_max_token_size(self) -> int:
         return self.max_token_size
     
-    async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt:
+    async def _get_prompt_from_session(self,chatsession:AIChatSession,is_groupchat=False) -> AgentPrompt:
+        # TODO: get prompt from group chat is different from single chat
         messages = chatsession.read_history() # read last 10 message
         result_prompt = AgentPrompt()
         for msg in reversed(messages):
             if msg.target == chatsession.owner_id:
-                result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"})
-            if msg.sender == chatsession.owner_id:
+                result_prompt.messages.append({"role":"user","content":msg.body})
+            elif msg.sender == chatsession.owner_id:
                 result_prompt.messages.append({"role":"assistant","content":msg.body})
         
         return result_prompt

src/aios_kernel/agent_message.py

@@ -1,27 +1,102 @@
 from enum import Enum
 import uuid
 import time 
+import re
 
-class AgentMsgState(Enum):
+class AgentMsgType(Enum):
+    TYPE_MSG = 0
+    TYPE_INTERNAL_CALL = 1
+    TYPE_ACTION = 2
+    TYPE_EVENT = 3
+
+class AgentMsgStatus(Enum):
     RESPONSED = 0
     INIT = 1
     SENDING = 2
     PROCESSING = 3
     ERROR = 4
+    RECVED = 5
+    EXECUTED = 6
+
+# msg is a msg / msg resp
+# msg body可以有内容类型（MIME标签），text, image, voice, video, file,以及富文本(html)
+# msg is a inner function call with result
+# msg is a Action with result
+
+# qutoe Msg
+# forword msg
+# reply msg
+
+# 逻辑上的同一个Message在同一个session中看到的msgid相同
+#       在不同的session中看到的msgid不同
 
 class AgentMsg:
-    def __init__(self) -> None:
+    def __init__(self,msg_type=AgentMsgType.TYPE_MSG) -> None:
+        self.msg_id = ""
+        self.msg_type:AgentMsgType = msg_type
+        
+        self.prev_msg_id:str = None
+        self.quote_msg_id:str = None    
+        self.rely_msg_id:str = None # if not none means this is a respone msg
+        self.session_id:str = None
+
         self.create_time = 0
-        self.sender:str = None
+        self.done_time = 0
+        self.topic:str = None # topic is use to find session, not store in db
+
+        self.sender:str = None # obj_id.sub_objid@tunnel_id
         self.target:str = None
+        self.mentions:[] = None #use in group chat only
+        #self.title:str = None
         self.body:str = None
-        self.topic:str = "T#" + uuid.uuid4().hex
-        #self.msg_type = 0
-        self.state = AgentMsgState.INIT
+        self.body_mime:str = None #//default is "text/plain",encode is utf8
+
+        #type is call / action
+        self.func_name = None 
+        self.args = None
+        self.result_str = None
+
+        #type is event
+        self.event_name = None
+        self.event_args = None
+
+        self.status = AgentMsgStatus.INIT
+        self.inner_call_chain = []
         self.resp_msg = None
 
+    @classmethod
+    def create_internal_call_msg(self,func_name:str,args:dict,prev_msg_id:str,caller:str):
+        msg = AgentMsg(AgentMsgType.TYPE_INTERNAL_CALL)
+        msg.func_name = func_name
+        msg.args = args
+        msg.prev_msg_id = prev_msg_id
+        msg.sender = caller
+        return msg
+    
+    def create_action_msg(self,action_name:str,args:dict,caller:str):
+        msg = AgentMsg(AgentMsgType.TYPE_ACTION)
+        msg.func_name = action_name
+        msg.args = args
+        msg.prev_msg_id = self.msg_id
+        msg.topic  = self.topic
+        msg.sender = caller
+        return msg
+    
+    def create_resp_msg(self,resp_body):
+        resp_msg = AgentMsg()
+        resp_msg.msg_id = "msg#" + uuid.uuid4().hex
+        self.create_time = time.time()
+
+        resp_msg.rely_msg_id = self.msg_id
+        resp_msg.sender = self.target
+        resp_msg.target = self.sender
+        resp_msg.body = resp_body
+        resp_msg.topic = self.topic
+
+        return resp_msg
+
     def set(self,sender:str,target:str,body:str,topic:str=None) -> None:
-        self.id = "msg#" + uuid.uuid4().hex
+        self.msg_id = "msg#" + uuid.uuid4().hex
         self.sender = sender
         self.target = target
         self.body = body
@@ -30,10 +105,35 @@ class AgentMsg:
             self.topic = topic
 
     def get_msg_id(self) -> str:
-        return self.id
+        return self.msg_id
 
     def get_sender(self) -> str:
         return self.sender
 
     def get_target(self) -> str:
         return self.target
+    
+    def get_prev_msg_id(self) -> str:
+        return self.prev_msg_id
+    
+    def get_quote_msg_id(self) -> str:
+        return self.quote_msg_id
+    
+    @classmethod
+    def parse_function_call(cls,func_string:str):
+        match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string)
+        if not match:
+            return None
+
+        func_name = match.group(1)
+        if func_name is None:
+            return None
+        if len(func_name) < 2:
+            return None
+        
+        params_string = match.group(2).strip()    
+        params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string)
+        params = [param.strip('"') for param in params]
+
+        return func_name, params
+        

src/aios_kernel/ai_function.py

@@ -1,25 +1,74 @@
+from abc import ABC, abstractmethod
+from typing import Dict,Coroutine,Callable
+
 class AIFunction:
     def __init__(self) -> None:
-        self.intro : str = None
+        self.description : str = None
     
-    def load_from_config(self,config:dict) -> bool:
+    @abstractmethod
+    def get_name(self) -> str:
+        """
+        return the name of the function (should be snake case)
+        """
         pass
 
+    @abstractmethod
+    def get_description(self) -> str:
+        """
+        return a detailed description of what the function does
+        """
+        return self.description
+
+    @abstractmethod
+    def get_parameters(self) -> Dict:
+        """
+        Return the list of parameters to execute this function in the form of
+        JSON schema as specified in the OpenAI documentation:
+        https://platform.openai.com/docs/api-reference/chat/create#chat/create-parameters
+
+        str = run_code(code:str)
+        parameters = {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "Python code which needs to be executed"
+                }
+            }
+        }
+
+        """
+        pass
+    
+    @abstractmethod
+    async def execute(self, **kwargs) -> str:
+        """
+        Execute the function and return a JSON serializable dict.
+        The parameters are passed in the form of kwargs
+        """
+        pass
+
+    @abstractmethod
     def is_local(self) -> bool:
+        """
+        is this function call need network?
+        """
         pass
 
+    @abstractmethod
     def is_in_zone(self) -> bool:
-        pass
-    
-    def is_readyonly(self) -> bool:
+        """
+        is this function call in Lan?
+        """
         pass
 
-    def get_intro(self) -> str:
-        return self.intro
-    
-    async def execute(self):
+    @abstractmethod  
+    def is_ready_only(self) -> bool:
         pass
 
+    #def load_from_config(self,config:dict) -> bool:
+    #    pass
+
 # call chain is a combination of ai_function,group of ai_function.
 class CallChain:
     def __init__(self) -> None:
@@ -29,4 +78,35 @@ class CallChain:
         pass
 
     async def execute(self):
-        pass
+        pass
+
+class SimpleAIFunction(AIFunction):
+    def __init__(self,func_id:str,description:str,func_handler:Coroutine,parameters:Dict = None) -> None:
+        self.func_id = func_id
+        self.description = description
+        self.func_handler = func_handler
+        self.parameters = parameters
+
+    def get_name(self) -> str: 
+        return self.func_id
+    
+    def get_parameters(self) -> Dict:
+        if self.parameters is not None:
+            return self.parameters
+        return {"type": "object", "properties": {}}
+
+    async def execute(self,**kwargs) -> str:
+        if self.func_handler is None:
+            return "error: function not implemented"
+        
+        return await self.func_handler(**kwargs)
+
+    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/bus.py

@@ -1,5 +1,5 @@
-from typing import Any
-from .agent_message import AgentMsg,AgentMsgState
+from typing import Coroutine,Dict,Any
+from .agent_message import AgentMsg,AgentMsgStatus
 import asyncio
 from asyncio import Queue
 
@@ -8,19 +8,32 @@ import logging
 logger = logging.getLogger(__name__)
 
 class AIBusHandler:
-    def __init__(self,handler:Any) -> None:
+    def __init__(self,handler:Coroutine,owner_bus,enable_defualt_proc=True) -> None:
         self.handler = handler
         self.working_task = None
-        self.results = {}
+        self.results = {} # recv resps
         self.queue:Queue = Queue()
+        self.enable_defualt_proc = enable_defualt_proc
+        self.owner_bus = owner_bus
 
     async def handle_message(self,msg:AgentMsg) -> Any:
         if self.handler is None:
             return None
         
-        return await self.handler(msg)
+        if self.enable_defualt_proc:
+            # do default process
+            if msg.rely_msg_id is not None:
+                self.results[msg.rely_msg_id] = msg
+                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
@@ -30,10 +43,13 @@ class AIBus:
         return cls._instance
 
     def __init__(self) -> None:
-        self.handlers = {}
-        self.unhandle_handler = None
-        
-    async def post_message(self,target_id,msg:AgentMsg,use_unhandle=True) -> bool:
+        self.handlers:Dict[AIBusHandler] = {}
+        self.unhandle_handler:Coroutine = None
+
+
+    async def post_message(self,msg:AgentMsg,use_unhandle=True) -> bool:
+        target_id = msg.target.split(".")[0]
+
         handler = self.handlers.get(target_id)
         if handler:
             handler.queue.put_nowait(msg)
@@ -43,45 +59,39 @@ class AIBus:
         if use_unhandle:
             if self.unhandle_handler is not None:
                 if await self.unhandle_handler(self,msg):
-                    return await self.post_message(target_id,msg,False)
+                    return await self.post_message(msg,False)
       
         logger.warn(f"post message to {msg.target} failed!,target not found")
         return False
 
-    def resp_message(self,my_id:str,org_msg_id:str,resp:AgentMsg) -> None:
-        handler = self.handlers.get(my_id)
-        if handler is None:
-            return None
-        handler.results[org_msg_id] = resp
+    async def resp_message(self,org_msg_id:str,resp:AgentMsg) -> None:
+        assert resp.rely_msg_id == org_msg_id
+        return await self.post_message(resp)
 
-    async def get_message_resp(self,name:str,msg_id:str) -> AgentMsg:
-        handler = self.handlers.get(name)
-        if handler is None:
+    async def send_message(self,msg:AgentMsg) -> AgentMsg:
+        sender_id = msg.sender.split(".")[0]
+        sender_handler = self.handlers.get(sender_id) # sender already register on bus
+        if sender_handler is None:
+            logger.warn(f"sender {sender_id} not register on AI_BUS!")
             return None
         
-        return handler.results.get(msg_id)
-
-    async def send_message(self,target_id:str,msg:AgentMsg) -> AgentMsg:
-        post_result = await self.post_message(target_id,msg)
+        post_result = await self.post_message(msg)
         if post_result is False:
             return None
-        
-        handler = self.handlers.get(target_id)
-        if handler is None:
-            return None
-                
+
         retry_times = 0
         while True:
-            resp = handler.results.get(msg.id)
+            resp = sender_handler.results.get(msg.msg_id)
             if resp is not None:
                 msg.resp_msg = resp
-                msg.state = AgentMsgState.RESPONSED
+                msg.status = AgentMsgStatus.RESPONSED
+                del sender_handler.results[msg.msg_id]
                 return resp
             
             await asyncio.sleep(0.2)
             retry_times += 1
             if retry_times > 100:
-                msg.state = AgentMsgState.ERROR
+                msg.status = AgentMsgStatus.ERROR
                 return None
             
         return None
@@ -91,7 +101,7 @@ class AIBus:
 
     # means sub
     def register_message_handler(self,handler_name:str,handler:Any) -> Queue:        
-        handler_node =  AIBusHandler(handler) 
+        handler_node =  AIBusHandler(handler,self) 
         self.handlers[handler_name] = handler_node
         return handler_node.queue
     
@@ -100,13 +110,12 @@ class AIBus:
             # Wait for a message
             message = await handler.queue.get()
 
-            try:
+            #try:
                 # Try to handle the message
-                result = await handler.handle_message(message)
-                handler.results[message.id] = result
-            except Exception as e:
+            await handler.handle_message(message)
+            #except Exception as e:
                 # If an error occurs, put the message back into the queue
-                logger.error(f"handle message {message.id} failed! {e}")
+            #    logger.error(f"handle message {message.msg_id} failed! {e}")
                 #self.queues[name].put_nowait(message)
         
         return
@@ -124,6 +133,4 @@ class AIBus:
             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

@@ -5,8 +5,9 @@ import logging
 import threading
 import datetime
 import uuid
+import json
 
-from .agent_message import AgentMsg
+from .agent_message import AgentMsgType, AgentMsg, AgentMsgStatus
 
 class ChatSessionDB:
     def __init__(self, db_file):
@@ -54,14 +55,31 @@ class ChatSessionDB:
             """)
 
             # create messages table
+            # reciver_id could be None
+        
             conn.execute("""
                 CREATE TABLE IF NOT EXISTS Messages (
                     MessageID TEXT PRIMARY KEY,
                     SessionID TEXT,
-                    SenderID TEXT,
+                    MsgType INTEGER,
+                    PrevMsgID TEXT,
+                    QuoteMsgID TEXT,
+                    RelyMsgID TEXT,
+                    
+                    SenderID TEXT, 
                     ReceiverID TEXT,
                     Timestamp TEXT,
+                    
+                    Topic TEXT,
+                    Mentions TEXT,
+                    ContentMIME TEXT,
                     Content TEXT,
+                    
+                    ActionName TEXT,
+                    ActionParams TEXT,
+                    ActionResult TEXT,
+                    DoneTime TEXT,     
+                         
                     Status INTEGER
                 );
             """)
@@ -83,15 +101,43 @@ class ChatSessionDB:
             logging.error("Error occurred while inserting session: %s", e)
             return -1  # return -1 if an error occurs
 
-    def insert_message(self, message_id, session_id, sender_id, receiver_id, timestamp, content, status):
+    def insert_message(self, msg:AgentMsg):
         """ insert a new message into the Messages table """
         try:
+            action_name = None
+            action_params = None
+            action_result = None
+            mentions = None
+            if msg.mentions:
+                mentions = json.dumps(msg.mentions)
+
+            match msg.msg_type:
+                case AgentMsgType.TYPE_MSG:
+                    pass
+                case AgentMsgType.TYPE_ACTION:
+                    action_name = msg.func_name
+                    action_params = json.dumps(msg.args)
+                    action_result = msg.result_str
+                case AgentMsgType.TYPE_INTERNAL_CALL:
+                    action_name = msg.func_name
+                    action_params = json.dumps(msg.args)
+                    action_result = msg.result_str
+                case AgentMsgType.TYPE_EVENT:
+                    action_name = msg.event_name
+                    action_params = json.dumps(msg.event_args)
+
+
             conn = self._get_conn()
             conn.execute("""
-                INSERT INTO Messages (MessageID, SessionID, SenderID, ReceiverID, Timestamp, Content, Status)
-                VALUES (?, ?, ?, ?, ?, ?, ?)
-            """, (message_id, session_id, sender_id, receiver_id, timestamp, content, status))
+                INSERT INTO Messages (MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status)
+                VALUES (?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (msg.msg_id, msg.session_id, msg.msg_type.value, msg.prev_msg_id, msg.sender, msg.target, msg.create_time, msg.topic,mentions,msg.body_mime,msg.body,action_name,action_params,action_result,msg.done_time,msg.status.value))
             conn.commit()
+
+            if msg.inner_call_chain:
+                for inner_call in msg.inner_call_chain:
+                    self.insert_message(inner_call)
+
             return 0  # return 0 if successful
         except Error as e:
             logging.error("Error occurred while inserting message: %s", e)
@@ -134,7 +180,7 @@ class ChatSessionDB:
         """Get a message by its ID"""
         conn =self._get_conn()
         c = conn.cursor()
-        c.execute("SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages WHERE MessageID = ?", (message_id,))
+        c.execute("SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages WHERE MessageID = ?", (message_id,))
         message = c.fetchone()
         return message
 
@@ -144,7 +190,7 @@ class ChatSessionDB:
             conn = self._get_conn()
             cursor = conn.cursor()
             cursor.execute("""
-                SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages
+                SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages
                 WHERE SessionID = ?
                 ORDER BY Timestamp DESC
                 LIMIT ? OFFSET ?
@@ -222,29 +268,31 @@ class AIChatSession:
         result = []
         for msg in msgs:
             agent_msg = AgentMsg()
-            agent_msg.id = msg[0]
-            agent_msg.sender = msg[2]
-            agent_msg.target = msg[3]
-            agent_msg.create_time = msg[4]
-            agent_msg.body = msg[5]
-            # agent_msg.state = msg[6]
+            agent_msg.msg_id = msg[0]
+            agent_msg.session_id = msg[1]
+            agent_msg.msg_type = AgentMsgType(msg[2])
+            agent_msg.prev_msg_id = msg[3]
+            agent_msg.sender = msg[4]
+            agent_msg.target = msg[5]
+            agent_msg.create_time = msg[6]
+            agent_msg.topic = msg[7]
+            if msg[8] is not None:
+                agent_msg.mentions = json.loads(msg[8])
+            agent_msg.body_mime = msg[9]
+            agent_msg.body = msg[10]
+            agent_msg.func_name = msg[11]
+            if msg[12] is not None:
+                agent_msg.args = json.loads(msg[12])
+            agent_msg.result_str = msg[13]
+            agent_msg.done_time = msg[14]
+            agent_msg.status = AgentMsgStatus(msg[15])
 
             result.append(agent_msg)
         return result
 
     def append(self,msg:AgentMsg) -> None:
-        self.db.insert_message(msg.id,self.session_id,msg.sender,msg.target,msg.create_time,msg.body,0)
-
-    def append_post(self,msg:AgentMsg) -> None:
-        """append msg to session, msg is post from session (owner => msg.target)"""
-        assert msg.sender == self.owner_id,"post message means msg.sender == self.owner_id"
-        self.append(msg)
-        
-
-    def append_recv(self,msg:AgentMsg) -> None:
-        """append msg to session, msg is recv from msg'sender (msg.sender => owner)"""
-        assert msg.target == self.owner_id,"recv message means msg.target == self.owner_id"
-        self.append(msg)        
+        msg.session_id = self.session_id
+        self.db.insert_message(msg)
 
     #def attach_event_handler(self,handler) -> None:
     #    """chat session changed event handler"""

src/aios_kernel/compute_kernel.py

@@ -6,97 +6,96 @@ 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
 
 logger = logging.getLogger(__name__)
 
 # How to dispatch different computing tasks (some tasks may contain a large amount of state for correct execution)
-# to suitable computing nodes, achieving a balance of speed, cost, and power consumption, 
+# to suitable computing nodes, achieving a balance of speed, cost, and power consumption,
 # is the CORE GOAL of the entire computing task schedule system (aios_kernel).
+
+
 class ComputeKernel:
     _instance = None
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super().__new__(cls)
-            cls._instance.is_start = False
-
+            cls._instance = ComputeKernel()
         return cls._instance
-
+    
     def __init__(self) -> None:
-        if self.is_start is True:
-            return
-        
-        self.is_start = True
+        self.is_start = False
         self.task_queue = Queue()
         self.is_start = False
         self.compute_nodes = {}
 
         self.start()
 
-    def run(self,task:ComputeTask) -> None:
+    def run(self, task: ComputeTask) -> None:
         # check there is compute node can support this task
         if self.is_task_support(task) is False:
-            logger.error(f"task {task.display()} is not support by any compute node")
+            logger.error(
+                f"task {task.display()} is not support by any compute node")
             return
         # add task to working_queue
         self.task_queue.put_nowait(task)
-        
 
     def start(self):
         if self.is_start is True:
             logger.warn("compute_kernel is already start")
             return
-        
+
         self.is_start = True
+
         async def _run_task_loop():
             while True:
-                logger.info("compute_kernel is waiting for task...")
                 task = await self.task_queue.get()
                 logger.info(f"compute_kernel get task: {task.display()}")
-                c_node:ComputeNode = self._schedule(task)
+                c_node: ComputeNode = self._schedule(task)
                 await c_node.push_task(task)
-            
-            logger.warn("compute_kernel is stoped!")
-        
-        asyncio.create_task(_run_task_loop())
-        
 
-    def _schedule(self,task) -> ComputeNode:
+            logger.warn("compute_kernel is stoped!")
+
+        asyncio.create_task(_run_task_loop())
+
+    def _schedule(self, task) -> ComputeNode:
         for node in self.compute_nodes.values():
             if node.is_support(task) is True:
                 return node
-        logger.warning(f"task {task.display()} is not support by any compute node")
+        logger.warning(
+            f"task {task.display()} is not support by any compute node")
         return None
 
-    def add_compute_node(self,node:ComputeNode):
+    def add_compute_node(self, node: ComputeNode):
         if self.compute_nodes.get(node.node_id) is not None:
-            logger.warn(f"compute_node {node.display()} already in compute_kernel")
+            logger.warn(
+                f"compute_node {node.display()} already in compute_kernel")
             return
         self.compute_nodes[node.node_id] = node
         logger.info(f"add compute_node {node.display()} to compute_kernel")
 
-    def disable_compute_node(self,node_id:str):
+    def disable_compute_node(self, node_id: str):
         node = self.compute_nodes.get(node_id)
         if node is None:
             logger.warn(f"compute_node {node_id} not in compute_kernel")
             return
         node.enable = False
 
-    def is_task_support(self,task:ComputeTask) -> bool:
+    def is_task_support(self, task: ComputeTask) -> bool:
         return True
 
-
     # friendly interface for use:
-    def llm_completion(self,prompt:AgentPrompt,mode_name:Optional[str] = None,max_token:int = 0):
+    def llm_completion(self, prompt: AgentPrompt, mode_name: Optional[str] = None, max_token: int = 0,inner_functions = None):
         # craete a llm_work_task ,push on queue's end
         # then task_schedule would run this task.(might schedule some work_task to another host)
         task_req = ComputeTask()
-        task_req.set_llm_params(prompt,mode_name,max_token)
+        task_req.set_llm_params(prompt, mode_name, max_token,inner_functions)
         self.run(task_req)
         return task_req
 
-    async def do_llm_completion(self,prompt:AgentPrompt,mode_name:Optional[str] = None,max_token:int = 0) -> str:
-        task_req = self.llm_completion(prompt,mode_name,max_token)
+    async def do_llm_completion(self, prompt: AgentPrompt, mode_name: Optional[str] = None, max_token: int = 0, inner_functions = None) -> str:
+        task_req = self.llm_completion(prompt, mode_name, max_token,inner_functions)
+
         async def check_timer():
             check_times = 0
             while True:
@@ -106,19 +105,19 @@ class ComputeKernel:
                 if task_req.state == ComputeTaskState.ERROR:
                     break
 
-                if check_times >=  20:
+                if check_times >= 20:
                     task_req.state = ComputeTaskState.ERROR
                     break
 
                 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_str
-              
+            return task_req.result
+
         return "error!"
-    
+
     def text_embedding(self,input:str,model_name:Optional[str] = None):
         task_req = ComputeTask()
         task_req.set_text_embedding_params(input,model_name)
@@ -148,4 +147,5 @@ class ComputeKernel:
             return task_req.result.result
               
         return "error!"
-    
+    
+

src/aios_kernel/compute_node.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from .compute_task import ComputeTask
+from .compute_task import ComputeTask, ComputeTaskType
 
 
 class ComputeNode(ABC):
@@ -8,15 +8,15 @@ class ComputeNode(ABC):
         self.enable = True
 
     @abstractmethod
-    async def push_task(self,task:ComputeTask,proiority:int = 0):
-        pass
-    
-    @abstractmethod
-    async def remove_task(self,task_id:str):
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
         pass
 
     @abstractmethod
-    def get_task_state(self,task_id:str):
+    async def remove_task(self, task_id: str):
+        pass
+
+    @abstractmethod
+    def get_task_state(self, task_id: str):
         pass
 
     @abstractmethod
@@ -37,17 +37,13 @@ class ComputeNode(ABC):
 
     def is_trusted(self) -> bool:
         return True
-    
+
     def get_fee_type(self) -> str:
         return "free"
-    
-    
 
 class LocalComputeNode(ComputeNode):
     def display(self) -> str:
         return super().display()
-    
-    def is_local(self) -> bool:
-        return True
-    
 
+    def is_local(self) -> bool:
+        return True

src/aios_kernel/compute_task.py

@@ -3,6 +3,7 @@ from enum import Enum
 import uuid
 import time
 
+
 class ComputeTaskState(Enum):
     DONE = 0
     INIT = 1
@@ -10,33 +11,47 @@ class ComputeTaskState(Enum):
     ERROR = 3
     PENDING = 4
 
+class ComputeTaskType(Enum):
+    NONE = -1
+    LLM_COMPLETION = 0
+    TEXT_2_IMAGE = 1
+    IMAGE_2_IMAGE = 2
+    VOICE_2_TEXT = 3
+    TEXT_2_VOICE = 4
+
 
 class ComputeTask:
     def __init__(self) -> None:
-        self.task_type = "llm_completion"
+        self.task_type = ComputeTaskType.NONE
         self.create_time = None
 
-        self.task_id:str = None
-        self.callchain_id:str = None
-        self.params:dict = {}
-        self.refers:dict = None
-        self.pading_data:bytearray = None
+        self.task_id: str = None
+        self.callchain_id: str = None
+        self.params: dict = {}
+        self.refers: dict = None
+        self.pading_data: bytearray = None
 
         self.state = ComputeTaskState.INIT
         self.result = None
         self.error_str = None
 
-    def set_llm_params(self,prompts,model_name,max_token_size,callchain_id = None):
-        self.task_type = "llm_completion"
+    def set_llm_params(self, prompts, model_name, max_token_size, inner_functions = None, callchain_id=None):
+        self.task_type = ComputeTaskType.LLM_COMPLETION
         self.create_time = time.time()
         self.task_id = uuid.uuid4().hex
         self.callchain_id = callchain_id
         self.params["prompts"] = prompts.messages
         if model_name is not None:
             self.params["model_name"] = model_name
-        else: 
+        else:
             self.params["model_name"] = "gpt-4-0613"
-        self.params["max_token_size"] = max_token_size
+        if max_token_size is None:
+            self.params["max_token_size"] = 4000
+        else:
+            self.params["max_token_size"] = max_token_size
+
+        if inner_functions is not None:
+            self.params["inner_functions"] = inner_functions
 
     def set_text_embedding_params(self, input, model_name=None, callchain_id = None):
         self.task_type = "text_embedding"
@@ -56,16 +71,15 @@ class ComputeTask:
 class ComputeTaskResult:
     def __init__(self) -> None:
         self.create_time = None
-        self.task_id:str = None    
-        self.callchain_id:str = None 
-        self.worker_id:str = None
-        self.result_code:int = 0
-        self.result_str:str = None   
+        self.task_id: str = None
+        self.callchain_id: str = None
+        self.worker_id: str = None
+        self.result_code: int = 0
+        self.result_str: str = None # easy to use,can read from result
+        self.result_message: dict = {}
+        self.result_refers: dict = None
+        self.pading_data: bytearray = None
 
-        self.result:dict = {}
-        self.result_refers:dict = None
-        self.pading_data:bytearray = None
-
-    def set_from_task(self,task:ComputeTask):
+    def set_from_task(self, task: ComputeTask):
         self.task_id = task.task_id
         self.callchain_id = task.callchain_id

src/aios_kernel/contact_manager.py

@@ -0,0 +1,34 @@
+from typing import List
+
+class Contact:
+    def __init__(self,name:str) -> None:
+        self.name = name
+        self.tags = []
+
+    def is_zone_owner(self,zone_id=None) -> bool:
+        return True
+
+    def get_tags(self)->List[str]:
+        return self.tags
+
+    def get_name(self)->str:
+        return self.name
+
+
+class ContactManager:
+    _instance = None
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            cls._instance = ContactManager()
+        return cls._instance
+
+    def __init__(self) -> None:
+        self.contacts = {}
+        self.contacts["liuzhicong"] = Contact("liuzhicong")
+
+    #def get_by_addr(self,addr:str) -> Contact:
+    #    pass
+
+    def get_by_name(self,name:str) -> Contact:
+        return self.contacts.get(name)

src/aios_kernel/email_tunnel.py

@@ -0,0 +1,143 @@
+import asyncio
+import aiosmtplib
+import aioimaplib
+import email
+from email.header import decode_header
+import mailparser
+import logging
+import time
+import datetime
+from .tunnel import AgentTunnel
+from .agent_message import AgentMsg
+
+from email.message import EmailMessage
+
+logger = logging.getLogger(__name__)
+
+class EmailTunnel(AgentTunnel):
+    @classmethod
+    def register_to_loader(cls):
+        async def load_email_tunnel(config:dict) -> AgentTunnel:
+            result_tunnel = EmailTunnel()
+            if await result_tunnel.load_from_config(config):
+                return result_tunnel
+            else:
+                return None
+            
+        AgentTunnel.register_loader("EmailTunnel",load_email_tunnel)
+
+    async def load_from_config(self,config:dict)->bool:
+        self.target_id = config["target"]
+        self.tunnel_id = config["tunnel_id"]
+
+        self.type = "TelegramTunnel"
+        self.email = config["email"]
+        self.imap_server = config["imap"]
+        s = self.imap_server.split(":")
+        if len(s) == 2:
+            self.imap_server = s[0]
+            self.imap_port = int(s[1])
+
+        self.smtp_server = config["smtp"]   
+        s = self.smtp_server.split(":")
+        if len(s) == 2:
+            self.smtp_server = s[0]
+            self.smtp_port = int(s[1])
+
+        self.login_user = config["user"]
+        self.login_password = config["password"]
+        self.folder = config["folder"]
+        self.check_interval = config["interval"]
+
+        return True  
+    
+    def __init__(self) -> None:
+        super().__init__()
+        self.is_start = False
+        self.read_email = {}
+
+    async def on_new_email(self,mail:mailparser.MailParser) -> None:
+        remote_email_addr = mail.from_[0][1]
+        remote_user_name = remote_email_addr.split("@")[0] 
+        agent_msg = self.conver_mail_to_agent_msg(mail)
+        agent_msg.sender = remote_user_name
+        agent_msg.target = self.target_id
+        self.ai_bus.register_message_handler(remote_user_name, self._process_message)
+
+        resp_msg = await self.ai_bus.send_message(agent_msg)
+        if resp_msg is None:
+            await self.reply_email(remote_email_addr,"Sorry, I can't understand your message","")
+        else:
+            if resp_msg.body_mime is None:
+                await self.reply_email(remote_email_addr,"result",resp_msg.body)
+
+    async def reply_email(self,target_email:str,title:str,msg:str) -> None:
+        email_msg = EmailMessage()
+        email_msg['Subject'] = f"Reply: {title}"
+        email_msg['From'] = self.email
+        email_msg['To'] = target_email
+        email_msg.set_content(msg)
+
+        await aiosmtplib.send(
+            email_msg,
+            hostname = self.smtp_server,
+            port=self.smtp_port,
+            username=self.login_user,
+            password=self.login_password,
+            )
+
+
+
+    def conver_mail_to_agent_msg(self,mail:mailparser.MailParser) -> AgentMsg:
+        msg = AgentMsg()
+        msg.set("",self.target_id,mail.text_plain[0])
+        msg.topic = "email"
+        return msg
+
+    async def check_email(self) -> None:
+        self.last_check_num = 0
+        self.last_check_time = datetime.datetime.now()
+        while True:
+            if self.is_start == False:
+                return
+            
+            await asyncio.sleep(self.check_interval)
+            imap_client = aioimaplib.IMAP4_SSL(host=self.imap_server,port=self.imap_port)
+            await imap_client.wait_hello_from_server()
+            await imap_client.login(self.login_user, self.login_password)
+
+            date_since = self.last_check_time.strftime("%d-%b-%Y")
+            
+            await imap_client.select(self.folder)
+            status, messages = await imap_client.search('UNSEEN',charset='US-ASCII')
+            self.last_check_time = datetime.datetime.now()
+            if status == "OK":
+                message_numbers = messages[0].split()
+                for num in message_numbers:
+                    num = int(num)
+                    if self.read_email.get(num) is not None:
+                        continue
+                    
+                    status, email_data = await imap_client.fetch(str(num), "(RFC822)")
+                    if status == "OK":
+                        #r = email.message_from_bytes(email_data[1])
+                        mail = mailparser.parse_from_bytes(email_data[1])
+                        self.read_email[num] = mail
+                        await self.on_new_email(mail)
+
+            await imap_client.logout()
+
+    async def start(self) -> bool:
+        if self.is_start:
+            logger.warning(f"tunnel {self.tunnel_id} is already started")
+            return False
+        self.is_start = True   
+
+        asyncio.create_task(self.check_email())
+        return True
+
+    async def close(self) -> None:
+        self.is_start = False
+
+    async def _process_message(self, msg: AgentMsg) -> None:
+        logger.warn(f"process message {msg.msg_id} from {msg.sender} to {msg.target}")

src/aios_kernel/environment.py

@@ -2,20 +2,134 @@
 # we have some built-in environment: Calender(include timer),Home(connect to IoT device in your home), ,KnwoledgeBase,FileSystem,
 
 from abc import ABC, abstractmethod
-from typing import Callable
+from typing import Any, Callable, Optional,Dict,Awaitable,List
+import logging
+
+from .ai_function import AIFunction
+
+logger = logging.getLogger(__name__)
 
 class EnvironmentEvent(ABC):
     @abstractmethod
     def display(self) -> str:
         pass    
 
+EnvironmentEventHandler = Callable[[str,EnvironmentEvent],Awaitable[Any]]
+
 class Environment:
-    def __init__(self) -> None:
-        pass
+    _all_env = {}
+    @classmethod
+    def get_env_by_id(cls,env_id:str):
+        return cls._all_env.get(env_id)
+    
+    @classmethod
+    def set_env_by_id(cls,id,env):
+        assert id == env.get_id()
+        cls._all_env[env.get_id()] = env
+
+    def __init__(self,env_id:str) -> None:
+        self.env_id = env_id
+        self.values:Dict[str,str] = {}
+        self.get_handlers:Dict[str,Callable] = {}
+        self.owner_env:Dict[str,Environment] = {}
+        # self.valid_keys:Dict[str,bool] = None
+        self.event_handlers:Dict[str,List[EnvironmentEventHandler]]= {}
+
+        self.functions : Dict[str,AIFunction] = {}
 
     def get_id(self) -> str:
+        return self.env_id
+    
+    def add_owner_env(self,env) -> None:
+        self.owner_env[env.get_id()] = env
+
+    #@abstractmethod
+    #TODO: how to use env? different env has different prompt
+    #def get_env_prompt(self) -> str:
+    #    pass
+
+    def add_ai_function(self,func:AIFunction) -> None:
+        if self.functions.get(func.get_name()) is not None:
+            logger.warn(f"add ai_function {func.get_name()} in env {self.env_id}:function already exist")
+            
+        self.functions[func.get_name()] = func
+
+    def get_ai_function(self,func_name:str) -> AIFunction:
+        return self.functions.get(func_name)
+
+    #def enable_ai_function(self,func_name:str) -> None:
+    #    pass
+
+    #def disable_ai_function(self,func_name:str) -> None:
+    #    pass
+
+    def get_all_ai_functions(self) -> List[AIFunction]:
+        return self.functions.values()
+
+    @abstractmethod
+    def _do_get_value(self,key:str) -> Optional[str]:
         pass
 
+    def register_get_handler(self,key:str,handler:Callable) -> None:
+        h = self.get_handlers.get(key)
+        if h is not None:
+            logger.warn(f"register get_handler {key} in env {self.env_id}:handler already exist")
+
+        self.get_handlers[key] = handler
+
+
     def attach_event_handler(self,event_id:str,handler:Callable) -> None:
-        pass
+        handler_list = self.event_handlers.get(event_id)
+        if handler_list is None:
+            handler_list = []
+            self.event_handlers[event_id] = handler_list
+
+        handler_list.append(handler)
+
+    def remove_event_handler(self,event_id:str,handler:Callable) -> None:
+        handler_list = self.event_handlers.get(event_id)
+        if handler is not None:
+            handler_list.remove(handler)
+            return
+
+        logger.warn(f"remove event_handler {event_id} in env {self.env_id}:handler not found")
+    
+    async def fire_event(self,event_id:str,event:EnvironmentEvent) -> None:
+        handler_list = self.event_handlers.get(event_id)
+        if handler_list is not None:
+            for handler in handler_list:
+                await handler(self.env_id,event)
+        else:
+            logger.debug(f"fire event {event_id} in env {self.env_id}:handler not found")
+        return
+    
+    def __getitem__(self, key):
+        return self.get_value(key)
+    
+    def get_value(self,key:str) -> Optional[str]:
+        handler = self.get_handlers.get(key)
+        if handler is not None:
+            return handler()
+    
+        s = self.values.get(key)
+        if isinstance(s,str):
+            return s
+        else:
+            logger.warn(f"get value {key} in env {self.env_id} failed!,type is not str")
+
+        s = self._do_get_value(key)
+        if s is not None:
+            return s       
+        if self.owner_env is not None:
+            for env in self.owner_env.values():
+                s = env.get_value(key)
+                if s is not None:
+                    return s
+
+        logger.warn(f"get value {key} in env {self.env_id} failed!,not found")
+        return None
+    
+    def set_value(self, key: str, str_value: str,is_storage:bool = True):
+        logger.info(f"set value {key} in env {self.env_id} to {str_value}")
+        self.values[key] = str_value
 

src/aios_kernel/google_text_to_speech_node.py

@@ -0,0 +1,104 @@
+
+import os
+import asyncio
+from asyncio import Queue
+import logging
+
+from google.cloud import texttospeech
+
+from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
+from .compute_node import ComputeNode
+
+logger = logging.getLogger(__name__)
+
+
+"""
+You need to set the GOOGLE_APPLICATION_CREDENTIALS environment variable when using it.
+see:https://cloud.google.com/text-to-speech/docs/before-you-begin
+"""
+
+
+class GoogleTextToSpeechNode(ComputeNode):
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super(GoogleTextToSpeechNode, cls).__new__(cls)
+            cls._instance.is_start = False
+        return cls._instance
+
+    def __init__(self):
+        super().__init__()
+        if self.is_start is True:
+            logger.warn("GoogleTextToSpeechNode is already start")
+            return
+
+        self.is_start = True
+        self.node_id = "google_text_to_speech_node"
+        self.task_queue = Queue()
+
+        self.client = texttospeech.TextToSpeechClient()
+
+        self.start()
+
+    def start(self):
+        async def _run_task_loop():
+            while True:
+                task = await self.task_queue.get()
+                try:
+                    result = self._run_task(task)
+                    if result is not None:
+                        task.state = ComputeTaskState.DONE
+                        task.result = result
+                except Exception as e:
+                    logger.error(f"google_text_to_speech_node run task error: {e}")
+                    task.state = ComputeTaskState.ERROR
+                    task.result = ComputeTaskResult()
+                    task.result.set_from_task(task)
+                    task.result.worker_id = self.node_id
+                    task.result.result_str = str(e)
+
+        asyncio.create_task(_run_task_loop())
+
+    def _run_task(self, task: ComputeTask):
+        task.state = ComputeTaskState.RUNNING
+        language_code = task.params["language_code"]
+        text = task.params["text"]
+
+        synthesis_input = texttospeech.SynthesisInput(text=text)
+        voice = texttospeech.VoiceSelectionParams(language_code=language_code,
+                                                  ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL)
+
+        audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3)
+
+        response = self.client.synthesize_speech(input=synthesis_input, voice=voice, audio_config=audio_config)
+
+        result = ComputeTaskResult()
+        result.set_from_task(task)
+        result.worker_id = self.node_id
+        result.result = response.audio_content
+        return result
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
+        logger.info(f"google_text_to_speech_node push task: {task.display()}")
+        self.task_queue.put_nowait(task)
+
+    async def remove_task(self, task_id: str):
+        pass
+
+    def get_task_state(self, task_id: str):
+        pass
+
+    def display(self) -> str:
+        return f"GoogleTextToSpeechNode: {self.node_id}"
+
+    def get_capacity(self):
+        return 0
+
+    def is_support(self, task_type: ComputeTaskType) -> bool:
+        if task_type == ComputeTaskType.TEXT_2_VOICE:
+            return True
+        return False
+
+    def is_local(self) -> bool:
+        return False

src/aios_kernel/local_llama_compute_node.py

@@ -0,0 +1,91 @@
+
+import logging
+import requests
+from typing import Optional, List
+from pydantic import BaseModel
+
+from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskType
+from .queue_compute_node import Queue_ComputeNode
+
+logger = logging.getLogger(__name__)
+
+"""
+This is a custom implementation, it should be redesigned.
+"""
+
+class LocalLlama_ComputeNode(Queue_ComputeNode):
+    async def execute_task(self, task: ComputeTask) -> {
+        "content": str,
+        "message": str,
+        "state": ComputeTaskState,
+        "error": {
+            "code": int,
+            "message": str,
+        }
+    }:
+        class GenerateResponse(BaseModel):
+            error: Optional[int]
+            msg: Optional[str]
+            results: Optional[List[str]]
+
+        try:
+            prompt_msgs = []
+            for prompt in task.params["prompts"]:
+                prompt_msgs.append(prompt["content"])
+                
+            body = {
+                "prompts": prompt_msgs
+            }
+            
+            response = requests.post("http://aigc:7880/generate", json = body, verify=False, headers={"Content-Type": "application/json"})
+            response.close()
+
+            logger.info(f"LocalLlama_ComputeNode task responsed, request: {body}, status-code: {response.status_code}, headers: {response.headers}, content: {response.content}")
+
+            if response.status_code != 200:
+                return {
+                    "state": ComputeTaskState.ERROR,
+                    "error": {
+                        "code": response.status_code,
+                        "message": "http request failed: " + response.status_code
+                    }
+                }
+            else:
+                resp = response.json()
+                if "error" in resp:
+                    return {
+                        "state": ComputeTaskState.ERROR,
+                        "error": {
+                            "code": resp["error"],
+                            "message": "local llama failed:" + resp["msg"]
+                        }
+                    }
+                else:
+                    return {
+                        "state": ComputeTaskState.DONE,
+                        "content": str(resp["results"]),
+                        "message": str(resp["results"])
+                    }
+        except Exception as err:
+            import traceback
+            logger.error(f"{traceback.format_exc()}, error: {err}")
+            
+            return {
+                "state": ComputeTaskState.ERROR,
+                "error": {
+                    "code": -1,
+                    "message": "unknown exception: " + str(err)
+                }
+            }
+
+    def display(self) -> str:
+        return f"LocalLlama_ComputeNode: {self.node_id}"
+
+    def get_capacity(self):
+        pass
+
+    def is_support(self, task: ComputeTask) -> bool:
+        return task.task_type == ComputeTaskType.LLM_COMPLETION and (not task.params["model_name"] or task.params["model_name"] == "llama")
+
+    def is_local(self) -> bool:
+        return True

src/aios_kernel/open_ai_node.py

@@ -5,78 +5,60 @@ import asyncio
 from asyncio import Queue
 import logging
 
-from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState
+from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
 from .compute_node import ComputeNode
+from .storage import AIStorage,UserConfig
 
 logger = logging.getLogger(__name__)
 
+
 class OpenAI_ComputeNode(ComputeNode):
     _instance = None
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super(OpenAI_ComputeNode, cls).__new__(cls)
-            cls._instance.is_start = False
+            cls._instance = OpenAI_ComputeNode()
         return cls._instance
     
+    @classmethod
+    def declare_user_config(cls):
+        if os.getenv("OPENAI_API_KEY_") is None:
+            user_config = AIStorage.get_instance().get_user_config()
+            user_config.add_user_config("openai_api_key","openai api key",False,None)
+
     def __init__(self) -> None:
         super().__init__()
-        if self.is_start is True:
-            logger.warn("OpenAI_ComputeNode is already start")
-            return
-        
-        self.is_start = True
-        #openai.organization = "org-AoKrOtF2myemvfiFfnsSU8rF" #buckycloud
-        self.openai_api_key = ""
-        self.node_id = "openai_node"
 
+        self.is_start = False
+        # openai.organization = "org-AoKrOtF2myemvfiFfnsSU8rF" #buckycloud
+        self.openai_api_key = None
+        self.node_id = "openai_node"
         self.task_queue = Queue()
 
-        if os.getenv("OPENAI_API_KEY") is not None: 
-            openai.api_key = os.getenv("OPENAI_API_KEY")
+
+    async def initial(self):
+        if os.getenv("OPENAI_API_KEY") is not None:
+            self.openai_api_key = os.getenv("OPENAI_API_KEY")
         else:
-            openai.api_key = self.openai_api_key
+            self.openai_api_key = AIStorage.get_instance().get_user_config().get_user_config("openai_api_key")
+
+        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()
-    
-    async def push_task(self,task:ComputeTask,proiority:int = 0):
+        return True
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
         logger.info(f"openai_node push task: {task.display()}")
         self.task_queue.put_nowait(task)
-    
-    async def remove_task(self,task_id:str):
+
+    async def remove_task(self, task_id: str):
         pass
-    
-    def _run_task(self,task:ComputeTask):
+
+    def _run_task(self, task: ComputeTask):
         task.state = ComputeTaskState.RUNNING
-        # switch tsak type
-        if task.task_type == "llm_completion":
-            mode_name = task.params["model_name"]
-            # max_token_size = task.params["max_token_size"]
-            prompts = task.params["prompts"]
-
-            mode_name = task.params["model_name"]
-	        # max_token_size = task.params["max_token_size"]
-            prompts = task.params["prompts"]
-
-            logger.info(f"call openai {mode_name} prompts: {prompts}")
-            resp = openai.ChatCompletion.create(model=mode_name,
-                                                messages=prompts,
-                                                max_tokens=4000,
-                                                temperature=1.2)
-            logger.info(f"openai response: {resp}")
-
-            status_code = resp["choices"][0]["finish_reason"]
-            if status_code != "stop":
-                task.state = ComputeTaskState.ERROR
-                task.error_str =f"The status code was {status_code}."
-                return None
-
-            result = ComputeTaskResult()    
-            result.set_from_task(task)
-            result.worker_id = self.node_id
-            result.result_str = resp["choices"][0]["message"]["content"]
-            result.result = resp["choices"][0]["message"]
-
-            return result
         if task.task_type == "text_embedding":
             model_name = task.params["model_name"]
             input = task.params["input"]
@@ -108,34 +90,85 @@ class OpenAI_ComputeNode(ComputeNode):
             result.result = resp["data"][0]["embedding"]
 
             return result
-            
+          
+        if task.task_type == "llm_completion":
+            mode_name = task.params["model_name"]
+            # max_token_size = task.params["max_token_size"]
+            prompts = task.params["prompts"]
+
+            mode_name = task.params["model_name"]
+	        # max_token_size = task.params["max_token_size"]
+            prompts = task.params["prompts"]          
+       
+
+            logger.info(f"call openai {mode_name} prompts: {prompts}")
+
+            if task.params.get("inner_functions") is None:
+                resp = openai.ChatCompletion.create(model=mode_name,
+                                                messages=prompts,
+                                                max_tokens=task.params["max_token_size"],
+                                                temperature=0.7)
+            else:
+                resp = openai.ChatCompletion.create(model=mode_name,
+                                                    messages=prompts,
+                                                    functions=task.params["inner_functions"],
+                                                    max_tokens=task.params["max_token_size"],
+                                                    temperature=0.7) # TODO: add temperature to task params?
+
+        
+            logger.info(f"openai response: {resp}")
+
+            result = ComputeTaskResult()
+            result.set_from_task(task)
+
+            status_code = resp["choices"][0]["finish_reason"]
+            match status_code:
+                case "function_call":
+                    task.state = ComputeTaskState.DONE
+                case "stop":
+                    task.state = ComputeTaskState.DONE
+                case _:
+                    task.state = ComputeTaskState.ERROR
+                    task.error_str = f"The status code was {status_code}."
+                    return None
+
+            result.worker_id = self.node_id
+            result.result_str = resp["choices"][0]["message"]["content"]
+            result.result_message = resp["choices"][0]["message"]
+
+        return result
+
+
     def start(self):
+        if self.is_start is True:
+            return
+        self.is_start = True
+        
         async def _run_task_loop():
             while True:
-                logger.info("openai_node is waiting for task...")
                 task = await self.task_queue.get()
                 logger.info(f"openai_node get task: {task.display()}")
                 result = self._run_task(task)
                 if result is not None:
                     task.state = ComputeTaskState.DONE
                     task.result = result
-        
+
         asyncio.create_task(_run_task_loop())
 
     def display(self) -> str:
         return f"OpenAI_ComputeNode: {self.node_id}"
-    
-    def get_task_state(self,task_id:str):
-        pass
 
+    def get_task_state(self, task_id: str):
+        pass
 
     def get_capacity(self):
         pass
 
 
     def is_support(self, task: ComputeTask) -> bool:
-        if task.task_type == "llm_completion":
-            return True
+        if task.task_type == ComputeTaskType.LLM_COMPLETION: 
+            if (not task.params["model_name"] or task.params["model_name"] == "gpt-4-0613")
+                return True
         if task.task_type == "text_embedding":
             if task.params["model_name"] == "text-embedding-ada-002":
                 return True
@@ -144,9 +177,3 @@ class OpenAI_ComputeNode(ComputeNode):
 
     def is_local(self) -> bool:
         return False
-
-
-
-    
-    
-    

src/aios_kernel/queue_compute_node.py

@@ -0,0 +1,69 @@
+
+import asyncio
+from asyncio import Queue
+import logging
+from abc import abstractmethod
+
+from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
+from .compute_node import ComputeNode
+
+logger = logging.getLogger(__name__)
+
+class Queue_ComputeNode(ComputeNode):
+    def __init__(self):
+        super().__init__()
+        self.task_queue = Queue()
+
+    @abstractmethod
+    async def execute_task(self, task: ComputeTask) -> {
+        "content": str,
+        "message": str,
+        "state": ComputeTaskState,
+        "error": {
+            "code": int,
+            "message": str,
+        }
+    }:
+        pass
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
+        logger.info(f"{self.display()} push task: {task.display()}")
+        self.task_queue.put_nowait(task)
+
+    async def remove_task(self, task_id: str):
+        pass
+
+    async def _run_task(self, task: ComputeTask):
+        task.state = ComputeTaskState.RUNNING
+        
+        resp = await self.execute_task(task)
+
+        result = ComputeTaskResult()
+        result.set_from_task(task)
+
+        task.state = resp["state"]
+
+        if task.state == ComputeTaskState.ERROR:
+            task.error_str = resp["error"]["message"]
+
+
+        result.worker_id = self.node_id
+        result.result_str = resp["content"]
+        result.result_message = resp["message"]
+
+        return result
+
+    def start(self):
+        async def _run_task_loop():
+            while True:
+                task = await self.task_queue.get()
+                logger.info(f"{self.display()} get task: {task.display()}")
+                result = await self._run_task(task)
+                if result is not None:
+                    task.result = result
+
+        asyncio.create_task(_run_task_loop())
+
+
+    def get_task_state(self, task_id: str):
+        pass

src/aios_kernel/role.py

@@ -6,6 +6,7 @@ class AIRole:
     def __init__(self) -> None:
         self.agent_instance_id : str = None
         self.role_name : str = None
+        self.role_id :str = None # $workflow_id.$sub_workflow_id.$role_name
         self.fullname : str = None
         self.agent_name : str = None
         self.prompt : AgentPrompt = None
@@ -19,6 +20,7 @@ class AIRole:
             return False
         self.role_name = name_node
 
+
         agent_id_node = config.get("agent")
         if agent_id_node is None:
             logging.error("agent id is not found!")
@@ -35,7 +37,10 @@ class AIRole:
         intro_node = config.get("intro")
         if intro_node is not None:
             self.introduce = intro_node
-            
+    
+    def get_role_id(self) -> str:
+        return self.role_id
+
     def get_intro(self) -> str:
         return self.introduce
 
@@ -48,6 +53,7 @@ class AIRole:
 class AIRoleGroup:
     def __init__(self) -> None:
         self.roles : dict[str,AIRole] = {}
+        self.owner_name : str = None
         
     def load_from_config(self,config:dict) -> bool:
         for k,v in config.items():
@@ -55,7 +61,7 @@ class AIRoleGroup:
             if role.load_from_config(v) is False:
                 logging.error(f"load role {k} failed!")
                 return False
-            
+            role.role_id = self.owner_name + "." + k
             self.roles[k] = role
         
         return True

src/aios_kernel/stability_node.py

@@ -0,0 +1,141 @@
+import os
+import io
+import asyncio
+from asyncio import Queue
+import logging
+
+from PIL import Image
+from stability_sdk import client
+import stability_sdk.interfaces.gooseai.generation.generation_pb2 as generation
+
+from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
+from .compute_node import ComputeNode
+from .storage import AIStorage,UserConfig
+
+logger = logging.getLogger(__name__)
+
+
+class Stability_ComputeNode(ComputeNode):
+    _instanace = None
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            cls._instance = Stability_ComputeNode()
+        return cls._instance
+    
+    @classmethod
+    def declare_user_config(cls):
+        user_config = AIStorage.get_instance().get_user_config()
+        user_config.add_user_config("stability_api_key",False,None,"STABILITY_API_KEY")
+
+    def __init__(self) -> None:
+        super().__init__()
+
+        self.is_start = False
+        self.node_id = "stability_node"
+        self.api_key = ""
+        self.engine = "stable-diffusion-512-v2-1"
+
+        self.task_queue = Queue()
+
+        if os.getenv("STABILITY_API_KEY") is not None:
+            self.api_key = os.getenv("STABILITY_API_KEY")
+
+        # Check out the following link for a list of available engines: https://platform.stability.ai/docs/features/api-parameters#engine
+        if os.getenv("STABILITY_ENGINE") is not None:
+            self.engine = os.getenv("STABILITY_ENGINE")
+
+        self.client = client.StabilityInference(
+            key=self.api_key,
+            verbose=True,  # Print debug messages.
+            engine=self.engine,
+        )
+
+        self.start()
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
+        logger.info(f"stability_node push task: {task.display()}")
+        self.task_queue.put_nowait(task)
+
+    async def remove_task(self, task_id: str):
+        pass
+
+    def _run_task(self, task: ComputeTask):
+        task.state = ComputeTaskState.RUNNING
+        # model_name && max_token_size not used here
+        prompts = task.params["prompts"]
+
+        logging.info(f"call stability {self.engine} prompts: {prompts}")
+        answers = self.client.generate(
+            prompt=prompts,
+            # If a seed is provided, the resulting generated image will be deterministic.
+            seed=0,
+            # What this means is that as long as all generation parameters remain the same, you can always recall the same image simply by generating it again.
+            # Note: This isn't quite the case for Clip Guided generations, which we'll tackle in a future example notebook.
+            # Amount of inference steps performed on image generation. Defaults to 30.
+            steps=30,
+            # Influences how strongly your generation is guided to match your prompt.
+            cfg_scale=7.0,
+            # Setting this value higher increases the strength in which it tries to match your prompt.
+            # Defaults to 7.0 if not specified.
+            width=512,  # Generation width, defaults to 512 if not included.
+            height=512,  # Generation height, defaults to 512 if not included.
+            # Number of images to generate, defaults to 1 if not included.
+            samples=1,
+            # Choose which sampler we want to denoise our generation with.
+            sampler=generation.SAMPLER_K_DPMPP_2M
+            # Defaults to k_dpmpp_2m if not specified. Clip Guidance only supports ancestral samplers.
+            # (Available Samplers: ddim, plms, k_euler, k_euler_ancestral, k_heun, k_dpm_2, k_dpm_2_ancestral, k_dpmpp_2s_ancestral, k_lms, k_dpmpp_2m, k_dpmpp_sde)
+        )
+
+        for resp in answers:
+            for artifact in resp.artifacts:
+                logger.info(f"artifact:{artifact.id},{artifact.type},{artifact.finish_reason}")
+                
+                if artifact.finish_reason == generation.FILTER:
+                    logging.warn("request activated the API's safety filters")
+                if artifact.type == generation.ARTIFACT_IMAGE:
+                    img = Image.open(io.BytesIO(artifact.binary))
+                    # Save our generated images with the task_id as the filename.
+                    file_name = task.task_id + ".png"  # which dir to save?
+                    img.save(file_name)
+
+                    result = ComputeTaskResult()
+                    result.set_from_task(task)
+                    result.worker_id = self.node_id
+                    result.result = {"file": file_name}
+
+                    return result
+
+        return None
+
+    def start(self):
+        if self.is_start:
+            return
+        self.is_start = True
+        async def _run_task_loop():
+            while True:
+                logger.info("stability_node is waiting for task...")
+                task = await self.task_queue.get()
+                logger.info(f"stability_node get task: {task.display()}")
+                result = self._run_task(task)
+                if result is not None:
+                    task.state = ComputeTaskState.DONE
+                    task.result = result
+
+        asyncio.create_task(_run_task_loop())
+
+    def display(self) -> str:
+        return f"Stability_ComputeNode: {self.node_id}"
+
+    def get_task_state(self, task_id: str):
+        pass
+
+    def get_capacity(self):
+        pass
+
+    def is_support(self, task: ComputeTask) -> bool:
+        return task.task_type == ComputeTaskType.TEXT_2_IMAGE
+
+    def is_local(self) -> bool:
+        return False

src/aios_kernel/storage.py

@@ -0,0 +1,171 @@
+from typing import Any
+from pathlib import Path
+import os
+import logging
+import toml
+import aiofiles
+
+logger = logging.getLogger(__name__)
+
+_file_dir = os.path.dirname(__file__)
+
+class ResourceLocation:
+    def __init__(self) -> None:
+        pass
+
+class UserConfigItem:
+    def __init__(self) -> None:
+        self.default_value = None 
+        self.is_optional = False
+        self.item_type = "str"
+        self.desc = None
+        self.value = None
+        self.user_set = False
+        
+
+class UserConfig:
+    def __init__(self) -> None:
+        self.config_table = {}
+        self.user_config_path:str = None
+
+
+    def add_user_config(self,key:str,desc:str,is_optional:bool,default_value:Any=None,item_type="str") -> None:
+        if self.config_table.get(key) is not None:
+            logger.warning("user config key %s already exist, will be overrided",key)
+    
+        new_config_item = UserConfigItem()
+        new_config_item.default_value = default_value
+        new_config_item.is_optional = is_optional
+        new_config_item.desc = desc
+        new_config_item.item_type = item_type
+        self.config_table[key] = new_config_item
+        
+    async def load_value_from_file(self,file_path:str,is_user_config = False) -> None:
+        try: 
+            all_config = toml.load(file_path)
+            if all_config is not None:
+                for key,value in all_config.items():
+                    config_item = self.config_table.get(key)
+                    if config_item is None:
+                        logger.warning("user config key %s not exist",key)
+                        continue
+                    config_item.value = value
+                    config_item.user_set = is_user_config
+
+        except Exception as e:
+            logger.warn(f"load user config from {file_path} failed!")
+
+    async def save_value_to_user_config(self) -> None:
+        will_save_config = {}
+        for key,value in self.config_table.items():
+            if value.user_set:
+                will_save_config[key] = value.value
+        
+        if len(will_save_config) > 0:
+            try:
+                directory = os.path.dirname(self.user_config_path)
+                if not os.path.exists(directory):
+                    os.makedirs(directory)
+
+                async with aiofiles.open(self.user_config_path,"w") as f:
+                    toml_str = toml.dumps(will_save_config)
+                    await f.write(toml_str)
+            except Exception as e:
+                logger.error(f"save user config to {self.user_config_path} failed!")
+                return False
+        
+        return True
+
+    def get_user_config(self,key:str) -> Any:
+        config_item = self.config_table.get(key)
+        if config_item is None:
+            raise Exception("user config key %s not exist",key)
+        
+        if config_item.value is None:
+            return config_item.default_value
+        
+        return config_item.value
+
+    def set_user_config(self,key:str,value:Any) -> None:
+        config_item = self.config_table.get(key)
+        if config_item is None:
+            logger.warning("user config key %s not exist",key)
+            return 
+        
+        config_item.value = value
+        config_item.user_set = True
+        #TODO: save to file?
+
+        
+    def check_user_config(self) -> None:
+        check_result = {}
+        for key,config_item in self.config_table.items():
+            if config_item.value is None and not config_item.is_optional:
+                check_result[key] = config_item
+        
+        if len(check_result) > 0:
+            return check_result
+        else:
+            return None
+
+# storage sytem for current user
+class AIStorage:
+    _instance = None
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            cls._instance = AIStorage()
+        return cls._instance
+
+    def __init__(self) -> None:
+        self.is_dev_mode = False
+        self.user_config = UserConfig()
+
+    async def initial(self)->bool:
+        self.user_config.user_config_path = str(self.get_myai_dir() / "etc/system.cfg.toml")
+        await self.user_config.load_value_from_file(self.get_system_dir() + "/system.cfg.toml")
+        await self.user_config.load_value_from_file(self.user_config.user_config_path,True)
+
+    def get_user_config(self) -> UserConfig:
+        return self.user_config
+
+    def get_system_dir(self) -> str:
+        """ 
+        system dir is dir for aios system
+        /opt/aios
+        """
+        if self.is_dev_mode:
+            return os.path.abspath(_file_dir + "/../")
+        else:
+            return "/opt/aios/"
+        
+    
+    def get_system_app_dir(self)->str:
+        """
+        system app dir is the dir for aios build-in app 
+        /opt/aios/app
+        """
+        if self.is_dev_mode:
+            return os.path.abspath(_file_dir + "/../../rootfs/")
+        else:
+            return "/opt/aios/app/"
+    
+    def get_myai_dir(self) -> str:
+        """
+        my ai dir is the dir for user to store their ai app and data
+        ~/myai/
+        """
+        return Path.home() / "myai"
+        
+    def get_db(self,app_name:str)->ResourceLocation:
+        pass
+
+    def open_file(self,file_path:str,options:dict):
+        pass
+
+    def get_named_object(self,name:str) -> Any:
+        pass
+
+    def put_named_object(self,name:str,obj:Any) -> None:
+        pass
+

src/aios_kernel/tg_tunnel.py

@@ -0,0 +1,116 @@
+import logging
+import threading
+import asyncio
+import uuid
+
+from typing import Callable
+
+from telegram import ForceReply, Update
+from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
+
+from .tunnel import AgentTunnel
+from .contact_manager import ContactManager
+from .agent_message import AgentMsg
+
+logger = logging.getLogger(__name__)
+
+class TelegramTunnel(AgentTunnel):
+
+    @classmethod
+    def register_to_loader(cls):
+        async def load_tg_tunnel(config:dict) -> AgentTunnel:
+            result_tunnel = TelegramTunnel("")
+            if await result_tunnel.load_from_config(config):
+                return result_tunnel
+            else:
+                return None
+            
+        AgentTunnel.register_loader("TelegramTunnel",load_tg_tunnel)
+
+        
+    async def load_from_config(self,config:dict)->bool:
+        self.tg_token = config["token"]
+        self.target_id = config["target"]
+        self.tunnel_id = config["tunnel_id"]
+        self.type = "TelegramTunnel"
+        return True
+
+    def dump_to_config(self) -> dict:
+        pass
+
+    def __init__(self,tg_token:str) -> None:
+        super().__init__()
+        self.is_start = False
+        self.tg_token = tg_token
+        #self.tunnel_id = "tg_tunnel#" + self.app.bot.id
+
+    async def start(self) -> bool:
+        if self.is_start:
+            logger.warning(f"tunnel {self.tunnel_id} is already started")
+            return False
+        self.is_start = True   
+
+        self.app:Application = Application.builder().token(self.tg_token).build()
+        self.app.add_handler(MessageHandler(filters.TEXT, self.on_message))
+
+        def _run_app():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            self.app.run_polling(allowed_updates=Update.ALL_TYPES)
+
+        self.poll_thread = threading.Thread(target=_run_app)
+        self.poll_thread.start()
+        return True
+
+    async def close(self) -> None:
+        pass
+
+    async def _process_message(self, msg: AgentMsg) -> None:
+        logger.warn(f"process message {msg.msg_id} from {msg.sender} to {msg.target}")
+
+    async def conver_tg_msg_to_agent_msg(self,update:Update) -> AgentMsg:
+        agent_msg = AgentMsg()
+        agent_msg.topic = "_telegram"
+        agent_msg.msg_id = "tg_msg#" + str(update.message.message_id) + "#" + uuid.uuid4().hex
+        agent_msg.target = self.target_id
+        agent_msg.body = update.message.text
+        agent_msg.create_time = update.message.date.timestamp()
+        #if update.message.photo is not None:
+        #    agent_msg.body_mime = "image"
+        #    agent_msg.body = update.message.photo[-1].get_file().download()
+        return agent_msg
+
+
+
+    async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        cm = ContactManager.get_instance()
+        reomte_user_name = f"{update.effective_user.id}@telegram"
+        #contact = cm.get_by_name(update.effective_user.username)
+        #if contact is not None:
+        #    reomte_user_name = contact.get_name()
+        #if contact is None:
+        #    update.message.reply_text(f"{self.target_id} process message error, unknown user!")
+        #if not contact.is_zone_owner():
+        #    update.message.reply_text(f"{self.target_id} process message error, you are not my owner!")
+
+        agent_msg = await self.conver_tg_msg_to_agent_msg(update)
+        agent_msg.sender = reomte_user_name
+        self.ai_bus.register_message_handler(reomte_user_name, self._process_message)
+        resp_msg = await self.ai_bus.send_message(agent_msg)
+        if resp_msg is None:
+            await update.message.reply_text(f"{self.target_id} process message error")
+        else:
+            if resp_msg.body_mime is None:
+                await update.message.reply_text(resp_msg.body)
+            else:
+                if resp_msg.body_mime.startswith("image"):
+                    photo_file = open(resp_msg.body,"rb")
+                    if photo_file:
+                        await update.message.reply_photo(resp_msg.body)
+                    else:
+                        await update.message.reply_text(resp_msg.body)
+                else:
+                    await update.message.reply_text(resp_msg.body)
+            
+        
+        

src/aios_kernel/tunnel.py

@@ -0,0 +1,63 @@
+from abc import ABC, abstractmethod
+import logging
+from typing import Coroutine
+from .agent_message import AgentMsg
+from .bus import AIBus
+
+logger = logging.getLogger(__name__)
+
+class AgentTunnel(ABC):
+    _all_loader = {}
+    _all_tunnels = {}
+    @classmethod
+    def register_loader(cls,tunnel_type:str,loader:Coroutine) -> None:
+        cls._all_loader[tunnel_type] = loader
+
+    @classmethod
+    async def load_all_tunnels_from_config(cls,config:dict) -> None:
+        for tunnel_config in config:
+            loader = cls._all_loader.get(tunnel_config["type"])
+            if loader is not None:
+                tunnel = await loader(tunnel_config)
+                if tunnel is not None:
+                    cls._all_tunnels[tunnel.tunnel_id] = tunnel
+                    tunnel.connect_to(AIBus.get_default_bus(),tunnel.target_id)
+                    await tunnel.start()
+                else:
+                    logger.error(f"load tunnel {tunnel_config['tunnel_id']} failed") 
+            else:
+                logger.error(f"load tunnel {tunnel_config['type']} failed,loader not found")
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.tunnel_id = None
+        self.target_id = None
+        self.target_type = None
+        self.ai_bus = None
+        self.is_connected = False
+
+    def connect_to(self, ai_bus:AIBus,target_id: str) -> None:
+        """
+        Connect to the agent with the given id
+        """
+        if self.is_connected:
+            logger.warning(f"tunnel {self.tunnel_id} is already connected to {self.target_id}")
+            return
+        self.target_id = target_id
+        self.target_type = "agent"
+        self.ai_bus = ai_bus
+        self.is_connected = True
+
+        
+
+    @abstractmethod
+    async def start(self) -> bool:
+        pass
+
+    @abstractmethod
+    async def close(self) -> None:
+        pass
+
+    @abstractmethod
+    async def _process_message(self, msg: AgentMsg) -> None:
+        pass

src/aios_kernel/whisper_node.py

@@ -0,0 +1,111 @@
+from asyncio import Queue
+import asyncio
+import openai
+import os
+import logging
+
+from .compute_node import ComputeNode
+from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType
+
+logger = logging.getLogger(__name__)
+
+
+class WhisperComputeNode(ComputeNode):
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance.is_start = False
+        return cls._instance
+
+    def __init__(self) -> None:
+        super().__init__()
+        if self.is_start is True:
+            logger.warn("WhisperComputeNode is already start")
+            return
+
+        self.is_start = True
+        self.node_id = "whisper_node"
+        self.enable = True
+        self.task_queue = Queue()
+        self.open_api_key = None
+
+        if self.open_api_key is None and os.getenv("OPENAI_API_KEY") is not None:
+            self.open_api_key = os.getenv("OPENAI_API_KEY")
+
+        if self.open_api_key is None:
+            raise Exception("WhisperComputeNode open_api_key is None")
+
+        self.start()
+
+    def start(self):
+        async def _run_task_loop():
+            while True:
+                task = await self.task_queue.get()
+                try:
+                    result = self._run_task(task)
+                    if result is not None:
+                        task.state = ComputeTaskState.DONE
+                        task.result = result
+                except Exception as e:
+                    logger.error(f"whisper_node run task error: {e}")
+                    task.state = ComputeTaskState.ERROR
+                    task.result = ComputeTaskResult()
+                    task.result.set_from_task(task)
+                    task.result.worker_id = self.node_id
+                    task.result.result_str = str(e)
+
+        asyncio.create_task(_run_task_loop())
+
+    def _run_task(self, task: ComputeTask):
+        task.state = ComputeTaskState.RUNNING
+        prompt = task.params["prompt"]
+        response_format = None
+        if "response_format" in task.params:
+            response_format = task.params["response_format"]
+        temperature = None
+        if "temperature" in task.params:
+            temperature = task.params["temperature"]
+        language = None
+        if "language" in task.params:
+            language = task.params["language"]
+        file = task.params["file"]
+
+        resp = openai.Audio.transcribe("whisper-1",
+                                       file,
+                                       self.open_api_key,
+                                       prompt=prompt,
+                                       response_format=response_format,
+                                       temperature=temperature,
+                                       language=language)
+        result = ComputeTaskResult()
+        result.set_from_task(task)
+        result.worker_id = self.node_id
+        result.result_str = resp["text"]
+        result.result = resp
+        return result
+
+    async def push_task(self, task: ComputeTask, proiority: int = 0):
+        logger.info(f"whisper_node push task: {task.display()}")
+        self.task_queue.put_nowait(task)
+
+    async def remove_task(self, task_id: str):
+        pass
+
+    def get_task_state(self, task_id: str):
+        pass
+
+    def display(self) -> str:
+        return f"WhisperComputeNode: {self.node_id}"
+
+    def get_capacity(self):
+        return 0
+
+    def is_support(self, task_type: ComputeTaskType) -> bool:
+        if task_type == ComputeTaskType.VOICE_2_TEXT:
+            return True
+        return False
+
+    def is_local(self) -> bool:
+        return False

src/aios_kernel/workflow.py

@@ -1,18 +1,23 @@
-
 import logging
 import asyncio
+import json
+import os
+import time 
 from asyncio import Queue
 from typing import Optional,Tuple
 from abc import ABC, abstractmethod
 
 from .environment import Environment,EnvironmentEvent
-from .agent_message import AgentMsg,AgentMsgState
+from .agent_message import AgentMsg,AgentMsgStatus
 from .agent import AgentPrompt,AgentMsg
 from .chatsession import AIChatSession
 from .role import AIRole,AIRoleGroup
-from .ai_function import CallChain
+from .ai_function import AIFunction
 from .compute_kernel import ComputeKernel
+from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState
 from .bus import AIBus
+from .workflow_env import WorkflowEnvironment
+
 
 logger = logging.getLogger(__name__)
 
@@ -33,9 +38,20 @@ class MessageFilter:
         return True
 
 
+class LLMResult:
+    def __init__(self) -> None:
+        self.state : str = "ignore"
+        self.resp : str = ""
+        self.post_msgs = []
+        self.send_msgs = []
+        self.calls = []
+        self.post_calls = []
+
+
 class Workflow:
     def __init__(self) -> None:
         self.workflow_name : str = None
+        self.workflow_id : str = None
         self.rule_prompt : AgentPrompt = None
         self.workflow_config = None
         self.role_group : dict = None
@@ -44,6 +60,8 @@ class Workflow:
         self.sub_workflows = {}
         self.owner_workflow = None
         self.db_file = None
+        self.env_db_file = None
+        self.workflow_env:WorkflowEnvironment = None
 
         self.is_start = False
         self.msg_queue = Queue()
@@ -62,27 +80,57 @@ class Workflow:
             logger.error("workflow config must have name")
             return False
         self.workflow_name = config.get("name")
+        if self.owner_workflow is None:
+            self.workflow_id = self.workflow_name
+        else:
+            self.workflow_id = self.owner_workflow.workflow_id + "." + self.workflow_name
+            self.db_file = self.owner_workflow.db_file
 
-        #if config.get("rule_prompt") is None:
-        #    logger.error("workflow config must have rule_prompt")
-        #    return False
-        #self.rule_prompt = AgentPrompt()
-        #if self.rule_prompt.load_from_config(config.get("rule_prompt")) is False:
-        #    logger.error("Workflow load rule_prompt failed")
-        #    return False
+        if config.get("prompt") is not None:
+            self.rule_prompt = AgentPrompt()
+            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
         self.role_group = AIRoleGroup()
+        self.role_group.owner_name = self.workflow_id
         if self.role_group.load_from_config(config.get("roles")) is False:
             logger.error("Workflow load role_group failed")
             return False
 
-        if config.get("input_filter") is not None:
+        if config.get("filter") is not None:
             self.input_filter = MessageFilter()
-            if self.input_filter.load_from_config(config.get("input_filter")) is False:
+            if self.input_filter.load_from_config(config.get("filter")) is False:
                 logger.error("Workflow load input_filter failed")
                 return False
+
+        if self.owner_workflow is None:
+            self.env_db_file = os.path.dirname(self.db_file) + "/" + self.workflow_id + "_env.db"
+        else:
+            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") 
+        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") 
+        if  connected_env_ndoe is not None:
+           for _node in connected_env_ndoe:
+                env_id = _node.get("env_id")
+                if env_id is None:
+                    continue
+
+                remote_env = Environment.get_env_by_id(_node.get(env_id))
+                if remote_env is None:
+                     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:
@@ -90,23 +138,95 @@ class Workflow:
                 logger.error("Workflow load sub workflows failed")
                 return False
             
-        #TODO: load env
-        
         return True
 
+    def _load_env_from_config(self,config:dict) -> bool:
+        for k,v in config.items():
+            self.workflow_env.set_value(k,v,False)
+
     def _load_sub_workflows(self,config:dict) -> bool:
         for k,v in config.items():
             sub_workflow = Workflow()
             sub_workflow.set_owner(self)
+
             if sub_workflow.load_from_config(v) is False:
                 logger.error(f"load sub workflow {k} failed!")
                 return False
             self.sub_workflows[k] = sub_workflow
         return True
+    
+    def _parse_msg_target(self,s:str)->list[str]:
+        return s.split(".")
+
+    async def _forword_msg(self,inner_obj_id,msg):
+        i : int = 1
+        current_workflow = self
+        while i < len(inner_obj_id):
+            if i == len(inner_obj_id) - 1:
+                the_role : AIRole = current_workflow.role_group.get(inner_obj_id[i])
+                current_workflow_chatsession = AIChatSession.get_session(current_workflow.workflow_id,msg.sender + "#" + msg.topic,current_workflow.db_file)
+                if the_role is not None:
+                    return await current_workflow.role_process_msg(msg,the_role,current_workflow_chatsession)
+                sub_workflow = current_workflow.sub_workflows.get(inner_obj_id[i])
+                if sub_workflow is not None:
+                    return await sub_workflow._process_msg(msg)
+                logger.error(f"{msg.target} not found! forword message failed!")
+                return None
+            else:
+                current_workflow = current_workflow.sub_workflows.get(inner_obj_id[i])
+                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:
+            return target
+        else:
+            result_str = ""
+            p = 0
+            for s in target_list:
+                p = p + 1
+                result_str += s
+                if p < len(target_list)-1:
+                    result_str += "."
+                else:
+                    return result_str
 
     async def _process_msg(self,msg:AgentMsg):
+        real_target = msg.target.split(".")[0]
+        targets = self._parse_msg_target(msg.target)
+        if len(targets) > 1:
+            return await self._forword_msg(targets,msg)
+        #0 we don't support workflow join a group right now, this cloud be a feture in future
+        if msg.mentions is not None:
+            logger.warn(f"workflow {self.workflow_id} recv a group chat message,not support ignore!")
+            return None
+
+        #1. workflow start process message
         final_result = None
-        chatsession = 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:
+                chatsesssion.append(msg)
+                logger.info(f"workflow {self.workflow_id} recv a group chat message from {msg.sender},but is not mentioned,ignore!")
+                return None
+
+            for mention in msg.mentions:
+                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: 
@@ -114,189 +234,302 @@ class Workflow:
                 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
-
-                result = await self._role_process_msg(msg,select_role)
-                if result is None:
-                    logger.error(f"_process_msg return None for :{msg}")
-                    return
-                if chatsession is not None:
-                    chatsession.append_post(result)
-                final_result = result
-            else:
-                logger.error(f"input_filter return None for :{msg}")
-                return
                 
-        else:
-            results = {}
-            for this_role in self.role_group.roles.values():
-                # TODO : we would do this in parallel
-                a_result = await self._role_process_msg(msg,this_role)
-                results[this_role.get_name()] = a_result
-
-            # merge result from all roles 
-            # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg
-            final_result:AgentMsg = self._merge_msg_result(results)
-            if chatsession is not None:
-                chatsession.append_post(final_result)
-        
+                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
 
+    @classmethod
+    def prase_llm_result(cls,llm_result_str:str)->LLMResult:
+        r = LLMResult()
+        if llm_result_str is None:
+            r.state = "ignore"
+            return r
+        if llm_result_str == "ignore":
+            r.state = "ignore"
+            return r
+        
+        lines = llm_result_str.splitlines()
+        is_need_wait = False
+        for line in lines:
+            func_call = AgentMsg.parse_function_call(line)
+            if func_call:
+                func_args = func_call[1]
+                match func_call[0]:
+                    case "sendmsg":# sendmsg($target_id,$msg_content)
+                        if len(func_args) != 2:
+                            logger.error(f"parse sendmsg failed! {func_call}")
+                            continue
+                        new_msg = AgentMsg()
+                        target_id = func_args[0]
+                        msg_content = func_args[1]
+                        new_msg.set("_",target_id,msg_content)
+    
+                        r.send_msgs.append(new_msg)
+                        is_need_wait = True
+                        continue
+                    case "postmsg":# postmsg($target_id,$msg_content)
+                        if len(func_args) != 2:
+                            logger.error(f"parse postmsg failed! {func_call}")
+                            continue
+                        new_msg = AgentMsg()
+                        target_id = func_args[0]
+                        msg_content = func_args[1]
+                        new_msg.set("_",target_id,msg_content)
+                        r.post_msgs.append(new_msg)
+                        continue
+                    case "call":# call($func_name,$args_str)
+                        r.calls.append(func_call)
+                        is_need_wait = True
+                        continue 
+                    case "post_call": # post_call($func_name,$args_str)
+                        r.post_calls.append(func_call)
+                        continue
+                
+                r.resp += line + "\n"
+            else:
+                r.resp += line + "\n"
+        
+        if is_need_wait:
+            r.state = "waiting"
+        else:
+            r.state = "reponsed"
+
+        return r
+
+    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
+            logger.info(f"{msg.sender} post message {msg.msg_id} to sub workflow: {msg.target}")
+            asyncio.create_task(target_workflow._process_msg(msg))
+
+        logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}")
+        await self.get_bus().post_message(msg.target,msg)
+        return
 
 
+    async def role_send_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} send message {msg.msg_id} to inner role: {msg.target}")
+            return await self.role_process_msg(msg,target_role,workflow_chat_session)
 
-    async def _role_process_msg(self,msg:AgentMsg,the_role:AIRole) -> None:
-        # TODO : we just record role's chatsession, but in future, we would record workflow's chatsession(like a groupo chat)
-        session_topic = f"{the_role.get_name()}#{msg.sender}#{msg.topic}"
-        chatsession = AIChatSession.get_session(self.workflow_name,session_topic,self.db_file)
-        if chatsession is None:
-            logger.error(f"get session {session_topic}@{self.workflow_name} failed!")
+        target_workflow = self.sub_workflows.get(msg.target)
+        if target_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)
+
+    async def role_call(self,call:tuple,the_role:AIRole):
+        logger.info(f"{the_role.role_id} call {call[0]} with args {call[1]}")
+        func_name = call[0]
+        arguments = call[1]
+
+        func_node : AIFunction = self.workflow_env.get_ai_function(func_name)
+        if func_node is None:
+            return "execute failed,function not found"
+    
+        result_str:str = await func_node.execute(**arguments)        
+        return result_str
+
+    async def role_post_call(self,call:tuple,the_role:AIRole):
+        logger.info(f"{the_role.role_id} post call {call[0]} with args {call[1]}")
+        return await self.role_call(call,the_role)
+
+    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)
+
+    def _get_inner_functions(self) -> dict:
+        all_inner_function = self.workflow_env.get_all_ai_functions()
+        if all_inner_function is None:
             return None
         
-        # prompt generat progress is most important part of workflow(app) develope
+        result_func = []
+        for inner_func in all_inner_function:
+            this_func = {}
+            this_func["name"] = inner_func.get_name()
+            this_func["description"] = inner_func.get_description()
+            this_func["parameters"] = inner_func.get_parameters()
+            result_func.append(this_func)
+        if len(result_func) > 0:
+            return result_func
+        return None 
+    
+    async def _role_execute_func(self,the_role:AIRole,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg) -> str:
+        from .compute_kernel import ComputeKernel
+
+        func_name = inenr_func_call_node.get("name")
+        arguments = json.loads(inenr_func_call_node.get("arguments"))
+
+        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()
+        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)
+
+        inner_func_call_node = task_result.result_message.get("function_call")
+        if inner_func_call_node:
+            return await self._role_execute_func(the_role,inner_func_call_node,prompt,org_msg)      
+        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):        
+        msg.target = the_role.get_role_id()
+  
+
         prompt = AgentPrompt()
         prompt.append(the_role.agent.prompt)
+        prompt.append(self.get_workflow_rule_prompt())
         prompt.append(the_role.get_prompt())
-
-        # prompt.append(self.get_workflow_rule_prompt())
         # prompt.append(self._get_function_prompt(the_role.get_name()))
         # prompt.append(self._get_knowlege_prompt(the_role.get_name()))
         
-        prompt.append(await self._get_prompt_from_session(chatsession))
-        #prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context
+        #support group chat, user content include sender name!
+        prompt.append(await self._get_prompt_from_session(workflow_chat_session))
 
         msg_prompt = AgentPrompt()
-        msg_prompt.messages = [{"role":"user","content":msg.body}]
+        msg_prompt.messages = [{"role":"user","content":f"{msg.sender}:{msg.body}"}]
         prompt.append(msg_prompt)
-        
-        result = await ComputeKernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
-        chatsession.append_recv(msg)
-        final_result = result        
-      
-        result_type : str = self._get_llm_result_type(result)
-        is_ignore = False
-        match result_type:
-            case "function":
-                callchain:CallChain = self._parse_function_call_chain(result)
-                resp = await callchain.exec()
-                if callchain.have_result():
-                    # generator proc resp prompt with WAITING state
-                    #proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession)
-                    final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
-                    return final_result
 
+        self._format_msg_by_env_value(prompt)
+        inner_functions = self._get_inner_functions()
+        
+        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}")
             
-            case "send_message":
-                # send message to other / sub workflow
-                next_msg:AgentMsg = self._parse_to_msg(result)
-                if next_msg is not None:
-                    next_msg.sender = self.workflow_name
-                    logger.info(f"W#{self.workflow_name} send message to {next_msg.get_target()}")
-                    resp_msg = await self.get_bus().send_message(next_msg.get_target(),next_msg)
-                    if resp_msg is not None:
-                        msg_prompt = AgentPrompt()
-                        msg_prompt.messages = [{"role":"assistant","content":result},{"role":"user","content":f"{next_msg.get_target()}:{resp_msg.body}"}]
-                                               
-                        final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
-                
-                
-            case "post_message":
-                # post message to other / sub workflow
-                next_msg:AgentMsg = self._parse_to_msg(result)
-                if next_msg is not None:
-                    next_msg.sender = self.workflow_name
-                    logger.info(f"W#{self.workflow_name} post message to {next_msg.get_target()}")
-                    self.get_bus().post_message(next_msg.get_target(),next_msg)
+            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)
 
-            case "ignore":
-                is_ignore = True
+            result = Workflow.prase_llm_result(result_str)
+            for postmsg in result.post_msgs:
+                postmsg.prev_msg_id = msg.get_msg_id()
+                # might be craete a new msg.topic for this postmsg
+                postmsg.topic = msg.topic
                 
-        if is_ignore:
-            return None
+                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)
+                    role_sesion.append(postmsg)
+                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":
+                    return None
+                case "reponsed":
+                    resp_msg = msg.create_resp_msg(result.resp)
+                    resp_msg.sender = the_role.get_role_id()
+                    # It is always the person handling the messages who puts them into the session.
+                    workflow_chat_session.append(msg)
+                    workflow_chat_session.append(resp_msg)
+                    #await self.get_bus().resp_message(resp_msg)
+                    return resp_msg
+                case "waiting":
+                    # TODO: Use role:"function" would be better
+                    for sendmsg in result.send_msgs:
+                        target = sendmsg.target
+                        sendmsg.topic = msg.topic
+                        sendmsg.prev_msg_id = msg.get_msg_id()
+                        send_resp = await self.role_send_msg(sendmsg,the_role,workflow_chat_session)
+                        if send_resp is not None:
+                            result_prompt_str += f"\n{target} response is :{send_resp.body}"
+
+                        if not self._is_in_same_workflow(sendmsg):
+                            role_sesion = AIChatSession.get_session(the_role.get_role_id(),f"{sendmsg.target}#{sendmsg.topic}",self.db_file)
+                            role_sesion.append(sendmsg)
+                            role_sesion.append(send_resp)
+                        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
+                            action_msg.result_str = call_result
+                            workflow_chat_session.append(action_msg)
+
+                    result_prompt = AgentPrompt()
+                    result_prompt.messages = [{"role":"user","content":result_prompt_str}]
+                    prompt.append(result_prompt)
+                    r = await _do_process_msg()
+                    return r
         
-        resp_msg = AgentMsg()
-        resp_msg.set(self.workflow_name,msg.sender,final_result)
-        chatsession.append_post(resp_msg)
-        return resp_msg
-
-    async def _pop_msg(self) -> AgentMsg:
-        pass
-
-    def _get_chat_session_for_msg(self,msg:AgentMsg) -> AIChatSession:
-        pass
+        return await _do_process_msg()
 
     async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt:
         messages = chatsession.read_history() # read last 10 message
         result_prompt = AgentPrompt()
         for msg in reversed(messages):
-            if msg.target == chatsession.owner_id:
-                result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"})
             if msg.sender == chatsession.owner_id:
                 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_msg_queue(self,session_id:str):
-        pass
-
-    def _merge_msg_result(self,results:dict) -> AgentMsg:
-        # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg
-        for k,v in results.items():
-            if v is not None:
-                return v
-
-    def _get_function_prompt(self,role_name:str) -> AgentPrompt:
-        pass
-    
     def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt:
         pass
 
-    def _get_resp_prompt(self,resp:str,msg:AgentMsg,role:AIRole,prompt:AgentPrompt) -> AgentPrompt:
-        pass
-
     def get_workflow_rule_prompt(self) -> AgentPrompt:
         return self.rule_prompt
-    
-    def _get_llm_result_type(self,llm_resp_str:str) -> str:
-        if llm_resp_str == "ignore":
-            return "ignore"
-        
-        if llm_resp_str.find("sendmsg(") != -1:
-            return "send_message"
-        
-        if llm_resp_str.find("postmsg(") != -1:
-            return "post_message"
-        
-        if llm_resp_str.find("call(") != -1:
-            return "function"
-        
-        return "text"
-
-    def _parse_function_call_chain(self,llm_resp_str) -> CallChain:
-        pass
-
-    def _parse_to_msg(self,llm_resp_str) -> AgentMsg:
-        lines = llm_resp_str.splitlines()
-        for line in lines:
-            if line.startswith("sendmsg("):
-                line = line[8:]
-                _index = line.find(",")
-                msg = AgentMsg()
-                msg.set("",line[:_index],line[_index+1:])
-                return msg
-            
-            if line.startswith("postmsg("):
-                line = line[8:]
-                _index = line.find(",")
-                msg = AgentMsg()
-                msg.set("",line[:_index],line[_index+1:])
-                return msg
-        
-        return None
-    
-    def get_workflow(self,workflow_name:str):
-        """get workflow from known workflow list or sub workflow list"""
-        pass
-
 
     def _env_event_to_msg(self,env_event:EnvironmentEvent) -> AgentMsg:
         pass
@@ -304,16 +537,20 @@ class Workflow:
     def get_inner_environment(self,env_id:str) -> Environment:
         pass
 
-    def connect_to_environment(self,env:Environment) -> None:
-        the_env = self.connected_environment.get(env.get_id())
-        if the_env is None:
-            self.connected_environment[env.get_id()] = env
-            def _env_msg_handler(env_event:EnvironmentEvent) -> None:
-                the_msg:AgentMsg= self._env_event_to_msg(env_event)
-                self.post_msg(the_msg)
-            
-            # register all event handler
-            the_env.attach_event_handler(None,_env_msg_handler)
-        else:
-            logger.warn(f"environment {env.get_id()} already connected!")
+    def connect_to_environment(self,the_env:Environment,conn_info:dict) -> None:
+        if the_env is not None:
+            self.workflow_env.add_owner_env(the_env)
+
+            #for event2msg in conn_info:
+            #    for k,v in event2msg:
+            #        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
+
+            #            the_env.attach_event_handler(k,_env_msg_handler)
+            #            break
 

src/aios_kernel/workflow_env.py

@@ -0,0 +1,151 @@
+
+from datetime import datetime
+import asyncio
+import sqlite3 # Because sqlite3 IO operation is small, so we can use sqlite3 directly.(so we don't need to use async sqlite3 now)
+from sqlite3 import Error
+import threading
+import logging
+from typing import Optional
+from .environment import Environment,EnvironmentEvent
+from .ai_function import SimpleAIFunction
+
+logger = logging.getLogger(__name__)
+
+
+class CalenderEvent(EnvironmentEvent):
+    def __init__(self,data) -> None:
+        super().__init__()
+        self.event_name = "timer"
+        self.data = data
+
+    def display(self) -> str:
+        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:
+        super().__init__(env_id)
+        self.is_run = False
+
+        self.add_ai_function(SimpleAIFunction("get_time",
+                                        "get current time",
+                                        self._get_now))
+
+    def _do_get_value(self,key:str) -> Optional[str]:
+        return None
+
+    def start(self) -> None:
+        if self.is_run:
+            return 
+        self.is_run = True
+
+        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')
+                env_event:CalenderEvent = CalenderEvent(formatted_time)
+                await self.fire_event("timer",env_event)
+
+            return
+
+        asyncio.create_task(timer_loop())
+
+    def stop(self):
+        self.is_run = False
+
+    def get_now(self,key:str)->str:
+        now = datetime.now()
+        formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
+        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:
+        super().__init__(env_id)
+        self.db_file = db_file
+        self.local = threading.local()
+        self.table_name = "WorkflowEnv_" + env_id
+
+
+    def _get_conn(self):
+        """ get db connection """
+        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
+        try:
+            conn = sqlite3.connect(self.db_file)
+        except Error as e:
+            logging.error("Error occurred while connecting to database: %s", e)
+            return None
+
+        if conn:
+            self._create_table(conn)
+
+        return conn
+    
+    def close(self):
+        if not hasattr(self.local, 'conn'):
+            return 
+        self.local.conn.close()
+
+    def _create_table(self, conn):
+        """ create table """
+        try:
+            # create sessions table
+            conn.execute(f"""
+                CREATE TABLE IF NOT EXISTS """ + self.table_name + """ (
+                    EnvKey TEXT PRIMARY KEY,
+                    EnvValue TEXT,
+                    UpdateTime TEXT
+                );
+            """)
+            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()
+            c = conn.cursor()
+            c.execute("SELECT EnvValue FROM " + self.table_name +" WHERE EnvKey = ?", (key,))
+            value = c.fetchone()
+            if value is None:
+                return None
+            return value[0]
+        except Error as e:
+            logging.error(f"Error occurred while _do_get_value{key}: {e}")
+            return None
+
+    def set_value(self, key: str, str_value: str, is_storage:bool=True):
+        super().set_value(key,str_value)
+        if is_storage is False:
+            return
+        
+        try:
+            conn = self._get_conn()
+            conn.execute("""
+                INSERT OR REPLACE INTO """ + self.table_name+ """ (EnvKey, EnvValue, UpdateTime)
+                VALUES (?, ?, ?) 
+            """, (key, str_value, datetime.now()))
+            conn.commit()
+            return 0  # return 0 if successful
+        except Error as e:
+            logging.error(f"Error occurred while update env{self.env_id}.{key} ,error:{e}")
+
+    def get_functions(self):
+        pass
+