diff --git a/rootfs/agents/JarvisPlus/agent.toml b/rootfs/agents/JarvisPlus/agent.toml index 19c621c..dc249a2 100644 --- a/rootfs/agents/JarvisPlus/agent.toml +++ b/rootfs/agents/JarvisPlus/agent.toml @@ -13,7 +13,7 @@ My name is JarvisPlus, I am the master's super personal assistant. I think hard The types of TODO I can handle include: - Scheduling, where I will try to contact the relevant personnel of the plan and confirm the details of the schedule with them. - Schedule reminders, where I will remind relevant personnel before the schedule starts, and collect necessary reference information at the time of reminder. -- Using the post_msg function to contact relevant personnel. +- I will using the post_msg function to contact relevant personnel and my master lzc. - Writing documents/letters, using op:'create' to save my work results. I receive a TODO described in json format, I will handle it according to the following rules: @@ -25,7 +25,7 @@ I receive a TODO described in json format, I will handle it according to the fol The result of my planned execution must be directly parsed by `python json.loads`. Here is an example: { - resp: 'My Plan is ...', + resp: '$what_did_I_do', post_msg : [ { target:'$target_name', @@ -46,12 +46,12 @@ The result of my planned execution must be directly parsed by `python json.loads { op: 'update_todo', id: '$todo_id', - state: 'doing' # pending,doing,done,cancel,failed,expired + state: 'cancel' # pending,cancel }, { op: 'write_file', - path: '/todos/$todo_path/.result/doc_name', - content:'doc content' + path: '/todos/$todo_path/.result/$doc_name', + content:'$doc_content' } ] } diff --git a/rootfs/agents/Thinker/agent.toml b/rootfs/agents/Thinker/agent.toml index b4016d5..05332f6 100644 --- a/rootfs/agents/Thinker/agent.toml +++ b/rootfs/agents/Thinker/agent.toml @@ -1,7 +1,7 @@ instance_id = "Thinker" fullname = "Thinker" -llm_model_name = "gpt-3.5-turbo-16k-0613" -max_token_size = 14000 +llm_model_name = "Llama-2-13b-chat" +max_token_size = 2000 #enable_function =["add_event"] #enable_kb = "true" diff --git a/rootfs/agents/Tracy/agent.toml b/rootfs/agents/Tracy/agent.toml index dfd3cd9..8d44476 100644 --- a/rootfs/agents/Tracy/agent.toml +++ b/rootfs/agents/Tracy/agent.toml @@ -1,6 +1,5 @@ instance_id = "Tracy" fullname = "Tracy" - [[prompt]] role = "system" content = """ diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 89797fc..2b1b8ef 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -24,5 +24,5 @@ from .stability_node import Stability_ComputeNode from .local_st_compute_node import LocalSentenceTransformer_Text_ComputeNode,LocalSentenceTransformer_Image_ComputeNode from .compute_node_config import ComputeNodeConfig from .ai_function import SimpleAIFunction -AIOS_Version = "0.5.1, build 2023-9-28" +AIOS_Version = "0.5.2, build 2023-11-1" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index cf3ae75..9e2ca15 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -12,7 +12,7 @@ import datetime import copy import sys -from .agent_base import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult,AgentPrompt,AgentReport,AgentTodo,AgentGoal,AgentTodoResult,AgentWorkLog +from .agent_base import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult,AgentPrompt,AgentReport,AgentTodo,AgentTodoResult,AgentWorkLog from .chatsession import AIChatSession from .compute_task import ComputeTaskResult,ComputeTaskResultCode from .ai_function import AIFunction @@ -100,6 +100,8 @@ class AIAgent: self.agent_energy = 15 self.agent_task = None self.last_recover_time = time.time() + self.enable_thread = False + self.can_do_unassigned_task = True self.agent_id:str = None @@ -158,6 +160,9 @@ class AIAgent: return False self.fullname = config["fullname"] + if config.get("enable_thread") is not None: + self.enable_thread = bool(config["enable_thread"]) + if config.get("prompt") is not None: self.agent_prompt = AgentPrompt() self.agent_prompt.load_from_config(config["prompt"]) @@ -304,7 +309,7 @@ class AIAgent: prompt.messages.append({"role":"function","content":result_str,"name":func_name}) task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") + logger.error(f"_execute_func llm compute error:{task_result.error_str}") return task_result ineternal_call_record.result_str = task_result.result_str @@ -425,6 +430,9 @@ class AIAgent: def need_session_summmary(self,msg:AgentMsg,session:AIChatSession) -> bool: return False + + async def _create_openai_thread(self) -> str: + return None async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: msg_prompt = AgentPrompt() @@ -447,6 +455,19 @@ class AIAgent: msg_prompt.messages = [{"role":"user","content":msg.body}] session_topic = msg.get_sender() + "#" + msg.topic chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) + if self.enable_thread: + need_create_thread = False + if chatsession.openai_thread_id is not None: + if len(chatsession.openai_thread_id) < 1: + need_create_thread = True + else: + need_create_thread = True + + if need_create_thread: + openai_thread_id = await self._create_openai_thread() + if openai_thread_id is not None: + chatsession.update_openai_thread_id(openai_thread_id) + workspace = self.get_workspace_by_msg(msg) @@ -544,6 +565,7 @@ class AIAgent: async def _get_history_prompt_for_think(self,chatsession:AIChatSession,summary:str,system_token_len:int,pos:int)->(AgentPrompt,int): + history_len = (self.max_token_size * 0.7) - system_token_len messages = chatsession.read_history(self.history_len,pos,"natural") # read @@ -663,43 +685,65 @@ class AIAgent: workspace : WorkspaceEnvironment = self.get_workspace_by_msg(None) logger.info(f"agent {self.agent_id} do my work start!") - # review todo能更整体的思考一次todo的优先级 - if await self.need_review_todos(): - await self._llm_review_todos(workspace) + # review todolist + #if await self.need_review_todolist(): + # await self._llm_review_todolist(workspace) todo_list = await workspace.get_todo_list(self.agent_id) + check_count = 0 + do_count = 0 for todo in todo_list: if self.agent_energy <= 0: break - - if await self.can_do(todo,workspace) is False: - continue - if todo.retry_count < 2: - need_think_todo_from_goal = False - await self._llm_do(todo,workspace) + if await self.need_review_todo(todo,workspace): + review_result = await self._llm_review_todo(todo,workspace) + todo.last_review_time = datetime.datetime.now().timestamp() + + elif await self.can_check(todo,workspace): + check_result : AgentTodoResult = await self._llm_check_todo(todo,workspace) + todo.last_check_time = datetime.datetime.now().timestamp() + + match check_result.result_code: + case AgentTodoResult.TODO_RESULT_CODE_LLM_ERROR: + continue + case AgentTodoResult.TODO_RESULT_CODE_OK: + await workspace.update_todo(todo.todo_id,AgentTodo.TODO_STATE_DONE) + case AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR: + await workspace.update_todo(todo.todo_id,AgentTodo.TDDO_STATE_CHECKFAILED) + + await workspace.append_worklog(todo,check_result) self.agent_energy -= 1 - if todo.state == "done": - await self._llm_check_todo(todo,workspace) - self.agent_energy -= 1 - - logger.info(f"agent {self.agent_id} do my work done!") + check_count += 1 + elif await self.can_do(todo,workspace): + do_result : AgentTodoResult = await self._llm_do(todo,workspace) + todo.last_do_time = datetime.datetime.now().timestamp() + todo.retry_count += 1 - def get_review_todo_prompt(self) -> AgentPrompt: + match do_result.result_code: + case AgentTodoResult.TODO_RESULT_CODE_LLM_ERROR: + continue + case AgentTodoResult.TODO_RESULT_CODE_OK: + await workspace.update_todo(todo.todo_id,AgentTodo.TODO_STATE_WAITING_CHECK) + case AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR: + await workspace.update_todo(todo.todo_id,AgentTodo.TODO_STATE_EXEC_FAILED) + + await workspace.append_worklog(todo,do_result) + self.agent_energy -= 2 + do_count += 1 + + logger.info(f"agent {self.agent_id} ,check:{check_count} todo,do:{do_count} todo.") + + def get_review_todo_prompt(self,todo:AgentTodo) -> AgentPrompt: return self.review_todo_prompt - async def need_review_todos(self) -> bool: - if self.get_review_todo_prompt() is None: - return False - return True - - async def _llm_review_todos(self,workspace:WorkspaceEnvironment): + async def _llm_review_todo(self,todo:AgentTodo,workspace:WorkspaceEnvironment): prompt = AgentPrompt() prompt.append(workspace.get_prompt()) prompt.append(workspace.get_role_prompt(self.agent_id)) - prompt.append(self.get_review_todo_prompt()) + prompt.append(self.get_review_todo_prompt(todo)) todo_tree = workspace.get_todo_tree("/") prompt.append(AgentPrompt(todo_tree)) @@ -712,49 +756,105 @@ class AIAgent: return - def get_do_prompt(self,todo_type:str=None) -> AgentPrompt: + def get_do_prompt(self,todo:AgentTodo) -> AgentPrompt: return self.do_prompt def get_prompt_from_todo(self,todo:AgentTodo) -> AgentPrompt: json_str = json.dumps(todo.raw_obj) return AgentPrompt(json_str) + + async def need_review_todo(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: + return False + + async def can_check(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: + if self.get_check_prompt(todo) is None: + return False + + if todo.can_check() is False: + return False + + if todo.checker is not None: + if todo.checker != self.agent_id: + return False + else: + if self.can_do_unassigned_task is False: + return False + else: + todo.checker = self.agent_id + + return True async def can_do(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: - return todo.can_do() + if todo.can_do() is False: + return False + + if todo.worker is not None: + if todo.worker != self.agent_id: + return False + else: + if self.can_do_unassigned_task is False: + return False + else: + todo.worker = self.agent_id + + return True async def _llm_do(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> AgentTodoResult: + result = AgentTodoResult() prompt : AgentPrompt = AgentPrompt() #prompt.append(self.agent_prompt) prompt.append(workspace.get_role_prompt(self.agent_id)) - do_prompt = workspace.get_do_prompt() + do_prompt = workspace.get_do_prompt(todo) if do_prompt is None: - do_prompt = self.get_do_prompt() + do_prompt = self.get_do_prompt(todo) prompt.append(do_prompt) - # 有通用的todo执行方法，也有定制的，针对特定类型TODO更高效的执行方法 - # 根据经验，Agent可以自主掌握/整理更多类型的TODO的执行方法 + # There are general methods for executing todos, as well as customized ones that are more efficient for specific types of TODOS. + # Based on experience, an Agent can autonomously master/organize execution methods for a greater variety of TODO types. - #prompt.append(do_log_prompt) + #prompt.append(work_log_prompt) prompt.append(self.get_prompt_from_todo(todo)) task_result:ComputeTaskResult = await self._do_llm_complection(prompt) if task_result.error_str is not None: logger.error(f"_llm_do compute error:{task_result.error_str}") + result.result_code = AgentTodoResult.TODO_RESULT_CODE_LLM_ERROR + result.error_str = task_result.error_str + return result + llm_result = LLMResult.from_str(task_result.result_str) + # result_str is the explain of how to do this todo + result.result_str = llm_result.resp + result.op_list = llm_result.op_list + if llm_result.post_msgs is not None: + for msg in llm_result.post_msgs: + msg.sender = self.agent_id + msg.topic = f"{todo.title}##{todo.todo_id}" + #msg.prev_msg_id = todo.todo_id + chatsession = AIChatSession.get_session(self.agent_id,f"{msg.target}#{msg.topic}",self.chat_db) + chatsession.append(msg) + resp = await AIBus.get_default_bus().post_message(msg) + logging.info(f"agent {self.agent_id} send msg to {msg.target} result:{resp}") - await workspace.exec_op_list(llm_result.op_list,self.agent_id) - await workspace.append_do_result(self.agent_id,llm_result) - - return task_result + op_errors,have_error = await workspace.exec_op_list(llm_result.op_list,self.agent_id) + if have_error: + result.result_code = AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR + #result.error_str = error_str + return result + + return result - def get_check_prompt(self) -> AgentPrompt: + async def append_toddo_result(self,todo,worksapce,llm_result,result_str): + pass + + def get_check_prompt(self,todo:AgentTodo) -> AgentPrompt: return self.check_prompt - async def _llm_check_todo(self, todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: + async def _llm_check_todo(self, todo:AgentTodo,workspace:WorkspaceEnvironment) : if self.get_check_prompt(todo) is None: - return True + return None prompt : AgentPrompt = AgentPrompt() prompt.append(self.agent_prompt) @@ -920,7 +1020,7 @@ class AIAgent: #3) llm summarize chat history task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,None) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") + logger.error(f"think_chatsession llm compute error:{task_result.error_str}") break else: new_summary= task_result.result_str @@ -930,6 +1030,8 @@ class AIAgent: async def get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len) -> AgentPrompt: # TODO: get prompt from group chat is different from single chat + if self.enable_thread: + return None history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len messages = chatsession.read_history(self.history_len) # read @@ -971,7 +1073,7 @@ class AIAgent: #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") + logger.error(f"_do_llm_complection llm compute error:{task_result.error_str}") #error_resp = msg.create_error_resp(task_result.error_str) return task_result @@ -987,7 +1089,15 @@ class AIAgent: return task_result def need_work(self) -> bool: - return True + if self.do_prompt is not None: + return True + if self.check_prompt is not None: + return True + + if self.agent_energy > 2: + return True + + return False def need_self_think(self) -> bool: return False @@ -1014,28 +1124,25 @@ class AIAgent: self.agent_energy += (now - self.last_recover_time) / 60 self.last_recover_time = now - if self.agent_energy <= 1: continue - # complete todo + # complete & check todo if self.need_work(): await self.do_my_work() # review other's todo # self.review_other_works() - # do work summary if self.need_self_think(): await self.do_self_think() - # if self.need_self_learn(): await self.do_self_learn() except Exception as e: tb_str = traceback.format_exc() - logger.error(f"agent {self.agent_id} on timer error:{tb_str},{e}") + logger.error(f"agent {self.agent_id} on timer error:{e},{tb_str}") continue diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index e1e2c6e..43575d2 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -258,8 +258,16 @@ class LLMResult: r.state = llm_json.get("state") r.resp = llm_json.get("resp") - r.post_msgs = llm_json.get("post_msgs") - r.send_msgs = llm_json.get("send_msgs") + post_msgs = llm_json.get("post_msg") + r.post_msgs = [] + if post_msgs: + for msg in post_msgs: + new_msg = AgentMsg() + target_id = msg.get("target") + msg_content = msg.get("content") + new_msg.set("",target_id,msg_content) + r.post_msgs.append(new_msg) + #new_msg.msg_type = AgentMsgType.TYPE_MSG r.calls = llm_json.get("calls") r.post_calls = llm_json.get("post_calls") @@ -352,23 +360,44 @@ class LLMResult: return r -class AgentGoal: - def __init__(self) -> None: - self.description = None - - class AgentReport: def __init__(self): pass class AgentTodoResult: + TODO_RESULT_CODE_OK = 0, + TODO_RESULT_CODE_LLM_ERROR = 1, + TODO_RESULT_CODE_EXEC_OP_ERROR = 2 + + def __init__(self) -> None: - self.result_state = "error" + self.result_code = AgentTodoResult.TODO_RESULT_CODE_OK + self.result_str = None + self.error_str = None + self.op_list = None + + def to_dict(self) -> dict: + result = {} + result["result_code"] = self.result_code + result["result_str"] = self.result_str + result["error_str"] = self.error_str + result["op_list"] = self.op_list + return result class AgentTodo: + TODO_STATE_WAIT_ASSIGN = "wait_assign" + TODO_STATE_INIT = "init" + TODO_STATE_PENDING = "pending" + TODO_STATE_WAITING_CHECK = "wait_check" + TODO_STATE_EXEC_FAILED = "exec_failed" + TDDO_STATE_CHECKFAILED = "check_failed" + TODO_STATE_CASNCEL = "cancel" + TODO_STATE_DONE = "done" + TODO_STATE_EXPIRED = "expired" + def __init__(self): self.todo_id = "todo#" + uuid.uuid4().hex self.title = None @@ -376,20 +405,23 @@ class AgentTodo: self.todo_path = None # get parent todo,sub todo by path #self.parent = None self.create_time = time.time() - self.due_date = time.time() + 3600 * 24 * 2 - self.state = "pending" - - self.depend_todo_ids = [] - self.sub_todos = {} - - self.need_check = True - self.result : ComputeTaskResult = None - self.last_check_result = None + self.state = "wait_assign" self.worker = None self.checker = None self.createor = None + self.need_check = True + self.due_date = time.time() + 3600 * 24 * 2 + self.last_do_time = None + self.last_check_time = None + self.last_review_time = None + + self.depend_todo_ids = [] + self.sub_todos = {} + + self.result : AgentTodoResult = None + self.last_check_result = None self.retry_count = 0 self.raw_obj = None @@ -400,14 +432,26 @@ class AgentTodo: todo.todo_id = json_obj.get("id") todo.title = json_obj.get("title") + todo.state = json_obj.get("state") create_time = json_obj.get("create_time") if create_time: todo.create_time = datetime.fromisoformat(create_time).timestamp() + todo.detail = json_obj.get("detail") due_date = json_obj.get("due_date") if due_date: todo.due_date = datetime.fromisoformat(due_date).timestamp() - todo.todo_path = json_obj.get("todo_path") + + last_do_time = json_obj.get("last_do_time") + if last_do_time: + todo.last_do_time = datetime.fromisoformat(last_do_time).timestamp() + last_check_time = json_obj.get("last_check_time") + if last_check_time: + todo.last_check_time = datetime.fromisoformat(last_check_time).timestamp() + last_review_time = json_obj.get("last_review_time") + if last_review_time: + todo.last_review_time = datetime.fromisoformat(last_review_time).timestamp() + todo.depend_todo_ids = json_obj.get("depend_todo_ids") todo.need_check = json_obj.get("need_check") #todo.result = json_obj.get("result") @@ -423,14 +467,21 @@ class AgentTodo: return todo def to_dict(self) -> dict: - result = {} + if self.raw_obj: + result = self.raw_obj + else: + result = {} + result["id"] = self.todo_id - result["todo_path"] = self.todo_path #result["parent_id"] = self.parent_id result["title"] = self.title + result["state"] = self.state result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() result["detail"] = self.detail result["due_date"] = datetime.fromtimestamp(self.due_date).isoformat() + result["last_do_time"] = datetime.fromtimestamp(self.last_do_time).isoformat() if self.last_do_time else None + result["last_check_time"] = datetime.fromtimestamp(self.last_check_time).isoformat() if self.last_check_time else None + result["last_review_time"] = datetime.fromtimestamp(self.last_review_time).isoformat() if self.last_review_time else None result["depend_todo_ids"] = self.depend_todo_ids result["need_check"] = self.need_check result["worker"] = self.worker @@ -439,19 +490,57 @@ class AgentTodo: result["retry_count"] = self.retry_count return result - def can_do(self) -> bool: - for sub_todo in self.sub_todos.values(): - if sub_todo.state: - if sub_todo.state != "done": - return False + + def can_check(self)->bool: + if self.state != AgentTodo.TODO_STATE_WAITING_CHECK: + return False now = datetime.now().timestamp() - time_diff = now - self.due_date - - if time_diff > 7*24*3600: - return False + if self.last_check_time: + time_diff = now - self.last_check_time + if time_diff < 60*15: + logger.info(f"todo {self.title} is already checked, ignore") + return False + return True + def can_do(self) -> bool: + match self.state: + case AgentTodo.TODO_STATE_DONE: + logger.info(f"todo {self.title} is done, ignore") + return False + case AgentTodo.TODO_STATE_CASNCEL: + logger.info(f"todo {self.title} is cancel, ignore") + return False + case AgentTodo.TODO_STATE_EXPIRED: + logger.info(f"todo {self.title} is expired, ignore") + return False + case AgentTodo.TODO_STATE_EXEC_FAILED: + if self.retry_count > 3: + logger.info(f"todo {self.title} retry count ({self.retry_count}) is too many, ignore") + return False + + now = datetime.now().timestamp() + time_diff = self.due_date - now + if time_diff < 0: + logger.info(f"todo {self.title} is expired, ignore") + self.state = AgentTodo.TODO_STATE_EXPIRED + return False + + if time_diff > 7*24*3600: + logger.info(f"todo {self.title} is far before due date, ignore") + return False + + if self.last_do_time: + time_diff = now - self.last_do_time + if time_diff < 60*15: + logger.info(f"todo {self.title} is already do ignore") + return False + + logger.info(f"todo {self.title} can do.") + return True + + class AgentWorkLog: def __init__(self) -> None: pass diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index ecbb676..94f789e 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -109,6 +109,9 @@ class AIBus: # means sub def register_message_handler(self,handler_name:str,handler:Any) -> Queue: handler_node = AIBusHandler(handler,self) + if self.handlers.get(handler_name) is not None: + logger.warn(f"handler {handler_name} already register on AI_BUS!") + self.handlers[handler_name] = handler_node return handler_node.queue diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index e5edcfc..9d704c8 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -52,7 +52,8 @@ class ChatSessionDB: SessionTopic TEXT, StartTime TEXT, SummarizePos INTEGER, - Summary TEXT + Summary TEXT, + ThreadID TEXT ); """) @@ -89,14 +90,14 @@ class ChatSessionDB: except Error as e: logging.error("Error occurred while creating tables: %s", e) - def insert_chatsession(self, session_id, session_owner,session_topic, start_time): + def insert_chatsession(self, session_id, session_owner,session_topic, start_time,thread_id = ""): """ insert a new session into the ChatSessions table """ try: conn = self._get_conn() conn.execute(""" - INSERT INTO ChatSessions (SessionID, SessionOwner,SessionTopic, StartTime,SummarizePos,Summary) - VALUES (?,?, ?, ?,0,"") - """, (session_id, session_owner,session_topic, start_time)) + INSERT INTO ChatSessions (SessionID, SessionOwner,SessionTopic, StartTime,SummarizePos,Summary,ThreadID) + VALUES (?,?, ?, ?,0,"",?) + """, (session_id, session_owner,session_topic, start_time,thread_id)) conn.commit() return 0 # return 0 if successful except Error as e: @@ -253,6 +254,21 @@ class ChatSessionDB: except Error as e: logging.error("Error occurred while updating session summary: %s", e) return -1 + + def update_session_thread_id(self, session_id, thread_id): + """ update the threadid of a session """ + try: + conn = self._get_conn() + conn.execute(""" + UPDATE ChatSessions + SET ThreadID = ? + WHERE SessionID = ? + """, (thread_id, session_id)) + conn.commit() + return 0 # return 0 if successful + except Error as e: + logging.error("Error occurred while updating session threadid: %s", e) + return -1 # chat session store the chat history between owner and agent # chat session might be large, so can read / write at stream mode. @@ -286,6 +302,7 @@ class AIChatSession: result.topic = session_topic result.summarize_pos = session[4] result.summary = session[5] + result.openai_thread_id = session[6] return result @@ -305,6 +322,7 @@ class AIChatSession: result.topic = session[2] result.summarize_pos = session[4] result.summary = session[5] + result.openai_thread_id = session[6] return result @@ -331,6 +349,7 @@ class AIChatSession: self.start_time : str = None self.summarize_pos : int = 0 self.summary = None + self.openai_thread_id = None def get_owner_id(self) -> str: return self.owner_id @@ -376,6 +395,10 @@ class AIChatSession: self.summarize_pos = progress self.summary = new_summary + def update_openai_thread_id(self,thread_id:str) -> None: + self.db.update_session_thread_id(self.session_id,thread_id) + self.openai_thread_id = thread_id + #def attach_event_handler(self,handler) -> None: # """chat session changed event handler""" # pass diff --git a/src/aios_kernel/contact.py b/src/aios_kernel/contact.py new file mode 100644 index 0000000..ba80df5 --- /dev/null +++ b/src/aios_kernel/contact.py @@ -0,0 +1,84 @@ +from typing import List +import logging +from datetime import datetime + +from .agent_base import AgentMsg +from .tunnel import AgentTunnel + + +logger = logging.getLogger(__name__) +class Contact: + def __init__(self, name, phone=None, email=None, telegram=None,added_by=None, tags=[], notes=""): + self.name = name + self.phone = phone + self.email = email + self.telegram = telegram + self.added_by = added_by + self.tags = tags + self.notes = notes + self.is_family_member = False + self.active_tunnels = {} + + def to_dict(self): + return { + "name": self.name, + "phone": self.phone, + "email": self.email, + "telegram" : self.telegram, + + "added_by": self.added_by, + "tags": self.tags, + "notes": self.notes, + "now" : datetime.now().strftime('%Y-%m-%d %H:%M:%S') + } + + async def _process_msg(self,msg:AgentMsg): + tunnel : AgentTunnel = self.get_active_tunnel(msg.sender) + if tunnel is not None: + await tunnel.post_message(msg) + return None + else: + tunnel = await self.create_default_tunnel(msg.sender) + if tunnel is not None: + self.active_tunnels[msg.sender] = tunnel + await tunnel.post_message(msg) + return None + + logger.warn(f"contact {self.name} cann't get tunnel,post message failed!") + + def get_active_tunnel(self,agent_id) -> AgentTunnel: + tunnel = self.active_tunnels.get(agent_id) + return tunnel + + def set_active_tunnel(self,agent_id,tunnel:AgentTunnel): + self.active_tunnels[agent_id] = tunnel + + async def create_default_tunnel(self,agent_id:str) -> AgentTunnel: + from .email_tunnel import EmailTunnel + + result_tunnels = AgentTunnel.get_tunnel_by_agentid(agent_id) + for tunnel in result_tunnels: + if isinstance(tunnel,EmailTunnel): + return tunnel + + return None + + @classmethod + def from_dict(cls, data): + return Contact(data.get("name"), data.get("phone"), data.get("email"), data.get("telegram"),data.get("added_by"), data.get("tags"), data.get("notes")) + +class FamilyMember(Contact): + def __init__(self, name, relationship,phone=None, email=None,telegram=None): + super().__init__(name, phone, email, telegram) + self.name = name + self.relationship = relationship + self.is_family_member = True + + def to_dict(self): + result = super().to_dict() + result["relationship"] = self.relationship + return result + + @classmethod + def from_dict(cls, data): + return FamilyMember(data.get("name"),data.get("relationship"),data.get("phone"), data.get("email"),data.get("telegram")) diff --git a/src/aios_kernel/contact_manager.py b/src/aios_kernel/contact_manager.py index 5cd1d6c..daeced6 100644 --- a/src/aios_kernel/contact_manager.py +++ b/src/aios_kernel/contact_manager.py @@ -2,49 +2,13 @@ from typing import List import toml import time from datetime import datetime -class Contact: - def __init__(self, name, phone=None, email=None, telegram=None,added_by=None, tags=[], notes=""): - self.name = name - self.phone = phone - self.email = email - self.telegram = telegram - self.added_by = added_by - self.tags = tags - self.notes = notes - self.is_family_member = False +from .agent_base import AgentMsg +from .tunnel import AgentTunnel +from .contact import Contact,FamilyMember +import logging - def to_dict(self): - return { - "name": self.name, - "phone": self.phone, - "email": self.email, - "telegram" : self.telegram, +logger = logging.getLogger(__name__) - "added_by": self.added_by, - "tags": self.tags, - "notes": self.notes, - "now" : datetime.now().strftime('%Y-%m-%d %H:%M:%S') - } - - @classmethod - def from_dict(cls, data): - return Contact(data.get("name"), data.get("phone"), data.get("email"), data.get("telegram"),data.get("added_by"), data.get("tags"), data.get("notes")) - -class FamilyMember(Contact): - def __init__(self, name, relationship,phone=None, email=None,telegram=None): - super().__init__(name, phone, email, telegram) - self.name = name - self.relationship = relationship - self.is_family_member = True - - def to_dict(self): - result = super().to_dict() - result["relationship"] = self.relationship - return result - - @classmethod - def from_dict(cls, data): - return FamilyMember(data.get("name"),data.get("relationship"),data.get("phone"), data.get("email"),data.get("telegram")) class ContactManager: _instance = None @@ -154,6 +118,9 @@ class ContactManager: def list_family_members(self): return self.family_members + #def register_to_ai_bus(self, ai_bus:AIBus): + # ai_bus.register_message_handler("contact_manager", self.process_msg) + #async def process_msg(self,msg:AgentMsg): # # forword message to contact diff --git a/src/aios_kernel/email_tunnel.py b/src/aios_kernel/email_tunnel.py index 0267ba3..c0be9d2 100644 --- a/src/aios_kernel/email_tunnel.py +++ b/src/aios_kernel/email_tunnel.py @@ -9,6 +9,7 @@ import time import datetime from .tunnel import AgentTunnel from .agent_base import AgentMsg +from .contact_manager import ContactManager from email.message import EmailMessage @@ -30,7 +31,7 @@ class EmailTunnel(AgentTunnel): self.target_id = config["target"] self.tunnel_id = config["tunnel_id"] - self.type = "TelegramTunnel" + self.type = "EmailTunnel" self.email = config["email"] self.imap_server = config["imap"] s = self.imap_server.split(":") @@ -46,8 +47,10 @@ class EmailTunnel(AgentTunnel): self.login_user = config["user"] self.login_password = config["password"] - self.folder = config["folder"] - self.check_interval = config["interval"] + if config.get("folder") is not None: + self.folder = config["folder"] + if config.get("interval") is not None: + self.check_interval = config["interval"] return True @@ -55,6 +58,8 @@ class EmailTunnel(AgentTunnel): super().__init__() self.is_start = False self.read_email = {} + self.folder = "INBOX" + self.check_interval = 60 async def on_new_email(self,mail:mailparser.MailParser) -> None: remote_email_addr = mail.from_[0][1] @@ -86,7 +91,32 @@ class EmailTunnel(AgentTunnel): password=self.login_password, ) + async def post_message(self, msg: AgentMsg) -> None: + cm = ContactManager.get_instance() + contact = cm.find_contact_by_name(msg.target) + if contact is None: + logger.error(f"can't find contact {msg.target} , post message through email_tunnel failed!") + return + + target_email = contact.email + if target_email is None: + logger.error(f"contact {msg.target} has no email, post message through email_tunnel failed!") + return + + email_msg = EmailMessage() + email_msg['Subject'] = f"{msg.topic},From AIAgent {msg.sender}" + 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() @@ -141,4 +171,5 @@ class EmailTunnel(AgentTunnel): 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/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index 97289e9..f454d41 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/src/aios_kernel/tg_tunnel.py @@ -21,7 +21,8 @@ from .agent_base import AgentMsg,AgentMsgType logger = logging.getLogger(__name__) class TelegramTunnel(AgentTunnel): - + all_bots = {} + default_chatid = {} @classmethod def register_to_loader(cls): async def load_tg_tunnel(config:dict) -> AgentTunnel: @@ -55,6 +56,7 @@ class TelegramTunnel(AgentTunnel): self.update_queue = None self.allow_group = "contact" self.in_process_tg_msg = {} + self.chatid_record = {} async def _do_process_raw_message(self,bot: Bot, update_id: int) -> int: # Request updates after the last update_id @@ -81,6 +83,7 @@ class TelegramTunnel(AgentTunnel): self.update_queue = asyncio.Queue() self.bot_updater = Updater(self.bot,update_queue=self.update_queue) + TelegramTunnel.all_bots[self.target_id] = self.bot async def _run_app(): try: @@ -110,8 +113,41 @@ class TelegramTunnel(AgentTunnel): 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 _process_message(self, msg: AgentMsg) -> bool: + logger.warn(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target}") + + # async def _process_message(self, msg: AgentMsg) -> bool: + # logger.info(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target}") + # cm = ContactManager.get_instance() + # contact = cm.find_contact_by_name(msg.target) + # bot = TelegramTunnel.all_bots.get(msg.sender) + # chatid_index = f"{self.target_id}#{msg.target}" + # chatid = TelegramTunnel.default_chatid.get(chatid_index) + # if chatid is None: + # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! chatid not found!") + # return None + + # if bot is None: + # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! bot not found!") + # return None + + # if contact: + # if contact.telegram: + # await bot.send_message(chat_id=chatid,text=msg.body) + # logging.info(f"tg_tunnel send message {msg.msg_id} from agent {msg.sender} to human {msg.target} @ chatid:{chatid}success!") + # return None + + # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! contact not found!") + # return None + + async def post_message(self, msg: AgentMsg) -> None: + chatid = self.chatid_record.get(msg.target) + if chatid: + # TODO: support image and audio + await self.bot.send_message(chat_id=chatid,text=msg.body) + logging.info(f"tg_tunnel send message {msg.msg_id} from agent {msg.sender} to human {msg.target} @ chatid:{chatid}success!") + else: + logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! chatid not found!") async def conver_tg_msg_to_agent_msg(self,message:Message) -> AgentMsg: @@ -189,6 +225,8 @@ class TelegramTunnel(AgentTunnel): if contact is not None: reomte_user_name = contact.name + + #TelegramTunnel.default_chatid[f"{self.target_id}#{reomte_user_name}"] = update.effective_chat.id if not contact.is_family_member: if self.allow_group != "contact" and self.allow_group !="guest": await update.message.reply_text(f"You're not supposed to talk to me! Please contact my father~") @@ -209,7 +247,11 @@ class TelegramTunnel(AgentTunnel): contact.added_by = self.target_id cm.add_contact(contact.name, contact) reomte_user_name = contact.name - + + if contact is not None: + contact.set_active_tunnel(self.target_id,self) + self.chatid_record[reomte_user_name] = update.effective_chat.id + self.ai_bus.register_message_handler(reomte_user_name,contact._process_msg) agent_msg.sender = reomte_user_name logger.info(f"process message {agent_msg.msg_id} from {agent_msg.sender} to {agent_msg.target}") @@ -217,7 +259,7 @@ class TelegramTunnel(AgentTunnel): self.ai_bus.register_message_handler(agent_msg.target, self._process_message) resp_msg = await self.ai_bus.send_message(agent_msg,self.target_id,agent_msg.target) else: - self.ai_bus.register_message_handler(reomte_user_name, self._process_message) + #self.ai_bus.register_message_handler(reomte_user_name, self._process_message) resp_msg = await self.ai_bus.send_message(agent_msg) #await bot.send_chat_action(chat_id=update.effective_chat.id, action="typing") diff --git a/src/aios_kernel/tunnel.py b/src/aios_kernel/tunnel.py index ed5f760..2300c2c 100644 --- a/src/aios_kernel/tunnel.py +++ b/src/aios_kernel/tunnel.py @@ -54,6 +54,14 @@ class AgentTunnel(ABC): return False + @classmethod + async def get_tunnel_by_agentid(cls,agent_id:str): + result = [] + for tunnel in cls._all_tunnels.values(): + if tunnel.target_id == agent_id: + result.append(tunnel) + return result + def __init__(self) -> None: super().__init__() @@ -75,6 +83,9 @@ class AgentTunnel(ABC): self.ai_bus = ai_bus self.is_connected = True + @abstractmethod + def post_message(self, msg: AgentMsg) -> None: + pass @abstractmethod diff --git a/src/aios_kernel/workspace_env.py b/src/aios_kernel/workspace_env.py index 8c67446..7aedfb5 100644 --- a/src/aios_kernel/workspace_env.py +++ b/src/aios_kernel/workspace_env.py @@ -17,7 +17,7 @@ from typing import Any,List import os import chardet -from .agent_base import AgentMsg,AgentTodo,AgentPrompt +from .agent_base import AgentMsg,AgentTodo,AgentPrompt,AgentTodoResult from .environment import Environment,EnvironmentEvent from .ai_function import AIFunction,SimpleAIFunction from .storage import AIStorage,ResourceLocation @@ -47,40 +47,53 @@ class WorkspaceEnvironment(Environment): def get_knowledge_base(self) -> str: pass - def get_do_prompt(self,todo_type:str=None)->AgentPrompt: + def get_do_prompt(self,todo:AgentTodo=None)->AgentPrompt: return None - async def exec_op_list(self,oplist:List,agent_id:str)->None: + # result mean: list[op_error_str],have_error + async def exec_op_list(self,oplist:List,agent_id:str)->tuple[List[str],bool]: + result_str = "op list is none" if oplist is None: - return + return None,False + result_str = [] + have_error = False for op in oplist: if op["op"] == "create": - return await self.create(op["path"],op["content"]) + await self.create(op["path"],op["content"]) elif op["op"] == "write_file": is_append = op.get("is_append") if is_append is None: is_append = False - return await self.write(op["path"],op["content"],is_append) + error_str = await self.write(op["path"],op["content"],is_append) elif op["op"] == "delete": - return await self.delete(op["path"]) + error_str = await self.delete(op["path"]) elif op["op"] == "rename": - return await self.rename(op["path"],op["new_name"]) + error_str = await self.rename(op["path"],op["new_name"]) elif op["op"] == "mkdir": - return await self.mkdir(op["path"]) + error_str = await self.mkdir(op["path"]) elif op["op"] == "create_todo": todoObj = AgentTodo.from_dict(op["todo"]) todoObj.worker = agent_id todoObj.createor = agent_id parent_id = op.get("parent") - await self.create_todo(parent_id,todoObj) + error_str = await self.create_todo(parent_id,todoObj) elif op["op"] == "update_todo": todo_id = op["id"] new_stat = op["state"] - await self.update_todo(todo_id,new_stat) - + error_str = await self.update_todo(todo_id,new_stat) else: logger.error(f"execute op list failed: unknown op:{op['op']}") + error_str = f"execute op list failed: unknown op:{op['op']}" + + if error_str: + have_error = True + result_str.append(error_str) + else: + result_str.append(f"execute success!") + + + return result_str,have_error async def list(self,path:str,only_dir:bool=False) -> str: directory_path = self.root_path + path @@ -112,13 +125,16 @@ class WorkspaceEnvironment(Environment): async def write(self,path:str,content:str,is_append:bool=False) -> str: file_path = self.root_path + path - if is_append: - async with aiofiles.open(file_path, mode='a', encoding="utf-8") as f: - await f.write(content) - else: - async with aiofiles.open(file_path, mode='w', encoding="utf-8") as f: - await f.write(content) - return "success" + try: + if is_append: + async with aiofiles.open(file_path, mode='a', encoding="utf-8") as f: + await f.write(content) + else: + async with aiofiles.open(file_path, mode='w', encoding="utf-8") as f: + await f.write(content) + except Exception as e: + return str(e) + return None async def create(self,path:str,content:str=None) -> bool: if content is None: @@ -132,21 +148,29 @@ class WorkspaceEnvironment(Environment): await f.write(content) return True - async def delete(self,path:str) -> bool: - file_path = self.root_path + path - os.remove(file_path) - return True + async def delete(self,path:str) -> str: + try: + file_path = self.root_path + path + os.remove(file_path) + except Exception as e: + return str(e) + + return None async def mkdir(self,path:str) -> bool: dir_path = self.root_path + path os.makedirs(dir_path) return True - async def rename(self,path:str,new_name:str) -> bool: - file_path = self.root_path + path - new_path = self.root_path + new_name - os.rename(file_path,new_path) - return True + async def rename(self,path:str,new_name:str) -> str: + try: + file_path = self.root_path + path + new_path = self.root_path + new_name + os.rename(file_path,new_path) + except Exception as e: + return str(e) + + return None async def get_todo_tree(self,path:str = None,deep:int = 4): if path: @@ -205,17 +229,17 @@ class WorkspaceEnvironment(Environment): continue todo = await self.get_todo_by_fullpath(entry.path) - if todo.worker != agent_id: - continue - - todo.rank = int(todo.create_time)>>deep - if todo: + if todo.worker: + if todo.worker != agent_id: + continue + if parent: parent.sub_todos[todo.todo_id] = todo + result_list.append(todo) - - await scan_dir(entry.path,deep + 1,todo) + todo.rank = int(todo.create_time)>>deep + await scan_dir(entry.path,deep + 1,todo) return @@ -226,68 +250,89 @@ class WorkspaceEnvironment(Environment): return result_list async def get_todo_by_fullpath(self,path:str) -> AgentTodo: - #logger.info("get_todo_by_fullpath:%s",path) - detail_path = path + "/detail" + logger.info("get_todo_by_fullpath:%s",path) - async with aiofiles.open(detail_path, mode='r', encoding="utf-8") as f: - content = await f.read(4096) - logger.debug("get_todo_by_fullpath:%s,content:%s",path,content) - todo_dict = json.loads(content) - result_todo = AgentTodo.from_dict(todo_dict) - if result_todo: - relative_path = os.path.relpath(path, self.root_path + "/todos/") - if not relative_path.startswith('/'): - relative_path = '/' + relative_path - result_todo.todo_path = relative_path - self.known_todo[result_todo.todo_id] = result_todo - else: - logger.error("get_todo_by_path:%s,parse failed!",path) - - return result_todo + detail_path = path + "/detail" + try: + async with aiofiles.open(detail_path, mode='r', encoding="utf-8") as f: + content = await f.read(4096) + logger.debug("get_todo_by_fullpath:%s,content:%s",path,content) + todo_dict = json.loads(content) + result_todo = AgentTodo.from_dict(todo_dict) + if result_todo: + relative_path = os.path.relpath(path, self.root_path + "/todos/") + if not relative_path.startswith('/'): + relative_path = '/' + relative_path + result_todo.todo_path = relative_path + self.known_todo[result_todo.todo_id] = result_todo + else: + logger.error("get_todo_by_path:%s,parse failed!",path) + + return result_todo + except Exception as e: + logger.error("get_todo_by_path:%s,failed:%s",path,e) + return None async def get_todo(self,id:str) -> AgentTodo: return self.known_todo.get(id) - async def create_todo(self,parent_id:str,todo:AgentTodo) -> None: - if parent_id: - if parent_id not in self.known_todo: - logger.error("create_todo failed: parent_id not found!") - return False - - parent_path = self.known_todo.get(parent_id).todo_path - todo_path = f"{parent_path}/{todo.title}" - else: - todo_path = todo.title + async def create_todo(self,parent_id:str,todo:AgentTodo) -> str: + try: + if parent_id: + if parent_id not in self.known_todo: + logger.error("create_todo failed: parent_id not found!") + return False + + parent_path = self.known_todo.get(parent_id).todo_path + todo_path = f"{parent_path}/{todo.title}" + else: + todo_path = todo.title - dir_path = f"{self.root_path}/todos/{todo_path}" - - os.makedirs(dir_path) - detail_path = f"{dir_path}/detail" - if todo.todo_path is None: - todo.todo_path = todo_path - logger.info("create_todo %s",detail_path) - async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: - await f.write(json.dumps(todo.to_dict())) - self.known_todo[todo.todo_id] = todo - return True - - async def update_todo(self,todo_id:str,new_stat:str)->bool: - todo : AgentTodo = self.known_todo.get(todo_id) - if todo: - todo.status = new_stat - if todo.raw_obj is None: - todo.raw_obj = todo.to_dict() - todo.raw_obj["status"] = new_stat - - detail_path = f"{self.root_path}/todos/{todo.todo_path}/detail" - async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: - await f.write(json.dumps(todo.raw_obj)) - return True - - return False + dir_path = f"{self.root_path}/todos/{todo_path}" - async def append_do_result(self,todo:AgentTodo,result): - return True + os.makedirs(dir_path) + detail_path = f"{dir_path}/detail" + if todo.todo_path is None: + todo.todo_path = todo_path + logger.info("create_todo %s",detail_path) + async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(todo.to_dict())) + self.known_todo[todo.todo_id] = todo + except Exception as e: + logger.error("create_todo failed:%s",e) + return str(e) + + return None + + async def update_todo(self,todo_id:str,new_stat:str)->str: + try: + todo : AgentTodo = self.known_todo.get(todo_id) + if todo: + todo.state = new_stat + detail_path = f"{self.root_path}/todos/{todo.todo_path}/detail" + async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(todo.to_dict())) + return None + else: + return "todo not found." + except Exception as e: + return str(e) + + async def append_worklog(self,todo:AgentTodo,result:AgentTodoResult): + worklog = f"{self.root_path}/todos/{todo.todo_path}/.worklog" + + async with aiofiles.open(worklog, mode='w+', encoding="utf-8") as f: + content = await f.read() + if len(content) > 0: + json_obj = json.loads(content) + else: + json_obj = {} + logs = json_obj.get("logs") + if logs is None: + logs = [] + logs.append(result.to_dict()) + json_obj["logs"] = logs + await f.write(json.dumps(json_obj)) class CodeInterpreter: def __init__(self, language, debug_mode): diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 18d9839..1f56dbe 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -92,6 +92,11 @@ class AIOS_Shell: if a_workflow is not None: bus.register_message_handler(target_id,a_workflow._process_msg) return True + + a_contact = await ContactManager.get_instance().get_contact(target_id) + if a_contact is not None: + bus.register_message_handler(target_id,a_contact._process_msg) + return True return False @@ -184,7 +189,7 @@ class AIOS_Shell: await ComputeKernel.get_instance().start() AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg) - AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) + #AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) pipelines = KnowledgePipelineManager.initial(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines")) @@ -212,6 +217,7 @@ class AIOS_Shell: return "0.5.1" async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str: + #AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) agent_msg = AgentMsg() agent_msg.set(sender,target_id,msg) agent_msg.topic = topic @@ -241,6 +247,11 @@ class AIOS_Shell: input_table["allow"] = UserConfigItem("allow group (default is member,you can choose contact or guest)") case "email": tunnel_config["type"] = "EmailTunnel" + input_table["email"] = UserConfigItem("email address agent will use \n") + input_table["imap"] = UserConfigItem("imap server address,like hostname:port") + input_table["smtp"] = UserConfigItem("smtp server address,like hostname:port") + input_table["user"] = UserConfigItem("mail server login user name") + input_table["password"] = UserConfigItem("main server login password") case _: error_text = FormattedText([("class:error", f"tunnel type {tunnel_type}not support!")]) print_formatted_text(error_text,style=shell_style) @@ -406,13 +417,21 @@ class AIOS_Shell: match func_name: case 'send': show_text = FormattedText([("class:error", f'send args error,/send Tracy "Hello! It is a good day!" default')]) + sender = None if len(args) == 3: target_id = args[0] msg_content = args[1] topic = args[2] - resp = await self.send_msg(msg_content,target_id,topic,self.username) - show_text = FormattedText([("class:title", f"{self.current_topic}@{self.current_target} >>> "), - ("class:content", resp)]) + sender = self.username + elif len(args) == 4: + target_id = args[0] + msg_content = args[1] + topic = args[2] + sender = args[3] + + resp = await self.send_msg(msg_content,target_id,topic,sender) + show_text = FormattedText([("class:title", f"{self.current_topic}@{self.current_target} >>> "), + ("class:content", resp)]) return show_text case 'set_config': show_text = FormattedText([("class:error", f"set config args error,/set_config $config_item! ")])