Merge branch 'MVP' into MVP

.dockerignore

@@ -0,0 +1,27 @@
+**/__pycache__
+**/*venv
+**/.classpath
+**/.dockerignore
+**/.env
+**/.git
+**/.gitignore
+**/.project
+**/.settings
+**/.toolstarget
+**/.vs
+**/.vscode
+**/*.*proj.user
+**/*.dbmdl
+**/*.jfm
+**/bin
+**/charts
+**/docker-compose*
+**/compose*
+**/Dockerfile*
+**/node_modules
+**/npm-debug.log
+**/obj
+**/secrets.dev.yaml
+**/values.dev.yaml
+**/.db
+**/.python-version

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.11
+WORKDIR /opt/aios
+COPY ./src /opt/aios
+COPY ./rootfs /opt/aios/app
+
+RUN mkdir -p /root/myai/app
+RUN mkdir -p /root/myai/data
+RUN mkdir -p /root/myai/etc
+
+RUN pip install --no-cache-dir -r /opt/aios/requirements.txt
+
+ENV PYTHONDONTWRITEBYTECODE 1
+ENV PYTHONUNBUFFERED 1
+
+CMD ["python3","./service/aios_shell/aios_shell.py"]

build_all_in_one.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/bash
+
+pipreqs ./src --force
+# Build the docker image
+docker build -t aios .

doc/mvp/workflow.drawio

@@ -1,6 +1,6 @@
 <mxfile host="65bd71144e" pages="3">
     <diagram id="C5RBs43oDa-KdzZeNtuy" name="Page-1">
-        <mxGraphModel dx="1881" dy="676" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="827" pageHeight="1169" math="0" shadow="0">
+        <mxGraphModel dx="2069" dy="1139" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="827" pageHeight="1169" math="0" shadow="0">
             <root>
                 <mxCell id="WIyWlLk6GJQsqaUBKTNV-0"/>
                 <mxCell id="WIyWlLk6GJQsqaUBKTNV-1" parent="WIyWlLk6GJQsqaUBKTNV-0"/>
@@ -365,4 +365,30 @@
             </root>
         </mxGraphModel>
     </diagram>
+    <diagram id="zNgk-d6xdACtSja1wnUq" name="Page-4">
+        <mxGraphModel dx="2069" dy="1139" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1100" pageHeight="850" math="0" shadow="0">
+            <root>
+                <mxCell id="0"/>
+                <mxCell id="1" parent="0"/>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-1" value="ChatSession" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="140" y="150" width="120" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-2" value="SubSession" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="230" y="240" width="120" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-3" value="SubSession" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="230" y="330" width="120" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-4" value="Agent Message" style="shape=hexagon;perimeter=hexagonPerimeter2;whiteSpace=wrap;html=1;fixedSize=1;" vertex="1" parent="1">
+                    <mxGeometry x="600" y="150" width="130" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-5" value="Agent Message" style="shape=hexagon;perimeter=hexagonPerimeter2;whiteSpace=wrap;html=1;fixedSize=1;" vertex="1" parent="1">
+                    <mxGeometry x="600" y="230" width="130" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="PKOyaMforMDOFoUY4PA1-6" value="Agent Message" style="shape=hexagon;perimeter=hexagonPerimeter2;whiteSpace=wrap;html=1;fixedSize=1;" vertex="1" parent="1">
+                    <mxGeometry x="600" y="320" width="130" height="60" as="geometry"/>
+                </mxCell>
+            </root>
+        </mxGraphModel>
+    </diagram>
 </mxfile>

rootfs/agents/agent_1/agent.toml

@@ -1,5 +1,6 @@
 instance_id = "agent_1"
 fullname = "tracy wang"
+
 [[prompt]]
 role = "system"
 content = "你是我的私人英文老师，和我用地道的美式英语进行交流。你会在和我交流的同时，调整我的输入成为更地道的美式句子，并根据你对我英文水平的预测，对可能发错英的单词标上音标。如果我给你发中文，说明我不知道这句话用美式英语怎么说，你依旧按上述规则回应我。"

rootfs/tunnels.cfg.toml

@@ -0,0 +1,19 @@
+[[tunnels]]
+tunnel_id = "MyRoobot"
+type="TelegramTunnel"
+target="agent_1"
+token="your_token"
+
+
+[[tunnels]]
+tunnel_id="MyEmailRobot"
+type="EmailTunnel"
+target="agent_1"
+
+email="youremail@msn.com"
+imap="outlook.office365.com:993"
+smtp="outlook.office365.com:587"
+user=""
+password=""
+folder="inbox"
+interval=10

rootfs/workflows/math_school/workflow.toml

@@ -7,7 +7,7 @@ GOAL="成为最好的学校"
 [[connected_env]]
 env_id = "calender"
 [[connected_env.event2msg]]
-timer = "现在是{data}"
+timer = "现在是{now}"
 role = "教导处主任"
 
 [filter]

rootfs/workflows/plan_maker/workflow.toml

@@ -42,11 +42,13 @@ fullname = "经理"
 agent="manager"
 [[roles.manager.prompt]] 
 role="system"
-content="""你是一个活动策划公司的经理,与客户对接并向团队下达指令。你的团队分为下面几个小组:嘉宾对接组,酒店预定组,行程预订组,财务组,活动摄像组。活动策划分为四个阶段：方案讨论，活动前，活动中，活动后。你会根据客户的需求，对团队进行分工，分别完成各个阶段的工作。你的基本工作模式是：
+content="""
+你是一个活动策划公司的经理,与客户对接并向团队下达指令。你的团队分为下面几个小组:嘉宾对接组,酒店预定组,行程预订组,财务组,活动摄像组。活动策划分为四个阶段：方案讨论，活动前，活动中，活动后。你会根据客户的需求，对团队进行分工，分别完成各个阶段的工作。你的基本工作模式是：
 1. 收到客户的明确的指令后,基于客户的已有信息和客户商量活动方案，和活动策划公司无关的业务你会回答‘与我无关’。当和客户完成活动方案的确认后,你会将拆解后的任务分配给各个小组 
 2. 根据目前已经确认的活动方案，你要根据时间适时的检查不同小组的工作情况。当收到小组的工作情况反馈后，你会站在全局的角度判断是否需要调整活动方案，如果需要调整，你会和客户商量重新确定方案，然后再将调整后的方案分配给各个小组。
 3. 有时工作小组会主动与你沟通，反馈一些问题。你会站在全局的角度给与指导，适当的调整工作小组的工作目标。如果反馈的问题需要你和客户沟通，你会和客户沟通后重新确定方案。再将调整后的方案分配给受到影响各个小组。
-4. 当你决定要和工作小组通信时，请使用 sendmsg(小组名称,内容) 的形式。"""
+4. 当你决定要和工作小组通信时，请使用 sendmsg(小组名称,内容) 的形式。
+"""
 
 
 

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
@@ -10,3 +10,12 @@ from .workflow import Workflow
 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

@@ -7,7 +7,7 @@ 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
@@ -73,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  
@@ -82,6 +82,7 @@ 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):
@@ -90,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
@@ -100,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"]
 
@@ -147,7 +148,7 @@ class AIAgent:
 
         return result_func 
 
-    async def _execute_func(self,inenr_func_call_node:dict,msg_prompt:AgentPrompt) -> str:
+    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")
@@ -157,14 +158,21 @@ class AIAgent:
         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()
-        msg_prompt.messages.append({"role":"function","content":result_str,"name":func_name})
-        task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(msg_prompt,self.llm_model_name,self.max_token_size,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,msg_prompt)       
+            return await self._execute_func(inner_func_call_node,prompt,org_msg)      
         else:
             return task_result.result_str
 
@@ -172,7 +180,13 @@ class AIAgent:
             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_knowlege_prompt(the_role.get_name()))
@@ -184,72 +198,32 @@ class AIAgent:
 
             inner_functions = self._get_inner_functions()
 
-            task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,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:
-                final_result = await self._execute_func(inner_func_call_node,msg_prompt)
+                #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
@@ -263,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

@@ -3,26 +3,100 @@ 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
@@ -31,7 +105,7 @@ 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
@@ -39,6 +113,12 @@ class AgentMsg:
     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)

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,11 +43,13 @@ class AIBus:
         return cls._instance
 
     def __init__(self) -> None:
-        self.handlers = {}
-        self.unhandle_handler = None
+        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]
 
-    async def post_message(self,target_id,msg:AgentMsg,use_unhandle=True) -> bool:
-        target_id = target_id.split(".")[0]
         handler = self.handlers.get(target_id)
         if handler:
             handler.queue.put_nowait(msg)
@@ -44,46 +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:
-        target_id = target_id.split(".")[0]
-        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
@@ -93,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
     
@@ -102,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
@@ -127,13 +134,3 @@ class AIBus:
             return
         
         handler.working_task = asyncio.create_task(self.process_queue(handler))
-
-    
-
-#send message to target logic:
-# find target handler:
-#   process_message(msg):
-#       session = get_session(msg.sender,msg.target)
-
-
-# history: open(sender,target,topic)

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,
+                    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

@@ -17,19 +17,14 @@ logger = logging.getLogger(__name__)
 
 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 = {}

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

@@ -70,12 +70,6 @@ class Environment:
     def _do_get_value(self,key:str) -> Optional[str]:
         pass
 
-    def get_functions(self):
-        # system functions
-        # env functions
-        # user install functions
-        pass
-
     def register_get_handler(self,key:str,handler:Callable) -> None:
         h = self.get_handlers.get(key)
         if h is not None:

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/open_ai_node.py

@@ -7,38 +7,48 @@ import logging
 
 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
+        self.is_start = False
         # openai.organization = "org-AoKrOtF2myemvfiFfnsSU8rF" #buckycloud
-        self.openai_api_key = ""
+        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")
-        else:
-            openai.api_key = self.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:
+            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()
+        return True
 
     async def push_task(self, task: ComputeTask, proiority: int = 0):
         logger.info(f"openai_node push task: {task.display()}")
@@ -54,14 +64,19 @@ class OpenAI_ComputeNode(ComputeNode):
         prompts = task.params["prompts"]
 
         logger.info(f"call openai {mode_name} prompts: {prompts}")
-        try:
+
+        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?
-        except Exception as e:
-            logger.error(f"openai.ChatCompletion.create failed! {e}")
+
         
         logger.info(f"openai response: {resp}")
         
@@ -86,6 +101,10 @@ class OpenAI_ComputeNode(ComputeNode):
         return result
 
     def start(self):
+        if self.is_start is True:
+            return
+        self.is_start = True
+        
         async def _run_task_loop():
             while True:
                 task = await self.task_queue.get()

src/aios_kernel/stability_node.py

@@ -10,26 +10,28 @@ 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
     
-    def __new__(cls):
-        if cls._instanace is None:
-            cls._instanace = super(Stability_ComputeNode, cls).__new__(cls)
-            cls._instanace.is_start = False
-        return cls._instanace
+    @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__()
-        if self.is_start is True:
-            logger.warn("Stability_ComputeNode is already start")
-            return
 
-        self.is_start = True
+        self.is_start = False
         self.node_id = "stability_node"
         self.api_key = ""
         self.engine = "stable-diffusion-512-v2-1"
@@ -88,8 +90,8 @@ class Stability_ComputeNode(ComputeNode):
 
         for resp in answers:
             for artifact in resp.artifacts:
-                logger.info("artifact:", artifact.id,
-                            artifact.type, artifact.finish_reason)
+                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:
@@ -108,6 +110,9 @@ class Stability_ComputeNode(ComputeNode):
         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...")

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

@@ -2,21 +2,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__)
 
 class MessageFilter:
@@ -120,7 +122,11 @@ class Workflow:
         connected_env_ndoe = config.get("connected_env") 
         if  connected_env_ndoe is not None:
            for _node in connected_env_ndoe:
-                remote_env = Environment.get_env_by_id(_node.get("env_id"))
+                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
@@ -158,11 +164,12 @@ class Workflow:
         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)
+                    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)
+                    return await sub_workflow._process_msg(msg)
                 logger.error(f"{msg.target} not found! forword message failed!")
                 return None
             else:
@@ -176,17 +183,49 @@ class Workflow:
         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):
-        # workflow can be a message handler, but never be a message sender
-        # all message forword to roles or sub workflow
-        # workflow no chatsession record, but role have
-        
-        final_result = None
-
+        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
+        
+        # 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)
@@ -196,26 +235,12 @@ class Workflow:
                     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
-
-                final_result = result
+                return await self.role_process_msg(msg,select_role,chatsesssion)
             else:
-                logger.error(f"input_filter return None for :{msg}")
-                return
-                
-        else:
-            # no input filter, we would process all message, slowly,not suggest to use 
-            results = {}
-            final_result:AgentMsg  = None 
-            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
-                final_result = a_result
+                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
@@ -276,52 +301,59 @@ class Workflow:
 
         return r
 
-    async def role_post_msg(self,msg:AgentMsg,the_role:AIRole):
+    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.id} to inner role: {msg.target}")
-            asyncio.create_task(self.role_process_msg(msg,target_role))
+            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.id} to sub workflow: {msg.target}")
+            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.id} to AIBus: {msg.target}")
+        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):
+    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.id} to inner role: {msg.target}")
-            return await self.role_process_msg(msg,target_role)
-
+            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)
 
         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.id} to sub workflow: {msg.target}")
+            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.id} to AIBus: {msg.target}")
-        return await self.get_bus().send_message(msg.target,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]}")
-        return """{result:"timeout"}"""
+        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
+        return await self.role_call(call,the_role)
 
     def _format_msg_by_env_value(self,prompt:AgentPrompt):
         if self.workflow_env is None:
@@ -331,66 +363,148 @@ class Workflow:
             old_content = msg.get("content")
             msg["content"] = old_content.format_map(self.workflow_env)
 
-    async def role_process_msg(self,msg:AgentMsg,the_role:AIRole):
-        session_topic = f"{msg.sender}#{msg.topic}"
-        session_owner = the_role.get_role_id()
-        chatsession : AIChatSession = AIChatSession.get_session(session_owner,session_topic,self.db_file)
-        if chatsession is None:
-            logger.error(f"get session {session_topic}@{session_owner} failed!")
+    def _get_inner_functions(self) -> dict:
+        all_inner_function = self.workflow_env.get_all_ai_functions()
+        if all_inner_function is None:
             return None
         
+        result_func = []
+        for inner_func in all_inner_function:
+            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_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))
+        
+        #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)
 
         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().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size())
+            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
-            result = Workflow.prase_llm_result(result_str)
             logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}")
+            
+            inner_func_call_node = task_result.result_message.get("function_call")
+            
+            if inner_func_call_node:
+                #TODO to save more token ,can i use msg_prompt?
+                result_str = await self._role_execute_func(the_role,inner_func_call_node,prompt,msg)
+
+            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
-                await self.role_post_msg(postmsg,the_role)
+                
+                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 = AgentMsg()
-                    resp_msg.topic = msg.topic
-                    resp_msg.set(session_owner,msg.sender,result.resp)
-            
-                    chatsession.append_recv(msg)
-                    chatsession.append_post(resp_msg)
+                    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
-                        send_resp = await self.role_send_msg(sendmsg,the_role)
+                        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}]
@@ -411,17 +525,9 @@ class Workflow:
         
         return result_prompt
     
-
-    def _get_function_prompt(self,role_name:str) -> AgentPrompt:
-        # system functions
-        # env functions
-        # user install functions
-        pass    
-    
     def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt:
         pass
 
-
     def get_workflow_rule_prompt(self) -> AgentPrompt:
         return self.rule_prompt
 
@@ -447,6 +553,4 @@ class Workflow:
 
             #            the_env.attach_event_handler(k,_env_msg_handler)
             #            break
-        else:
-            logger.warn(f"environment {env.get_id()} already connected!")
 

src/aios_kernel/workflow_env.py

@@ -29,7 +29,7 @@ class CalenderEnvironment(Environment):
 
         self.add_ai_function(SimpleAIFunction("get_time",
                                         "get current time",
-                                        self.get_now))
+                                        self._get_now))
 
     def _do_get_value(self,key:str) -> Optional[str]:
         return None
@@ -58,7 +58,8 @@ class CalenderEnvironment(Environment):
     def stop(self):
         self.is_run = False
 
-    def get_now(self) -> str:
+    async def _get_now(self) -> str:
+
         now = datetime.now()
         formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
         return formatted_time
@@ -71,6 +72,7 @@ class WorkflowEnvironment(Environment):
         self.local = threading.local()
         self.table_name = "WorkflowEnv_" + env_id
 
+
     def _get_conn(self):
         """ get db connection """
         if not hasattr(self.local, 'conn'):

src/component/agent_manager/agent_manager.py

@@ -2,7 +2,7 @@
 import logging
 import toml
 
-from aios_kernel import AIAgent,AIAgentTemplete
+from aios_kernel import AIAgent,AIAgentTemplete,AIStorage
 from package_manager import PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask
 
 logger = logging.getLogger(__name__)
@@ -11,16 +11,19 @@ logger = logging.getLogger(__name__)
 class AgentManager:
     _instance = None
    
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls)->'AgentManager':
         if cls._instance is None:
-            cls._instance = super(AgentManager, cls).__new__(cls)
+            cls._instance = AgentManager()
         return cls._instance
     
+    def initial(self) -> None:
+        system_app_dir = AIStorage.get_instance().get_system_app_dir()
+        user_data_dir = AIStorage.get_instance().get_myai_dir()
 
-    def initial(self,root_dir:str) -> None:
-        self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}/templetes/templetes.cfg")
-        self.agent_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}/agents/agents.cfg")
-        self.db_path = f"{root_dir}/agents_chat.db"
+        self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{system_app_dir}/templates/templetes.cfg")
+        self.agent_env : PackageEnv = PackageEnvManager().get_env(f"{system_app_dir}/agents/agents.cfg")
+        self.db_path = f"{user_data_dir}/messages.db"
         self.loaded_agent_instance = {}
         if self.agent_templete_env is None:
             raise Exception("agent_manager initial failed")

src/component/package_manager/env.py

@@ -134,16 +134,15 @@ class PackageEnv:
 
 class PackageEnvManager:
     _instance = None
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super(PackageEnvManager, cls).__new__(cls)
+            cls._instance = PackageEnvManager()
         return cls._instance
     
     def __init__(self) -> None:
         self._pkg_envs = {}
         
-        pass
-
     def get_env(self,cfg_path:str) -> PackageEnv:
         if cfg_path in self._pkg_envs:
             return self._pkg_envs[cfg_path]

src/component/workflow_manager/workflow_manager.py

@@ -1,24 +1,29 @@
 import logging
 import toml
-from aios_kernel import Workflow
+import os
+
+from aios_kernel import Workflow,AIStorage
 from package_manager import PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask
 from agent_manager import AgentManager
 logger = logging.getLogger(__name__)
-import os
 
 class WorkflowManager:
     _instance = None
 
-    def __new__(cls):
+    @classmethod
+    def get_instance(cls):
         if cls._instance is None:
-            cls._instance = super(WorkflowManager, cls).__new__(cls)
+            cls._instance = WorkflowManager()
         return cls._instance
 
 
-    def initial(self,root_dir:str) -> None:
+    def initial(self) -> None:
         self.loaded_workflow = {}
-        self.workflow_env = PackageEnvManager().get_env(f"{root_dir}/workflows.cfg")
-        self.db_file = os.path.abspath(f"{root_dir}/workflows.db")
+        system_app_dir = AIStorage.get_instance().get_system_app_dir()
+        user_data_dir = AIStorage.get_instance().get_myai_dir()
+
+        self.workflow_env = PackageEnvManager().get_env(f"{system_app_dir}/workflows.cfg")
+        self.db_file = os.path.abspath(f"{user_data_dir}/messages.db")
         if self.workflow_env is None:
             raise Exception("WorkflowManager initial failed")
         

src/requirements.txt

@@ -1,8 +1,15 @@
 aiofiles==23.2.1
-aiohttp==3.8.5
-openai==0.28.0
-Pillow==10.0.0
-Pillow==10.0.0
+aiohttp==3.7.0
+aioimaplib==1.0.1
+aiosmtplib==2.0.2
+beautifulsoup4==4.12.2
+mail_parser==3.15.0
+openai==0.27.10
+Pillow==10.0.1
+Pillow==10.0.1
 prompt_toolkit==3.0.39
+python-telegram-bot==20.5
+Requests==2.31.0
 stability_sdk==0.8.4
 toml==0.10.2
+google-cloud-texttospeech==2.14.1

src/service/aios_shell/aios_shell.py

@@ -4,6 +4,7 @@ import sys
 import os
 import logging
 import re
+import toml
 
 from typing import Any, Optional, TypeVar, Tuple, Sequence
 import argparse
@@ -17,6 +18,19 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import WordCompleter
 from prompt_toolkit.styles import Style
 
+directory = os.path.dirname(__file__)
+sys.path.append(directory + '/../../')
+
+from aios_kernel import AIOS_Version,UserConfigItem,AIStorage,Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel,LocalLlama_ComputeNode
+
+
+sys.path.append(directory + '/../../component/')
+from agent_manager import AgentManager
+from workflow_manager import WorkflowManager
+
+
+logger = logging.getLogger(__name__)
+
 shell_style = Style.from_dict({
     'title': '#87d7ff bold', #RGB
     'content': '#007f00 bold',
@@ -24,31 +38,30 @@ shell_style = Style.from_dict({
 })
 
 
-directory = os.path.dirname(__file__)
-sys.path.append(directory + '/../../')
-from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,LocalLlama_ComputeNode
-
-sys.path.append(directory + '/../../component/')
-from agent_manager import AgentManager
-from workflow_manager import WorkflowManager
-from aios_kernel import CalenderEnvironment,Environment
-
-
 class AIOS_Shell:
     def __init__(self,username:str) -> None:
         self.username = username
         self.current_target = "_"
         self.current_topic = "default"
+        self.is_working = True
+
+    def declare_all_user_config(self):
+        user_config = AIStorage.get_instance().get_user_config()
+        user_config.add_user_config("username","username is your full name when using AIOS",False,None,)
+
+        openai_node = OpenAI_ComputeNode.get_instance()
+        openai_node.declare_user_config()
+
 
     async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool:
         target_id = msg.target.split(".")[0]
-        agent : AIAgent = await AgentManager().get(target_id)
+        agent : AIAgent = await AgentManager.get_instance().get(target_id)
         if agent is not None:
             agent.owner_env = Environment.get_env_by_id("calender") 
             bus.register_message_handler(target_id,agent._process_msg)
             return True
         
-        a_workflow = await WorkflowManager().get_workflow(target_id)
+        a_workflow = await WorkflowManager.get_instance().get_workflow(target_id)
         if a_workflow is not None:
             bus.register_message_handler(target_id,a_workflow._process_msg)
             return True
@@ -56,7 +69,7 @@ class AIOS_Shell:
         return False
     
     async def is_agent(self,target_id:str) -> bool:
-        agent : AIAgent = await AgentManager().get(target_id)
+        agent : AIAgent = await AgentManager.get_instance().get(target_id)
         if agent is not None:
             return True
         else:
@@ -67,35 +80,55 @@ class AIOS_Shell:
         cal_env.start()
         Environment.set_env_by_id("calender",cal_env)
         
-        AgentManager().initial(os.path.abspath(directory + "/../../../rootfs/"))
-        WorkflowManager().initial(os.path.abspath(directory + "/../../../rootfs/workflows/"))
+        AgentManager.get_instance().initial()
+        WorkflowManager.get_instance().initial()
 
-        open_ai_node = OpenAI_ComputeNode()
-        open_ai_node.start()
-        ComputeKernel().add_compute_node(open_ai_node)
+        open_ai_node = OpenAI_ComputeNode.get_instance()
+        if await open_ai_node.initial() is not True:
+            logger.error("openai node initial failed!")
+            return False
+        
+        ComputeKernel.get_instance().add_compute_node(open_ai_node)
         
         llama_ai_node = LocalLlama_ComputeNode()
         llama_ai_node.start()
         ComputeKernel().add_compute_node(llama_ai_node)
 
         AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg)
+        AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
+
+        TelegramTunnel.register_to_loader()
+        EmailTunnel.register_to_loader()
+
+        user_data_dir = AIStorage.get_instance().get_myai_dir()
+        tunnels_config_path = os.path.abspath(f"{user_data_dir}/tunnels.cfg.toml")
+        tunnel_config = None
+        try: 
+            tunnel_config = toml.load(tunnels_config_path)
+            if tunnel_config is not None:
+                await AgentTunnel.load_all_tunnels_from_config(tunnel_config["tunnels"])
+        except Exception as e:
+            logger.warning(f"load tunnels config from {tunnels_config_path} failed!")
+            
+
+
         return True 
         
 
     def get_version(self) -> str:
-        return "0.0.1"
+        return "0.5.1"
 
     async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str:
         agent_msg = AgentMsg()
         agent_msg.set(sender,target_id,msg)
         agent_msg.topic = topic
-        resp = await AIBus.get_default_bus().send_message(target_id,agent_msg)
+        resp = await AIBus.get_default_bus().send_message(agent_msg)
         if resp is not None:
             return resp.body
         else:
             return "error!"
 
-    async def install_workflow(self,workflow_id:Workflow) -> None:
+    async def _user_process_msg(self,msg:AgentMsg) -> AgentMsg:
         pass
 
     async def call_func(self,func_name, args):
@@ -121,6 +154,7 @@ class AIOS_Shell:
             case 'login':
                 if len(args) >= 1:
                     self.username = args[0]
+                AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
                 return self.username + " login success!"    
             case 'history':
                 num = 10
@@ -133,14 +167,14 @@ class AIOS_Shell:
 
                 db_path = ""
                 if await self.is_agent(self.current_target):
-                    db_path = AgentManager().db_path
+                    db_path = AgentManager.get_instance().db_path
                 else:
-                    db_path = WorkflowManager().db_file
+                    db_path = WorkflowManager.get_instance().db_file
                 chatsession:AIChatSession = AIChatSession.get_session(self.current_target,f"{self.username}#{self.current_topic}",db_path,False)
                 if chatsession is not None:
                     msgs = chatsession.read_history(num,offset)
                     format_texts = []
-                    for msg in reversed(msgs):
+                    for msg in msgs:
                         format_texts.append(("class:content",f"{msg.sender} >>> {msg.body}"))
                         format_texts.append(("",f"\n-------------------\n"))
                     return FormattedText(format_texts)
@@ -151,8 +185,8 @@ class AIOS_Shell:
                 return FormattedText([("class:title", f"help~~~")])
 
 
-#######################################################################################    
-history = FileHistory('history.txt')
+##########################################################################################################################    
+history = FileHistory('aios_shell_history.txt')
 session = PromptSession(history=history) 
 
 def parse_function_call(func_string):
@@ -169,20 +203,102 @@ def parse_function_call(func_string):
 
     return func_name, params
 
+async def get_user_config_from_input(check_result:dict) -> bool:
+    for key,item in check_result.items():
+        user_input = await session.prompt_async(f"{key} ({item.desc}) not define! \nPlease input:",style=shell_style)
+        if len(user_input) > 0:
+            AIStorage.get_instance().get_user_config().set_user_config(key,user_input)
+
+    await AIStorage.get_instance().get_user_config().save_value_to_user_config()
+    return True
+
+async def main_daemon_loop(shell:AIOS_Shell):
+    while shell.is_working:
+        await asyncio.sleep(1)
+
+    return 0
+
+def print_welcome_screen():
+    print("\033[1;31m")  
+    logo = """
+\t_______                    ____________________   __
+\t__  __ \______________________  __ \__    |__  | / /
+\t_  / / /__  __ \  _ \_  __ \_  / / /_  /| |_   |/ / 
+\t/ /_/ /__  /_/ /  __/  / / /  /_/ /_  ___ |  /|  /  
+\t\____/ _  .___/\___//_/ /_//_____/ /_/  |_/_/ |_/   
+\t        /_/                                          
+
+    """
+    print(logo)
+    print("\033[0m")  
+
+    print("\033[1;32m \t\tWelcome to OpenDAN - Your Personal AI OS\033[0m\n")  
+    introduce = """
+\tThe core goal of version 0.5.1 is to turn the concept of AIOS into code and get it up and running as quickly as possible. 
+\tAfter three weeks of development, our plans have undergone some changes based on the actual progress of the system. 
+\tUnder the guidance of this goal, some components do not need to be fully implemented. Furthermore, 
+\tbased on the actual development experience from several demo Intelligent Applications, 
+\twe intend to strengthen some components. This document will explain these changes and provide an update 
+\ton the current development progress of MVP(0.5.1,0.5.2)
+
+"""
+    print(introduce)
+
+    print(f"\033[1;34m \t\tVersion: {AIOS_Version}\n\033")  
+    print("\033[1;33m \tOpenDAN is an open-source project, let's define the future of Humans and AI together.\033[0m")  
+    print("\033[1;33m \tGithub\t: https://github.com/fiatrete/OpenDAN-Personal-AI-OS\033[0m")  
+    print("\033[1;33m \tWebsite\t: https://www.opendan.ai\033[0m")
+    print("\n\n")
+
 
 async def main():
-    print("aios shell prepareing...")
+    print_welcome_screen()
+    print("OpenDAN is booting...")
     logging.basicConfig(filename="aios_shell.log",filemode="w",encoding='utf-8',force=True,
                         level=logging.INFO,
                         format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s')
-    shell = AIOS_Shell("user")
-    await shell.initial()
-    print(f"aios shell {shell.get_version()} ready.")
     
+    if os.path.isdir(f"{directory}/../../../rootfs"):
+        AIStorage.get_instance().is_dev_mode = True
+    else:
+        AIStorage.get_instance().is_dev_mode = False    
+
+    is_daemon = False
+    if os.name != 'nt':
+        if os.getppid() == 1:
+            is_daemon = True
+
+    shell = AIOS_Shell("user")   
+    shell.declare_all_user_config() 
+    await AIStorage.get_instance().initial()
+    check_result = AIStorage.get_instance().get_user_config().check_user_config()
+    if check_result is not None:
+        if is_daemon:
+            logger.error(check_result)
+            return 1
+        else:
+            #Remind users to enter necessary configurations.
+            if await get_user_config_from_input(check_result) is False:
+                return 1
+    
+    init_result = await shell.initial()
+    if init_result is False:
+        if is_daemon:
+            logger.error("aios shell initial failed!")
+            return 1
+        else:
+            print("aios shell initial failed!")
+
+    print(f"aios shell {shell.get_version()} ready.")
+    if is_daemon:
+        return await main_daemon_loop(shell)
+
+    #TODO: read last input config
     completer = WordCompleter(['send($target,$msg,$topic)', 
                                'open($target,$topic)', 
                                'history($num,$offset)',
-                               'login($username)'
+                               'login($username)',
+                               'connect($target)',
                                'show()',
                                'exit()', 
                                'help()'], ignore_case=True)
@@ -206,8 +322,6 @@ async def main():
         print_formatted_text(show_text,style=shell_style)
         #print_formatted_text(f"{shell.username}<->{shell.current_topic}@{shell.current_target} >>> {resp}",style=shell_style)
 
-    
-
 if __name__ == "__main__":    
     asyncio.run(main())
 

src/service/aios_shell/tg_aios_shell.py

@@ -1,223 +0,0 @@
-
-import asyncio
-import sys
-import os
-import logging
-import re
-
-from typing import Any, Optional, TypeVar, Tuple, Sequence
-import argparse
-
-from telegram import ForceReply, Update
-from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
-
-
-directory = os.path.dirname(__file__)
-sys.path.append(directory + '/../../')
-from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession
-
-sys.path.append(directory + '/../../component/')
-from agent_manager import AgentManager
-from workflow_manager import WorkflowManager
-from aios_kernel import CalenderEnvironment,Environment
-
-import logging
-
-# Enable logging
-logging.basicConfig(
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
-)
-# set higher logging level for httpx to avoid all GET and POST requests being logged
-logging.getLogger("httpx").setLevel(logging.WARNING)
-
-logger = logging.getLogger(__name__)
-
-class AIOS_Shell:
-    def __init__(self,username:str) -> None:
-        self.username = username
-        self.current_target = "_"
-        self.current_topic = "default"
-
-    async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool:
-        target_id = msg.target.split(".")[0]
-        agent : AIAgent = await AgentManager().get(target_id)
-        if agent is not None:
-            bus.register_message_handler(target_id,agent._process_msg)
-            return True
-        
-        a_workflow = await WorkflowManager().get_workflow(target_id)
-        if a_workflow is not None:
-            bus.register_message_handler(target_id,a_workflow._process_msg)
-            return True
-        
-        return False
-    
-    async def is_agent(self,target_id:str) -> bool:
-        agent : AIAgent = await AgentManager().get(target_id)
-        if agent is not None:
-            return True
-        else:
-            return False
-
-    async def initial(self) -> bool:
-        cal_env = CalenderEnvironment("calender")
-        cal_env.start()
-        Environment.set_env_by_id("calender",cal_env)
-        
-        AgentManager().initial(os.path.abspath(directory + "/../../../rootfs/"))
-        WorkflowManager().initial(os.path.abspath(directory + "/../../../rootfs/workflows/"))
-        open_ai_node = OpenAI_ComputeNode()
-        open_ai_node.start()
-        ComputeKernel().add_compute_node(open_ai_node)
-        AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg)
-        return True 
-        
-
-    def get_version(self) -> str:
-        return "0.0.1"
-
-    async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str:
-        agent_msg = AgentMsg()
-        agent_msg.set(sender,target_id,msg)
-        agent_msg.topic = topic
-        resp = await AIBus.get_default_bus().send_message(target_id,agent_msg)
-        if resp is not None:
-            return resp.body
-        else:
-            return "error!"
-
-    async def install_workflow(self,workflow_id:Workflow) -> None:
-        pass
-
-    async def call_func(self,func_name, args):
-        match func_name:
-            case 'send':
-                target_id = args[0]
-                msg_content = args[1]
-                topic = args[2]
-                resp = await self.send_msg(msg_content,target_id,topic,self.username)
-        
-            
-                return resp
-            case 'open':
-                if len(args) >= 1:
-                    target_id = args[0]
-                if len(args) >= 2:
-                    topic = args[1]
-
-                self.current_target = target_id
-                self.current_topic = topic
-                show_text = f"current session switch to {topic}@{target_id}"
-                return show_text
-  
-            
-            case 'history':
-                num = 10
-                offset = 0
-                if args is not None:
-                    if len(args) >= 1:
-                        num = args[0]
-                    if len(args) >= 2:
-                        offset = args[1]
-
-                db_path = ""
-                if await self.is_agent(self.current_target):
-                    db_path = AgentManager().db_path
-                else:
-                    db_path = WorkflowManager().db_file
-                chatsession:AIChatSession = AIChatSession.get_session(self.current_target,f"{self.username}#{self.current_topic}",db_path,False)
-                if chatsession is not None:
-                    msgs = chatsession.read_history(num,offset)
-                    result_str = ""
-                    for msg in reversed(msgs):
-                        result_str += f"{msg.sender} >>> {msg.body}"
-                        result_str += f"\n-------------------\n"
-                    return result_str
-                return f"chatsession not found"
-            case 'help':
-                return "help info "
-
-shell:AIOS_Shell = None
-
-async def _init_shell():
-    global shell
-    if shell is None:
-        shell = AIOS_Shell("user")
-        await shell.initial()
-    return 
-
-def parse_function_call(func_string):
-    match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string)
-    if not match:
-        return None
-
-    func_name = match.group(1)
-    params_string = match.group(2).strip()    
-    params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string)
-    params = [param.strip('"') for param in params]
-    if len(params[0]) == 0:
-        params = None
-
-    return func_name, params
-
-async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-    """Send a message when the command /start is issued."""
-    await _init_shell()
-    user = update.effective_user
-    await update.message.reply_html(
-        rf"Hi {user.mention_html()}!",
-        reply_markup=ForceReply(selective=True),
-    )
-
-
-async def on_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-    await _init_shell()
-    cmd_msg = update.message.text[1:]
-    logger.info("command:" + cmd_msg)
-    func_call = parse_function_call(cmd_msg)
-    if func_call:
-        shell.username = update.effective_user
-        show_text = await shell.call_func(func_call[0], func_call[1])
-        await update.message.reply_text(show_text)
-    else:
-        await update.message.reply_text("command not found!")
-
-async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-    await _init_shell()
-    shell.username = update.effective_user
-    resp = await shell.send_msg(update.message.text,shell.current_target,shell.current_topic,shell.username)
-    await update.message.reply_text(resp)
-
-
-
-
-def main() -> None:
-    """Start the bot."""
-    # Create the Application and pass it your bot's token.
-    tg_robot_token = os.getenv("TG_ROBOT_TOKEN")
-    if tg_robot_token is None:
-        print("TG_ROBOT_TOKEN not found!")
-        tg_robot_token = input("input your telegram robot token:")
-
-
-    if tg_robot_token is None:
-        return
-
-    application = Application.builder().token(tg_robot_token).build()
-
-    # on different commands - answer in Telegram
-    application.add_handler(CommandHandler("start", start))
-    application.add_handler(CommandHandler("help", on_command))
-    application.add_handler(CommandHandler("send", on_command))
-    application.add_handler(CommandHandler("open", on_command))
-    application.add_handler(CommandHandler("history", on_command))
-
-    # on non command i.e message - echo the message on Telegram
-    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_message))
-
-    # Run the bot until the user presses Ctrl-C
-    application.run_polling(allowed_updates=Update.ALL_TYPES)
-
-
-if __name__ == "__main__":
-    main()

test/daemon_test.py

@@ -0,0 +1,17 @@
+import daemon
+from time import sleep
+import logging
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(filename="daemon_test.log",filemode="w",encoding='utf-8',force=True,
+                    level=logging.INFO,
+                    format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s')
+
+def main_program():
+    while True:
+        logger.info("hello world")
+        sleep(1)
+
+with daemon.DaemonContext():
+    main_program()