From c4d97942cb25d81d6f1cb040a115716bf34ae4eb Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Thu, 14 Sep 2023 01:50:18 -0700 Subject: [PATCH 1/5] Refactor chatsession impl, fix bug. --- doc/mvp/workflow.drawio | 28 ++- rootfs/agents/agent_1/agent.toml | 1 + rootfs/workflows/math_school/workflow.toml | 2 +- rootfs/workflows/plan_maker/workflow.toml | 6 +- src/aios_kernel/__init__.py | 2 +- src/aios_kernel/agent.py | 91 +++----- src/aios_kernel/agent_message.py | 96 +++++++- src/aios_kernel/bus.py | 95 ++++---- src/aios_kernel/chatsession.py | 100 ++++++--- src/aios_kernel/environment.py | 6 - src/aios_kernel/open_ai_node.py | 12 +- src/aios_kernel/workflow.py | 250 ++++++++++++++------- src/aios_kernel/workflow_env.py | 10 +- src/service/aios_shell/aios_shell.py | 12 +- src/service/aios_shell/tg_aios_shell.py | 2 +- 15 files changed, 477 insertions(+), 236 deletions(-) diff --git a/doc/mvp/workflow.drawio b/doc/mvp/workflow.drawio index 54e724c..1ca245b 100644 --- a/doc/mvp/workflow.drawio +++ b/doc/mvp/workflow.drawio @@ -1,6 +1,6 @@ - + @@ -365,4 +365,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/rootfs/agents/agent_1/agent.toml b/rootfs/agents/agent_1/agent.toml index 81966ab..53d0605 100644 --- a/rootfs/agents/agent_1/agent.toml +++ b/rootfs/agents/agent_1/agent.toml @@ -1,5 +1,6 @@ instance_id = "agent_1" fullname = "tracy wang" + [[prompt]] role = "system" content = "你是我的私人英文老师，和我用地道的美式英语进行交流。你会在和我交流的同时，调整我的输入成为更地道的美式句子，并根据你对我英文水平的预测，对可能发错英的单词标上音标。如果我给你发中文，说明我不知道这句话用美式英语怎么说，你依旧按上述规则回应我。" diff --git a/rootfs/workflows/math_school/workflow.toml b/rootfs/workflows/math_school/workflow.toml index 48c9831..69b998b 100644 --- a/rootfs/workflows/math_school/workflow.toml +++ b/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] diff --git a/rootfs/workflows/plan_maker/workflow.toml b/rootfs/workflows/plan_maker/workflow.toml index 1dc80bf..3975313 100644 --- a/rootfs/workflows/plan_maker/workflow.toml +++ b/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(小组名称,内容) 的形式。 +""" diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index bef117f..fb09d88 100644 --- a/src/aios_kernel/__init__.py +++ b/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 diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 0a3f8b0..0c3e106 100644 --- a/src/aios_kernel/agent.py +++ b/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 : AIBus = 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) + inner_functions = self._get_inner_functions() + prompt.messages.append({"role":"function","content":result_str,"name":func_name}) + task_result:ComputeTaskResult = await ComputeKernel().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())) @@ -189,67 +203,27 @@ class AIAgent: 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,7 +237,8 @@ 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): diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py index 242cc66..0640319 100644 --- a/src/aios_kernel/agent_message.py +++ b/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) diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index ae1853e..22bf209 100644 --- a/src/aios_kernel/bus.py +++ b/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 - - async def post_message(self,target_id,msg:AgentMsg,use_unhandle=True) -> bool: - target_id = target_id.split(".")[0] + self.handlers:Dict[AIBusHandler] = {} + self.unhandle_handler:Coroutine = None + + + async def post_message(self,msg:AgentMsg,use_unhandle=True) -> bool: + target_id = msg.target.split(".")[0] + handler = self.handlers.get(target_id) if handler: handler.queue.put_nowait(msg) @@ -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 @@ -126,14 +133,4 @@ class AIBus: logger.warn(f"handler {target_name} is already working!") 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) \ No newline at end of file + handler.working_task = asyncio.create_task(self.process_queue(handler)) \ No newline at end of file diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index ecd90a3..5893683 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -5,8 +5,9 @@ import logging import threading import datetime import uuid +import json -from .agent_message import AgentMsg +from .agent_message import AgentMsgType, AgentMsg, AgentMsgStatus class ChatSessionDB: def __init__(self, db_file): @@ -54,14 +55,31 @@ class ChatSessionDB: """) # create messages table + # reciver_id could be None + conn.execute(""" CREATE TABLE IF NOT EXISTS Messages ( MessageID TEXT PRIMARY KEY, SessionID TEXT, - SenderID TEXT, + MsgType INTEGER, + PrevMsgID TEXT, + QuoteMsgID TEXT, + RelyMsgID TEXT, + + SenderID TEXT, ReceiverID TEXT, Timestamp TEXT, + + Topic TEXT, + Mentions TEXT, + ContentMIME TEXT, Content TEXT, + + ActionName TEXT, + ActionParams TEXT, + ActionResult TEXT, + DoneTime TEXT, + Status INTEGER ); """) @@ -83,15 +101,43 @@ class ChatSessionDB: logging.error("Error occurred while inserting session: %s", e) return -1 # return -1 if an error occurs - def insert_message(self, message_id, session_id, sender_id, receiver_id, timestamp, content, status): + def insert_message(self, msg:AgentMsg): """ insert a new message into the Messages table """ try: + action_name = None + action_params = None + action_result = None + mentions = None + if msg.mentions: + mentions = json.dumps(msg.mentions) + + match msg.msg_type: + case AgentMsgType.TYPE_MSG: + pass + case AgentMsgType.TYPE_ACTION: + action_name = msg.func_name + action_params = json.dumps(msg.args) + action_result = msg.result_str + case AgentMsgType.TYPE_INTERNAL_CALL: + action_name = msg.func_name + action_params = json.dumps(msg.args) + action_result = msg.result_str + case AgentMsgType.TYPE_EVENT: + action_name = msg.event_name + action_params = json.dumps(msg.event_args) + + conn = self._get_conn() conn.execute(""" - INSERT INTO Messages (MessageID, SessionID, SenderID, ReceiverID, Timestamp, Content, Status) - VALUES (?, ?, ?, ?, ?, ?, ?) - """, (message_id, session_id, sender_id, receiver_id, timestamp, content, status)) + INSERT INTO Messages (MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status) + VALUES (?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (msg.msg_id, msg.session_id, msg.msg_type.value, msg.prev_msg_id, msg.sender, msg.target, msg.create_time, msg.topic,mentions,msg.body_mime,msg.body,action_name,action_params,action_result,msg.done_time,msg.status.value)) conn.commit() + + if msg.inner_call_chain: + for inner_call in msg.inner_call_chain: + self.insert_message(inner_call) + return 0 # return 0 if successful except Error as e: logging.error("Error occurred while inserting message: %s", e) @@ -134,7 +180,7 @@ class ChatSessionDB: """Get a message by its ID""" conn =self._get_conn() c = conn.cursor() - c.execute("SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages WHERE MessageID = ?", (message_id,)) + c.execute("SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages WHERE MessageID = ?", (message_id,)) message = c.fetchone() return message @@ -144,7 +190,7 @@ class ChatSessionDB: conn = self._get_conn() cursor = conn.cursor() cursor.execute(""" - SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages + SELECT MessageID, SessionID, MsgType, PrevMsgID, SenderID, ReceiverID, Timestamp, Topic,Mentions,ContentMIME,Content,ActionName,ActionParams,ActionResult,DoneTime,Status FROM Messages WHERE SessionID = ? ORDER BY Timestamp DESC LIMIT ? OFFSET ? @@ -222,29 +268,31 @@ class AIChatSession: result = [] for msg in msgs: agent_msg = AgentMsg() - agent_msg.id = msg[0] - agent_msg.sender = msg[2] - agent_msg.target = msg[3] - agent_msg.create_time = msg[4] - agent_msg.body = msg[5] - # agent_msg.state = msg[6] + agent_msg.msg_id = msg[0] + agent_msg.session_id = msg[1] + agent_msg.msg_type = AgentMsgType(msg[2]) + agent_msg.prev_msg_id = msg[3] + agent_msg.sender = msg[4] + agent_msg.target = msg[5] + agent_msg.create_time = msg[6] + agent_msg.topic = msg[7] + if msg[8] is not None: + agent_msg.mentions = json.loads(msg[8]) + agent_msg.body_mime = msg[9] + agent_msg.body = msg[10] + agent_msg.func_name = msg[11] + if msg[12] is not None: + agent_msg.args = json.loads(msg[12]) + agent_msg.result_str = msg[13] + agent_msg.done_time = msg[14] + agent_msg.status = AgentMsgStatus(msg[15]) result.append(agent_msg) return result def append(self,msg:AgentMsg) -> None: - self.db.insert_message(msg.id,self.session_id,msg.sender,msg.target,msg.create_time,msg.body,0) - - def append_post(self,msg:AgentMsg) -> None: - """append msg to session, msg is post from session (owner => msg.target)""" - #assert msg.sender == self.owner_id,"post message means msg.sender == self.owner_id" - self.append(msg) - - - def append_recv(self,msg:AgentMsg) -> None: - """append msg to session, msg is recv from msg'sender (msg.sender => owner)""" - #assert msg.target == self.owner_id,"recv message means msg.target == self.owner_id" - self.append(msg) + msg.session_id = self.session_id + self.db.insert_message(msg) #def attach_event_handler(self,handler) -> None: # """chat session changed event handler""" diff --git a/src/aios_kernel/environment.py b/src/aios_kernel/environment.py index 3a11d3b..49ea72b 100644 --- a/src/aios_kernel/environment.py +++ b/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: diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 1be47fd..3a665a3 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -54,11 +54,17 @@ class OpenAI_ComputeNode(ComputeNode): prompts = task.params["prompts"] logger.info(f"call openai {mode_name} prompts: {prompts}") - resp = openai.ChatCompletion.create(model=mode_name, + if task.params.get("inner_functions") is None: + 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? + temperature=0.7) + else: + resp = openai.ChatCompletion.create(model=mode_name, + messages=prompts, + functions=task.params["inner_functions"], + max_tokens=task.params["max_token_size"], + temperature=0.7) # TODO: add temperature to task params? logger.info(f"openai response: {resp}") diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 0a01c29..58d0538 100644 --- a/src/aios_kernel/workflow.py +++ b/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: @@ -158,11 +160,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,18 +179,50 @@ 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) if select_role_id is not None: @@ -195,27 +230,13 @@ class Workflow: if select_role is None: logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group") return None - - result = await self.role_process_msg(msg,select_role) - if result is None: - logger.error(f"_process_msg return None for :{msg}") - return - - final_result = result - 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 - + return await self.role_process_msg(msg,select_role,chatsesssion) + else: + logger.error(f"input_filter return None for :{msg.body}") + return None + + logger.error(f"{self.workflow_id}:no role can process this msg:{msg.body}") return final_result @classmethod @@ -276,52 +297,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 +359,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().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().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}] @@ -410,18 +520,10 @@ class Workflow: result_prompt.messages.append({"role":"user","content":f"{msg.body}"}) 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 +549,4 @@ class Workflow: # the_env.attach_event_handler(k,_env_msg_handler) # break - else: - logger.warn(f"environment {env.get_id()} already connected!") diff --git a/src/aios_kernel/workflow_env.py b/src/aios_kernel/workflow_env.py index 9ca2ee3..1107c8f 100644 --- a/src/aios_kernel/workflow_env.py +++ b/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,12 @@ class CalenderEnvironment(Environment): def stop(self): self.is_run = False - async def get_now(self) -> str: + def get_now(self,key)->str: + now = datetime.now() + formatted_time = now.strftime('%Y-%m-%d %H:%M:%S') + return formatted_time + + async def _get_now(self) -> str: now = datetime.now() formatted_time = now.strftime('%Y-%m-%d %H:%M:%S') return formatted_time @@ -71,6 +76,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'): diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 27f87e4..73d8c93 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -26,7 +26,7 @@ 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 +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession sys.path.append(directory + '/../../component/') from agent_manager import AgentManager @@ -73,6 +73,7 @@ class AIOS_Shell: open_ai_node.start() ComputeKernel().add_compute_node(open_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) return True @@ -83,7 +84,7 @@ class AIOS_Shell: 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: @@ -91,6 +92,9 @@ class AIOS_Shell: async def install_workflow(self,workflow_id:Workflow) -> None: pass + + async def _user_process_msg(self,msg:AgentMsg) -> AgentMsg: + pass async def call_func(self,func_name, args): match func_name: @@ -115,6 +119,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 @@ -134,7 +139,7 @@ class AIOS_Shell: 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) @@ -171,6 +176,7 @@ async def main(): format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s') shell = AIOS_Shell("user") await shell.initial() + print(f"aios shell {shell.get_version()} ready.") completer = WordCompleter(['send($target,$msg,$topic)', diff --git a/src/service/aios_shell/tg_aios_shell.py b/src/service/aios_shell/tg_aios_shell.py index 98844ae..6160ec8 100644 --- a/src/service/aios_shell/tg_aios_shell.py +++ b/src/service/aios_shell/tg_aios_shell.py @@ -14,7 +14,7 @@ from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandl directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession sys.path.append(directory + '/../../component/') from agent_manager import AgentManager From cbfff71d6bffc663f61d626879704066d358c7b6 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 15 Sep 2023 00:33:18 +0800 Subject: [PATCH 2/5] add whisper speech to text node and google text to speech node --- src/aios_kernel/__init__.py | 6 +- src/aios_kernel/google_text_to_speech_node.py | 104 ++++++++++++++++ src/aios_kernel/whisper_node.py | 111 ++++++++++++++++++ src/requirements.txt | 1 + 4 files changed, 220 insertions(+), 2 deletions(-) create mode 100644 src/aios_kernel/google_text_to_speech_node.py create mode 100644 src/aios_kernel/whisper_node.py diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index bef117f..2a87b5c 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -4,8 +4,10 @@ from .chatsession import AIChatSession from .agent import AIAgent,AIAgentTemplete,AgentPrompt from .compute_kernel import ComputeKernel,ComputeTask from .compute_node import ComputeNode,LocalComputeNode -from .open_ai_node import OpenAI_ComputeNode +from .open_ai_node import OpenAI_ComputeNode from .role import AIRole,AIRoleGroup from .workflow import Workflow from .bus import AIBus -from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent \ No newline at end of file +from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent +from .whisper_node import WhisperComputeNode +from .google_text_to_speech_node import GoogleTextToSpeechNode diff --git a/src/aios_kernel/google_text_to_speech_node.py b/src/aios_kernel/google_text_to_speech_node.py new file mode 100644 index 0000000..7709eef --- /dev/null +++ b/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 diff --git a/src/aios_kernel/whisper_node.py b/src/aios_kernel/whisper_node.py new file mode 100644 index 0000000..b036ad5 --- /dev/null +++ b/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 diff --git a/src/requirements.txt b/src/requirements.txt index 605dc72..7de369e 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -6,3 +6,4 @@ Pillow==10.0.0 prompt_toolkit==3.0.39 stability_sdk==0.8.4 toml==0.10.2 +google-cloud-texttospeech==2.14.1 From ef9ab83e63cb40249e7332a892fb6e28a4b32fa7 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Fri, 15 Sep 2023 17:35:12 -0700 Subject: [PATCH 3/5] Add tunnel support ,and implment TelegramTunnel. Now we can use telegram bot connect to installed agent/workflow. --- rootfs/tunnels.cfg.toml | 6 + src/aios_kernel/__init__.py | 4 +- src/aios_kernel/agent.py | 4 +- src/aios_kernel/contact_manager.py | 34 ++++ src/aios_kernel/tg_tunnel.py | 116 ++++++++++++ src/aios_kernel/tunnel.py | 63 +++++++ src/service/aios_shell/aios_shell.py | 12 +- src/service/aios_shell/tg_aios_shell.py | 223 ------------------------ 8 files changed, 234 insertions(+), 228 deletions(-) create mode 100644 rootfs/tunnels.cfg.toml create mode 100644 src/aios_kernel/contact_manager.py create mode 100644 src/aios_kernel/tg_tunnel.py create mode 100644 src/aios_kernel/tunnel.py delete mode 100644 src/service/aios_shell/tg_aios_shell.py diff --git a/rootfs/tunnels.cfg.toml b/rootfs/tunnels.cfg.toml new file mode 100644 index 0000000..c89e504 --- /dev/null +++ b/rootfs/tunnels.cfg.toml @@ -0,0 +1,6 @@ +[[tunnels]] +tunnel_id = "MyRoobot" +type="TelegramTunnel" +target="agent_1" +token="" + diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index fb09d88..0d582df 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -8,4 +8,6 @@ from .open_ai_node import OpenAI_ComputeNode from .role import AIRole,AIRoleGroup from .workflow import Workflow from .bus import AIBus -from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent \ No newline at end of file +from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent +from .tunnel import AgentTunnel +from .tg_tunnel import TelegramTunnel \ No newline at end of file diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 0c3e106..fd25af2 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -243,8 +243,8 @@ class AIAgent: 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 diff --git a/src/aios_kernel/contact_manager.py b/src/aios_kernel/contact_manager.py new file mode 100644 index 0000000..cdc050a --- /dev/null +++ b/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) \ No newline at end of file diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py new file mode 100644 index 0000000..efa23fc --- /dev/null +++ b/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) + + + \ No newline at end of file diff --git a/src/aios_kernel/tunnel.py b/src/aios_kernel/tunnel.py new file mode 100644 index 0000000..bbd18d3 --- /dev/null +++ b/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 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 73d8c93..b6364c1 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/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 @@ -26,12 +27,12 @@ shell_style = Style.from_dict({ directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment sys.path.append(directory + '/../../component/') from agent_manager import AgentManager from workflow_manager import WorkflowManager -from aios_kernel import CalenderEnvironment,Environment + class AIOS_Shell: @@ -74,6 +75,13 @@ class AIOS_Shell: ComputeKernel().add_compute_node(open_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() + + tunnels_config_path = os.path.abspath(directory + "/../../../rootfs/tunnels.cfg.toml") + tunnel_config = toml.load(tunnels_config_path) + await AgentTunnel.load_all_tunnels_from_config(tunnel_config["tunnels"]) + return True diff --git a/src/service/aios_shell/tg_aios_shell.py b/src/service/aios_shell/tg_aios_shell.py deleted file mode 100644 index 6160ec8..0000000 --- a/src/service/aios_shell/tg_aios_shell.py +++ /dev/null @@ -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,AgentMsgStatus,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() \ No newline at end of file From 43c70489e791230ce35733dae6294c54021f8ecb Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sat, 16 Sep 2023 11:41:59 -0700 Subject: [PATCH 4/5] 1) Add e-mail tunnel support 2) Modify all __new__ to get_instance() 3) init docker build scripts --- .dockerignore | 27 ++++ Dockerfile | 10 ++ rootfs/tunnels.cfg.toml | 23 ++- src/aios_kernel/__init__.py | 3 +- src/aios_kernel/agent.py | 6 +- src/aios_kernel/compute_kernel.py | 15 +- src/aios_kernel/email_tunnel.py | 143 ++++++++++++++++++ src/aios_kernel/open_ai_node.py | 17 ++- src/aios_kernel/stability_node.py | 22 +-- src/aios_kernel/workflow.py | 10 +- src/component/agent_manager/agent_manager.py | 6 +- src/component/package_manager/env.py | 9 +- .../workflow_manager/workflow_manager.py | 5 +- src/requirements.txt | 13 +- src/service/aios_shell/aios_shell.py | 33 ++-- 15 files changed, 272 insertions(+), 70 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 src/aios_kernel/email_tunnel.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..58b3427 --- /dev/null +++ b/.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 diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..65296e4 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.11 +WORKDIR /opt/aios +COPY ./src /opt/aios +COPY ./rootfs /var/aios +RUN pip install --no-cache-dir -r /opt/opendan/requirements.txt + +ENV PYTHONDONTWRITEBYTECODE 1 +ENV PYTHONUNBUFFERED 1 + +CMD ["python3","./service/aios_shell/aios_shell.py"] \ No newline at end of file diff --git a/rootfs/tunnels.cfg.toml b/rootfs/tunnels.cfg.toml index c89e504..07682fc 100644 --- a/rootfs/tunnels.cfg.toml +++ b/rootfs/tunnels.cfg.toml @@ -1,6 +1,19 @@ -[[tunnels]] -tunnel_id = "MyRoobot" -type="TelegramTunnel" -target="agent_1" -token="" +#[[tunnels]] +#tunnel_id = "MyRoobot" +#type="TelegramTunnel" +#target="agent_1" +#token="" + +#[[tunnels]] +#tunnel_id="MyEmailRobot" +#type="EmailTunnel" +#target="agent_1" + +#email="your_email@msn.com" +#imap="outlook.office365.com:993" +#smtp="outlook.office365.com:587" +#user="" +#password="" +#folder="inbox" +#interval=10 \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 0d582df..ddf63b5 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -10,4 +10,5 @@ from .workflow import Workflow from .bus import AIBus from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent from .tunnel import AgentTunnel -from .tg_tunnel import TelegramTunnel \ No newline at end of file +from .tg_tunnel import TelegramTunnel +from .email_tunnel import EmailTunnel \ No newline at end of file diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index fd25af2..52dd2df 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -82,7 +82,7 @@ class AIAgent: self.chat_db = None self.unread_msg = Queue() # msg from other agent self.owner_env : Environment = None - self.owenr_bus : AIBus = None + self.owenr_bus = None @classmethod def create_from_templete(cls,templete:AIAgentTemplete, fullname:str): @@ -164,7 +164,7 @@ class AIAgent: inner_functions = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - 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) ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() @@ -198,7 +198,7 @@ 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") diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 5fea958..b69b879 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/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 = {} diff --git a/src/aios_kernel/email_tunnel.py b/src/aios_kernel/email_tunnel.py new file mode 100644 index 0000000..c4f75b4 --- /dev/null +++ b/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}") diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 3a665a3..2e87710 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -13,20 +13,17 @@ 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 + 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.node_id = "openai_node" @@ -89,6 +86,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() diff --git a/src/aios_kernel/stability_node.py b/src/aios_kernel/stability_node.py index 8d40878..1efc7c8 100644 --- a/src/aios_kernel/stability_node.py +++ b/src/aios_kernel/stability_node.py @@ -16,20 +16,17 @@ 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 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 +85,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 +105,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...") diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 58d0538..46b7507 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -122,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 @@ -391,7 +395,7 @@ class Workflow: inner_functions = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt, + 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) @@ -431,7 +435,7 @@ class Workflow: 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(),inner_functions) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size(),inner_functions) result_str = task_result.result_str logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}") diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index 9b05b62..7c27215 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -11,12 +11,12 @@ logger = logging.getLogger(__name__) class AgentManager: _instance = None - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(AgentManager, cls).__new__(cls) + cls._instance = AgentManager() return cls._instance - 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") diff --git a/src/component/package_manager/env.py b/src/component/package_manager/env.py index 0ef9cfe..4e1b68a 100644 --- a/src/component/package_manager/env.py +++ b/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] diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index 5603745..daeee13 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -9,9 +9,10 @@ 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 diff --git a/src/requirements.txt b/src/requirements.txt index 605dc72..d917717 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -1,8 +1,13 @@ 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 prompt_toolkit==3.0.39 +python-telegram-bot==20.5 +Requests==2.31.0 stability_sdk==0.8.4 toml==0.10.2 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index b6364c1..b182e48 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -18,22 +18,24 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory from prompt_toolkit.completion import WordCompleter from prompt_toolkit.styles import Style -shell_style = Style.from_dict({ - 'title': '#87d7ff bold', #RGB - 'content': '#007f00 bold', - 'prompt': '#00FF00', -}) + directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel sys.path.append(directory + '/../../component/') from agent_manager import AgentManager from workflow_manager import WorkflowManager +shell_style = Style.from_dict({ + 'title': '#87d7ff bold', #RGB + 'content': '#007f00 bold', + 'prompt': '#00FF00', +}) + class AIOS_Shell: def __init__(self,username:str) -> None: @@ -43,13 +45,13 @@ class AIOS_Shell: 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 @@ -57,7 +59,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: @@ -68,15 +70,16 @@ 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/")) - open_ai_node = OpenAI_ComputeNode() + AgentManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/")) + WorkflowManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/workflows/")) + open_ai_node = OpenAI_ComputeNode.get_instance() open_ai_node.start() - ComputeKernel().add_compute_node(open_ai_node) + ComputeKernel.get_instance().add_compute_node(open_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() tunnels_config_path = os.path.abspath(directory + "/../../../rootfs/tunnels.cfg.toml") tunnel_config = toml.load(tunnels_config_path) @@ -140,9 +143,9 @@ 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) From a20f331e8fc9f337d682573838191124e0a3e14f Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sun, 17 Sep 2023 18:18:54 -0700 Subject: [PATCH 5/5] =?UTF-8?q?=EF=BB=BF0)=20Complete=20Docker=20build=20s?= =?UTF-8?q?criptes.=201)=20Refactor=20storage=20dir=20struct=20for=20relea?= =?UTF-8?q?se.=20=202)=20Add=20a=20common=20'user=5Fconfig'=20model.=203)M?= =?UTF-8?q?ore=20friendly=20CLI=20experience?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Dockerfile | 9 +- build.sh | 0 build_all_in_one.sh | 5 + rootfs/tunnels.cfg.toml | 32 ++-- src/aios_kernel/__init__.py | 3 + src/aios_kernel/open_ai_node.py | 25 ++- src/aios_kernel/stability_node.py | 7 +- src/aios_kernel/storage.py | 171 ++++++++++++++++++ src/component/agent_manager/agent_manager.py | 15 +- .../workflow_manager/workflow_manager.py | 14 +- src/requirements.txt | 1 + src/service/aios_shell/aios_shell.py | 142 ++++++++++++--- test/daemon_test.py | 17 ++ 13 files changed, 382 insertions(+), 59 deletions(-) delete mode 100644 build.sh create mode 100644 build_all_in_one.sh create mode 100644 src/aios_kernel/storage.py create mode 100644 test/daemon_test.py diff --git a/Dockerfile b/Dockerfile index 65296e4..2045f08 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,8 +1,13 @@ FROM python:3.11 WORKDIR /opt/aios COPY ./src /opt/aios -COPY ./rootfs /var/aios -RUN pip install --no-cache-dir -r /opt/opendan/requirements.txt +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 diff --git a/build.sh b/build.sh deleted file mode 100644 index e69de29..0000000 diff --git a/build_all_in_one.sh b/build_all_in_one.sh new file mode 100644 index 0000000..d1eb62b --- /dev/null +++ b/build_all_in_one.sh @@ -0,0 +1,5 @@ +#!/usr/bin/bash + +pipreqs ./src --force +# Build the docker image +docker build -t aios . diff --git a/rootfs/tunnels.cfg.toml b/rootfs/tunnels.cfg.toml index 07682fc..32975de 100644 --- a/rootfs/tunnels.cfg.toml +++ b/rootfs/tunnels.cfg.toml @@ -1,19 +1,19 @@ -#[[tunnels]] -#tunnel_id = "MyRoobot" -#type="TelegramTunnel" -#target="agent_1" +[[tunnels]] +tunnel_id = "MyRoobot" +type="TelegramTunnel" +target="agent_1" +token="your_token" -#token="" -#[[tunnels]] -#tunnel_id="MyEmailRobot" -#type="EmailTunnel" -#target="agent_1" +[[tunnels]] +tunnel_id="MyEmailRobot" +type="EmailTunnel" +target="agent_1" -#email="your_email@msn.com" -#imap="outlook.office365.com:993" -#smtp="outlook.office365.com:587" -#user="" -#password="" -#folder="inbox" -#interval=10 \ No newline at end of file +email="youremail@msn.com" +imap="outlook.office365.com:993" +smtp="outlook.office365.com:587" +user="" +password="" +folder="inbox" +interval=10 \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 6daa1b1..9a2b699 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -14,3 +14,6 @@ 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" diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 2e87710..029eaef 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -7,6 +7,7 @@ import logging from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType from .compute_node import ComputeNode +from .storage import AIStorage,UserConfig logger = logging.getLogger(__name__) @@ -19,23 +20,35 @@ class OpenAI_ComputeNode(ComputeNode): 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__() 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()}") diff --git a/src/aios_kernel/stability_node.py b/src/aios_kernel/stability_node.py index 1efc7c8..b77839e 100644 --- a/src/aios_kernel/stability_node.py +++ b/src/aios_kernel/stability_node.py @@ -10,6 +10,7 @@ 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__) @@ -21,7 +22,11 @@ class Stability_ComputeNode(ComputeNode): if cls._instance is None: cls._instance = Stability_ComputeNode() return cls._instance - + + @classmethod + def declare_user_config(cls): + user_config = AIStorage.get_instance().get_user_config() + user_config.add_user_config("stability_api_key",False,None,"STABILITY_API_KEY") def __init__(self) -> None: super().__init__() diff --git a/src/aios_kernel/storage.py b/src/aios_kernel/storage.py new file mode 100644 index 0000000..64a2c0e --- /dev/null +++ b/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 + diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index 7c27215..5d20cf6 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/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__) @@ -12,15 +12,18 @@ class AgentManager: _instance = None @classmethod - def get_instance(cls): + def get_instance(cls)->'AgentManager': if cls._instance is None: cls._instance = AgentManager() return cls._instance - 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" + def initial(self) -> None: + system_app_dir = AIStorage.get_instance().get_system_app_dir() + user_data_dir = AIStorage.get_instance().get_myai_dir() + + 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") diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index daeee13..d3d49fe 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -1,10 +1,11 @@ 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 @@ -16,10 +17,13 @@ class 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") diff --git a/src/requirements.txt b/src/requirements.txt index ac6723c..643f956 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -6,6 +6,7 @@ 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 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index b182e48..e96415c 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -18,18 +18,17 @@ 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 Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel +from aios_kernel import AIOS_Version,UserConfigItem,AIStorage,Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel 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', @@ -42,6 +41,15 @@ class AIOS_Shell: 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] @@ -70,10 +78,14 @@ class AIOS_Shell: cal_env.start() Environment.set_env_by_id("calender",cal_env) - AgentManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/")) - WorkflowManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/workflows/")) + AgentManager.get_instance().initial() + WorkflowManager.get_instance().initial() + open_ai_node = OpenAI_ComputeNode.get_instance() - open_ai_node.start() + 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) 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) @@ -81,15 +93,23 @@ class AIOS_Shell: TelegramTunnel.register_to_loader() EmailTunnel.register_to_loader() - tunnels_config_path = os.path.abspath(directory + "/../../../rootfs/tunnels.cfg.toml") - tunnel_config = toml.load(tunnels_config_path) - await AgentTunnel.load_all_tunnels_from_config(tunnel_config["tunnels"]) + 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() @@ -101,9 +121,6 @@ class AIOS_Shell: else: return "error!" - async def install_workflow(self,workflow_id:Workflow) -> None: - pass - async def _user_process_msg(self,msg:AgentMsg) -> AgentMsg: pass @@ -161,8 +178,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): @@ -178,22 +195,103 @@ def parse_function_call(func_string): params = None 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() + + 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) @@ -217,8 +315,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()) diff --git a/test/daemon_test.py b/test/daemon_test.py new file mode 100644 index 0000000..b4eff77 --- /dev/null +++ b/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()