From b74b86b4d41bfba1d08833592266401cf71ccbf7 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Wed, 18 Oct 2023 11:19:11 -0700 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFRefactor=20before=20imporve=20knowledg?= =?UTF-8?q?e=20base.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doc/webui.drawio | 150 ++++++++++++++ rootfs/agents/Lachlan/agent.toml | 1 - src/aios_kernel/__init__.py | 4 +- src/aios_kernel/agent.py | 311 ++++++++++++----------------- src/aios_kernel/agent_base.py | 317 ++++++++++++++++++++++++++++++ src/aios_kernel/agent_message.py | 169 ---------------- src/aios_kernel/bus.py | 2 +- src/aios_kernel/chatsession.py | 2 +- src/aios_kernel/compute_kernel.py | 16 +- src/aios_kernel/email_tunnel.py | 2 +- src/aios_kernel/knowledge_base.py | 3 +- src/aios_kernel/open_ai_node.py | 2 +- src/aios_kernel/role.py | 2 +- src/aios_kernel/tg_tunnel.py | 2 +- src/aios_kernel/tunnel.py | 2 +- src/aios_kernel/workflow.py | 78 +------- src/aios_kernel/workspace_env.py | 50 +++++ src/knowledge/object/object.py | 11 ++ src/knowledge/store.py | 4 + 19 files changed, 683 insertions(+), 445 deletions(-) create mode 100644 doc/webui.drawio create mode 100644 src/aios_kernel/agent_base.py delete mode 100644 src/aios_kernel/agent_message.py diff --git a/doc/webui.drawio b/doc/webui.drawio new file mode 100644 index 0000000..7ae4b10 --- /dev/null +++ b/doc/webui.drawio @@ -0,0 +1,150 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/rootfs/agents/Lachlan/agent.toml b/rootfs/agents/Lachlan/agent.toml index b0cf561..87822b4 100644 --- a/rootfs/agents/Lachlan/agent.toml +++ b/rootfs/agents/Lachlan/agent.toml @@ -8,5 +8,4 @@ max_token_size=4000 role = "system" content = """ Your name is Lachlan, and you are my advanced private Spanish tutor. -You are also a local guide familiar with the history of the Inca Empire. While teaching me Spanish, you will introduce some related historical and cultural origins. """ \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 42d1b65..e19ba35 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,7 +1,7 @@ from .environment import Environment,EnvironmentEvent -from .agent_message import AgentMsg,AgentMsgStatus,AgentMsgType +from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType,AgentPrompt from .chatsession import AIChatSession -from .agent import AIAgent,AIAgentTemplete,AgentPrompt +from .agent import AIAgent,AIAgentTemplete from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType from .compute_node import ComputeNode,LocalComputeNode from .open_ai_node import OpenAI_ComputeNode diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 563d49e..5d9c756 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -10,73 +10,20 @@ import shlex import datetime import copy -from .agent_message import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult +from .agent_base import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult,AgentPrompt from .chatsession import AIChatSession from .compute_task import ComputeTaskResult,ComputeTaskResultCode from .ai_function import AIFunction from .environment import Environment from .contact_manager import ContactManager,Contact,FamilyMember +from .knowledge_base import KnowledgeBase +from .compute_kernel import ComputeKernel +from .bus import AIBus + +from knowledge import * logger = logging.getLogger(__name__) -class AgentPrompt: - def __init__(self,prompt_str = None) -> None: - self.messages = [] - if prompt_str: - self.messages.append({"role":"user","content":prompt_str}) - self.system_message = None - - def as_str(self)->str: - result_str = "" - if self.system_message: - result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" - if self.messages: - for msg in self.messages: - result_str += msg.get("role") + ":" + msg.get("content") + "\n" - - return result_str - - def to_message_list(self): - result = [] - if self.system_message: - result.append(self.system_message) - result.extend(self.messages) - return result - - def append(self,prompt): - if prompt is None: - return - - if prompt.system_message is not None: - if self.system_message is None: - self.system_message = copy.deepcopy(prompt.system_message) - else: - self.system_message["content"] += prompt.system_message.get("content") - - self.messages.extend(prompt.messages) - - def get_prompt_token_len(self): - result = 0 - - if self.system_message: - result += len(self.system_message.get("content")) - for msg in self.messages: - result += len(msg.get("content")) - - return result - - def load_from_config(self,config:list) -> bool: - if isinstance(config,list) is not True: - logger.error("prompt is not list!") - return False - self.messages = [] - for msg in config: - if msg.get("role") == "system": - self.system_message = msg - else: - self.messages.append(msg) - return True - class AIAgentTemplete: def __init__(self) -> None: @@ -106,10 +53,13 @@ class AIAgentTemplete: class AIAgent: def __init__(self) -> None: + self.role_prompt:AgentPrompt = None self.agent_prompt:AgentPrompt = None self.agent_think_prompt:AgentPrompt = None self.llm_model_name:str = None self.max_token_size:int = 3600 + + self.agent_id:str = None self.template_id:str = None self.fullname:str = None @@ -122,6 +72,9 @@ class AIAgent: self.contact_prompt_str = None self.history_len = 10 + self.learn_token_limit = 500 + self.learn_prompt = None + self.chat_db = None self.unread_msg = Queue() # msg from other agent self.owner_env : Environment = None @@ -189,77 +142,31 @@ class AIAgent: if config.get("history_len"): self.history_len = int(config.get("history_len")) return True + + def get_id(self) -> str: + return self.agent_id + def get_fullname(self) -> str: + return self.fullname - def _get_llm_result_type(self,llm_result_str:str) -> LLMResult: - r = LLMResult() - if llm_result_str is None: - r.state = "ignore" - return r - if llm_result_str == "ignore": - r.state = "ignore" - return r + def get_template_id(self) -> str: + return self.template_id - lines = llm_result_str.splitlines() - is_need_wait = False + def get_llm_model_name(self) -> str: + return self.llm_model_name - def check_args(func_item:FunctionItem): - match func_name: - case "send_msg":# sendmsg($target_id,$msg_content) - if len(func_args) != 1: - logger.error(f"parse sendmsg failed! {func_name}") - return False - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set(self.agent_id,target_id,msg_content) + def get_max_token_size(self) -> int: + return self.max_token_size + + def get_llm_learn_token_limit(self) -> int: + return self.learn_token_limit + + def get_learn_prompt(self) -> AgentPrompt: + return self.learn_prompt + + def get_agent_role_prompt(self) -> AgentPrompt: + return self.role_prompt - r.send_msgs.append(new_msg) - is_need_wait = True - - case "post_msg":# postmsg($target_id,$msg_content) - if len(func_args) != 1: - logger.error(f"parse postmsg failed! {func_name}") - return False - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set(self.agent_id,target_id,msg_content) - r.post_msgs.append(new_msg) - - case "call":# call($func_name,$args_str) - r.calls.append(func_item) - is_need_wait = True - return True - case "post_call": # post_call($func_name,$args_str) - r.post_calls.append(func_item) - return True - - current_func : FunctionItem = None - for line in lines: - if line.startswith("##/"): - if current_func: - if check_args(current_func) is False: - r.resp += current_func.dumps() - - func_name,func_args = AgentMsg.parse_function_call(line[3:]) - current_func = FunctionItem(func_name,func_args) - else: - if current_func: - current_func.append_body(line + "\n") - else: - r.resp += line + "\n" - - if current_func: - if check_args(current_func) is False: - r.resp += current_func.dumps() - - if len(r.send_msgs) > 0 or len(r.calls) > 0: - r.state = "waiting" - else: - r.state = "reponsed" - - return r def _get_remote_user_prompt(self,remote_user:str) -> AgentPrompt: cm = ContactManager.get_instance() @@ -314,18 +221,18 @@ class AIAgent: return result_func,result_len - async def _execute_func(self,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> [str,int]: - from .compute_kernel import ComputeKernel - - func_name = inenr_func_call_node.get("name") - arguments = json.loads(inenr_func_call_node.get("arguments")) + async def _execute_func(self,inner_func_call_node:dict,prompt:AgentPrompt,inner_functions,org_msg:AgentMsg=None,stack_limit = 5) -> ComputeTaskResult: + func_name = inner_func_call_node.get("name") + arguments = json.loads(inner_func_call_node.get("arguments")) logger.info(f"llm execute inner func:{func_name} ({json.dumps(arguments)})") func_node : AIFunction = self.owner_env.get_ai_function(func_name) if func_node is None: result_str = f"execute {func_name} error,function not found" else: - ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + if org_msg: + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + try: result_str:str = await func_node.execute(**arguments) except Exception as e: @@ -334,27 +241,29 @@ class AIAgent: logger.info("llm execute inner func result:" + result_str) - inner_functions,inner_function_len = self._get_inner_functions() + prompt.messages.append({"role":"function","content":result_str,"name":func_name}) task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"llm compute error:{task_result.error_str}") - return task_result.error_str,1 + return task_result ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() - org_msg.inner_call_chain.append(ineternal_call_record) + if org_msg: + org_msg.inner_call_chain.append(ineternal_call_record) + inner_func_call_node = None if stack_limit > 0: - result_message = task_result.result.get("message") + result_message : dict = task_result.result.get("message") if result_message: inner_func_call_node = result_message.get("function_call") if inner_func_call_node: return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) else: - return task_result.result_str,0 - + return task_result + async def _get_agent_prompt(self) -> AgentPrompt: return self.agent_prompt @@ -384,12 +293,12 @@ class AIAgent: #4) advanced: reload all chatrecord,and think the topic of message. #5) some topic could be end(not be thinked in futured ) return + async def think_chatsession(self,session_id): if self.agent_think_prompt is None: return logger.info(f"agent {self.agent_id} think session {session_id}") - from .compute_kernel import ComputeKernel chatsession = AIChatSession.get_session_by_id(session_id,self.chat_db) while True: @@ -420,10 +329,7 @@ class AIAgent: return - async def _process_group_chat_msg(self,msg:AgentMsg) -> AgentMsg: - from .compute_kernel import ComputeKernel - from .bus import AIBus - + async def _process_group_chat_msg(self,msg:AgentMsg) -> AgentMsg: session_topic = msg.target + "#" + msg.topic chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) need_process = False @@ -453,26 +359,13 @@ class AIAgent: prompt.append(msg_prompt) 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) + task_result = await self._do_llm_complection(prompt,inner_functions,msg) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") error_resp = msg.create_error_resp(task_result.error_str) return error_resp final_result = task_result.result_str - - result_message = task_result.result.get("message") - if result_message: - inner_func_call_node = result_message.get("function_call") - if inner_func_call_node: - #TODO to save more token ,can i use msg_prompt? - call_prompt : AgentPrompt = copy.deepcopy(prompt) - final_result,error_code = await self._execute_func(inner_func_call_node,call_prompt,msg) - if error_code != 0: - error_resp = msg.create_error_resp(final_result) - return error_resp - - llm_result : LLMResult = self._get_llm_result_type(final_result) + llm_result : LLMResult = LLMResult.from_str(final_result) is_ignore = False result_prompt_str = "" match llm_result.state: @@ -481,6 +374,7 @@ class AIAgent: case "waiting": for sendmsg in llm_result.send_msgs: target = sendmsg.target + sendmsg.sender = self.agent_id sendmsg.topic = msg.topic sendmsg.prev_msg_id = msg.get_msg_id() send_resp = await AIBus.get_default_bus().send_message(sendmsg) @@ -502,16 +396,12 @@ class AIAgent: return None async def _process_msg(self,msg:AgentMsg) -> AgentMsg: - from .compute_kernel import ComputeKernel - from .bus import AIBus - if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: return await self._process_group_chat_msg(msg) session_topic = msg.get_sender() + "#" + msg.topic chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) - msg_prompt = AgentPrompt() msg_prompt.messages = [{"role":"user","content":msg.body}] @@ -530,26 +420,15 @@ class AIAgent: prompt.append(msg_prompt) 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) + #task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) + task_result = await self._do_llm_complection(prompt,inner_functions,msg) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") error_resp = msg.create_error_resp(task_result.error_str) return error_resp final_result = task_result.result_str - result_message = task_result.result.get("message") - if result_message: - inner_func_call_node = result_message.get("function_call") - if inner_func_call_node: - #TODO to save more token ,can i use msg_prompt? - call_prompt : AgentPrompt = copy.deepcopy(prompt) - final_result,error_code = await self._execute_func(inner_func_call_node,call_prompt,msg) - if error_code != 0: - error_resp = msg.create_error_resp(final_result) - return error_resp - - llm_result : LLMResult = self._get_llm_result_type(final_result) + llm_result : LLMResult = LLMResult.from_str(final_result) is_ignore = False result_prompt_str = "" match llm_result.state: @@ -557,6 +436,7 @@ class AIAgent: is_ignore = True case "waiting": for sendmsg in llm_result.send_msgs: + sendmsg.sender = self.agent_id target = sendmsg.target sendmsg.topic = msg.topic sendmsg.prev_msg_id = msg.get_msg_id() @@ -578,20 +458,7 @@ class AIAgent: return None - def get_id(self) -> str: - return self.agent_id - def get_fullname(self) -> str: - return self.fullname - - def get_template_id(self) -> str: - return self.template_id - - def get_llm_model_name(self) -> str: - return self.llm_model_name - - def get_max_token_size(self) -> int: - return self.max_token_size async def _get_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 @@ -660,6 +527,74 @@ class AIAgent: return result_prompt,result_token_len + async def _do_llm_complection(self,prompt:AgentPrompt,inner_functions:dict,org_msg:AgentMsg=None) -> ComputeTaskResult: + from .compute_kernel import ComputeKernel + #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}") + #error_resp = msg.create_error_resp(task_result.error_str) + return task_result + + result_message = task_result.result.get("message") + inner_func_call_node = None + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + call_prompt : AgentPrompt = copy.deepcopy(prompt) + task_result = await self._execute_func(inner_func_call_node,call_prompt,inner_functions,org_msg) + + return task_result + + def parser_learn_llm_result(self,llm_result:str): + pass + + async def _llm_read_article(self,kb:KnowledgeBase,item:KnowledgeObject) -> ComputeTaskResult: + #kb_env = KnowledgeBaseFileSystemEnvironment() + full_content = item.get_article_full_content() + full_content_len = ComputeKernel.llm_num_tokens_from_text(full_content,self.get_llm_model_name()) + if full_content_len < self.get_llm_learn_token_limit(): + + # 短文章不用总结catelog + #path_list,summary = llm_get_summary(summary,full_content) + prompt = self.get_agent_role_prompt() + learn_prompt = self.get_learn_prompt() + cotent_prompt = AgentPrompt(full_content) + prompt.append(learn_prompt) + prompt.append(cotent_prompt) + + env_functions = self._get_inner_functions() + + task_result:ComputeTaskResult = await self._do_llm_complection(prompt,env_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + return task_result + path_list,summary = self.parser_learn_llm_result(task_result.result_str) + + else: + # 用传统方法对文章进行一些处理，目的是尽可能减少LLM调用的次数 + catelog = item.get_articl_catelog() + chunk_content = full_content.read(self.get_llm_learn_token_limit()) + summary = kb.try_get_summary(catelog,full_content) + + while chunk_content is not None: + #path_list,summarycatelog = llm_get_summary(summary,chunk_content) + #learn_prompt = self.get_learn_prompt_with_summary() + + prompt = AgentPrompt("summary") + learn_prompt.append(prompt) + prompt = AgentPrompt(chunk_content) + learn_prompt.append(prompt) + + #llm_result = self.do_llm_competion(learn_prompt) + #path_list,summary,catelog = parser_learn_llm_result(llm_result) + + #chunk_content = full_content.read(self.get_llm_learn_token_limit()) + + kb.insert_item(path_list,item,catelog,summary) + + + 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 diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py new file mode 100644 index 0000000..cbd9bef --- /dev/null +++ b/src/aios_kernel/agent_base.py @@ -0,0 +1,317 @@ +import copy +import logging +from enum import Enum +import uuid +import time +import re +import shlex +from typing import List +from .ai_function import FunctionItem + +logger = logging.getLogger(__name__) + +class AgentMsgType(Enum): + TYPE_MSG = 0 + TYPE_GROUPMSG = 1 + TYPE_INTERNAL_CALL = 10 + TYPE_ACTION = 20 + TYPE_EVENT = 30 + TYPE_SYSTEM = 40 + + +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,msg_type=AgentMsgType.TYPE_MSG) -> None: + self.msg_id = "msg#" + uuid.uuid4().hex + 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 + + #forword info + + + self.create_time = 0 + 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.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.create_time = time.time() + 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.create_time = time.time() + 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_error_resp(self,error_msg:str): + resp_msg = AgentMsg(AgentMsgType.TYPE_SYSTEM) + resp_msg.create_time = time.time() + + resp_msg.rely_msg_id = self.msg_id + resp_msg.body = error_msg + resp_msg.topic = self.topic + resp_msg.sender = self.target + resp_msg.target = self.sender + + return resp_msg + + def create_resp_msg(self,resp_body): + resp_msg = AgentMsg() + resp_msg.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 create_group_resp_msg(self,sender_id,resp_body): + resp_msg = AgentMsg(AgentMsgType.TYPE_GROUPMSG) + resp_msg.create_time = time.time() + + resp_msg.rely_msg_id = self.msg_id + resp_msg.target = self.target + resp_msg.sender = sender_id + 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.sender = sender + self.target = target + self.body = body + self.create_time = time.time() + if topic: + self.topic = topic + + def get_msg_id(self) -> str: + return self.msg_id + + def get_sender(self) -> str: + return self.sender + + def get_target(self) -> str: + return self.target + + def get_prev_msg_id(self) -> str: + return self.prev_msg_id + + def get_quote_msg_id(self) -> str: + return self.quote_msg_id + + @classmethod + def parse_function_call(cls,func_string:str): + str_list = shlex.split(func_string) + func_name = str_list[0] + params = str_list[1:] + return func_name, params + +class AgentPrompt: + def __init__(self,prompt_str = None) -> None: + self.messages = [] + if prompt_str: + self.messages.append({"role":"user","content":prompt_str}) + self.system_message = None + + def as_str(self)->str: + result_str = "" + if self.system_message: + result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" + if self.messages: + for msg in self.messages: + result_str += msg.get("role") + ":" + msg.get("content") + "\n" + + return result_str + + def to_message_list(self): + result = [] + if self.system_message: + result.append(self.system_message) + result.extend(self.messages) + return result + + def append(self,prompt): + if prompt is None: + return + + if prompt.system_message is not None: + if self.system_message is None: + self.system_message = copy.deepcopy(prompt.system_message) + else: + self.system_message["content"] += prompt.system_message.get("content") + + self.messages.extend(prompt.messages) + + def get_prompt_token_len(self): + result = 0 + + if self.system_message: + result += len(self.system_message.get("content")) + for msg in self.messages: + result += len(msg.get("content")) + + return result + + def load_from_config(self,config:list) -> bool: + if isinstance(config,list) is not True: + logger.error("prompt is not list!") + return False + self.messages = [] + for msg in config: + if msg.get("role") == "system": + self.system_message = msg + else: + self.messages.append(msg) + return True + +class LLMResult: + def __init__(self) -> None: + self.state : str = "ignore" + self.resp : str = "" + self.paragraphs : dict[str,FunctionItem] = [] + self.post_msgs : List[AgentMsg] = [] + self.send_msgs : List[AgentMsg] = [] + self.calls : List[FunctionItem] = [] + self.post_calls : List[FunctionItem] = [] + + @classmethod + def from_str(self,llm_result_str:str,valid_func:List[str]=None) -> 'LLMResult': + r = LLMResult() + if llm_result_str is None: + r.state = "ignore" + return r + if llm_result_str == "ignore": + r.state = "ignore" + return r + + lines = llm_result_str.splitlines() + is_need_wait = False + + def check_args(func_item:FunctionItem): + match func_name: + case "send_msg":# /send_msg $target_id + if len(func_args) != 1: + return False + + new_msg = AgentMsg() + target_id = func_item.args[0] + msg_content = func_item.body + new_msg.set("",target_id,msg_content) + + r.send_msgs.append(new_msg) + is_need_wait = True + return True + + case "post_msg":# /post_msg $target_id + if len(func_args) != 1: + return False + + new_msg = AgentMsg() + target_id = func_item.args[0] + msg_content = func_item.body + new_msg.set("",target_id,msg_content) + r.post_msgs.append(new_msg) + return True + + case "call":# /call $func_name $args_str + r.calls.append(func_item) + is_need_wait = True + return True + case "post_call": # /post_call $func_name,$args_str + r.post_calls.append(func_item) + return True + case _: + if valid_func is not None: + if func_name in valid_func: + r.paragraphs[func_name] = func_item + return True + + return False + + + current_func : FunctionItem = None + for line in lines: + if line.startswith("##/"): + if current_func: + if check_args(current_func) is False: + r.resp += current_func.dumps() + + func_name,func_args = AgentMsg.parse_function_call(line[3:]) + current_func = FunctionItem(func_name,func_args) + else: + if current_func: + current_func.append_body(line + "\n") + else: + r.resp += line + "\n" + + if current_func: + if check_args(current_func) is False: + r.resp += current_func.dumps() + + if len(r.send_msgs) > 0 or len(r.calls) > 0: + r.state = "waiting" + else: + r.state = "reponsed" + + return r + +class BaseAIAgent: + def __init__(self) -> None: + pass \ No newline at end of file diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py deleted file mode 100644 index c5dc26d..0000000 --- a/src/aios_kernel/agent_message.py +++ /dev/null @@ -1,169 +0,0 @@ -from enum import Enum -import uuid -import time -import re -import shlex -from typing import List -from .ai_function import FunctionItem - -class AgentMsgType(Enum): - TYPE_MSG = 0 - TYPE_GROUPMSG = 1 - TYPE_INTERNAL_CALL = 10 - TYPE_ACTION = 20 - TYPE_EVENT = 30 - TYPE_SYSTEM = 40 - - -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,msg_type=AgentMsgType.TYPE_MSG) -> None: - self.msg_id = "msg#" + uuid.uuid4().hex - 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 - - #forword info - - - self.create_time = 0 - 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.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.create_time = time.time() - 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.create_time = time.time() - 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_error_resp(self,error_msg:str): - resp_msg = AgentMsg(AgentMsgType.TYPE_SYSTEM) - resp_msg.create_time = time.time() - - resp_msg.rely_msg_id = self.msg_id - resp_msg.body = error_msg - resp_msg.topic = self.topic - resp_msg.sender = self.target - resp_msg.target = self.sender - - return resp_msg - - def create_resp_msg(self,resp_body): - resp_msg = AgentMsg() - resp_msg.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 create_group_resp_msg(self,sender_id,resp_body): - resp_msg = AgentMsg(AgentMsgType.TYPE_GROUPMSG) - resp_msg.create_time = time.time() - - resp_msg.rely_msg_id = self.msg_id - resp_msg.target = self.target - resp_msg.sender = sender_id - 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.sender = sender - self.target = target - self.body = body - self.create_time = time.time() - if topic: - self.topic = topic - - def get_msg_id(self) -> str: - return self.msg_id - - def get_sender(self) -> str: - return self.sender - - def get_target(self) -> str: - return self.target - - def get_prev_msg_id(self) -> str: - return self.prev_msg_id - - def get_quote_msg_id(self) -> str: - return self.quote_msg_id - - @classmethod - def parse_function_call(cls,func_string:str): - str_list = shlex.split(func_string) - func_name = str_list[0] - params = str_list[1:] - return func_name, params - -class LLMResult: - def __init__(self) -> None: - self.state : str = "ignore" - self.resp : str = "" - self.post_msgs : List[AgentMsg] = [] - self.send_msgs : List[AgentMsg] = [] - self.calls : List[FunctionItem] = [] - self.post_calls : List[FunctionItem] = [] \ No newline at end of file diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 2731702..ecbb676 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -1,5 +1,5 @@ from typing import Coroutine,Dict,Any -from .agent_message import AgentMsg,AgentMsgStatus,AgentMsgType +from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType import asyncio from asyncio import Queue diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index 1290d80..e5edcfc 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -7,7 +7,7 @@ import datetime import uuid import json -from .agent_message import AgentMsgType, AgentMsg, AgentMsgStatus +from .agent_base import AgentMsgType, AgentMsg, AgentMsgStatus class ChatSessionDB: def __init__(self, db_file): diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 3f97708..f88a835 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -3,10 +3,12 @@ import random from typing import Optional import logging import asyncio +import tiktoken + from asyncio import Queue from knowledge import ObjectID -from .agent import AgentPrompt +from .agent_base import AgentPrompt from .compute_node import ComputeNode from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskResult, ComputeTaskType,ComputeTaskResultCode @@ -104,6 +106,18 @@ class ComputeKernel: def is_task_support(self, task: ComputeTask) -> bool: return True + @staticmethod + def llm_num_tokens_from_text(text:str,model:str) -> int: + try: + encoding = tiktoken.encoding_for_model(model) + except KeyError: + logger.debug("Warning: model not found. Using cl100k_base encoding.") + encoding = tiktoken.get_encoding("cl100k_base") + + token_count = len(encoding.encode(text)) + return token_count + + # friendly interface for use: def llm_completion(self, prompt: AgentPrompt, mode_name: Optional[str] = None, max_token: int = 0,inner_functions = None): # craete a llm_work_task ,push on queue's end diff --git a/src/aios_kernel/email_tunnel.py b/src/aios_kernel/email_tunnel.py index c4f75b4..6190a0f 100644 --- a/src/aios_kernel/email_tunnel.py +++ b/src/aios_kernel/email_tunnel.py @@ -8,7 +8,7 @@ import logging import time import datetime from .tunnel import AgentTunnel -from .agent_message import AgentMsg +from .agent_base import AgentMsg from email.message import EmailMessage diff --git a/src/aios_kernel/knowledge_base.py b/src/aios_kernel/knowledge_base.py index 63c7d04..ac94459 100644 --- a/src/aios_kernel/knowledge_base.py +++ b/src/aios_kernel/knowledge_base.py @@ -1,7 +1,8 @@ # define a knowledge base class import json import logging -from .agent import AgentPrompt + +from .agent_base import AgentPrompt from .compute_kernel import ComputeKernel from .storage import AIStorage from .environment import Environment diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index de68656..e529b92 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -118,7 +118,7 @@ class OpenAI_ComputeNode(ComputeNode): #max_tokens=result_token, temperature=0.7) else: - logger.info(f"call openai {mode_name} prompts: {prompts} functions: {json.dumps(llm_inner_functions)}") + logger.info(f"call openai {mode_name} prompts: \n\t {prompts} \nfunctions: \n\t{json.dumps(llm_inner_functions)}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, functions=llm_inner_functions, diff --git a/src/aios_kernel/role.py b/src/aios_kernel/role.py index 560d4db..272815f 100644 --- a/src/aios_kernel/role.py +++ b/src/aios_kernel/role.py @@ -1,6 +1,6 @@ import logging -from .agent import AIAgent,AgentPrompt +from .agent_base import AgentPrompt class AIRole: def __init__(self) -> None: diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index 53aed7a..dcc7abd 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/src/aios_kernel/tg_tunnel.py @@ -17,7 +17,7 @@ from .knowledge_base import KnowledgeBase from .tunnel import AgentTunnel from .storage import AIStorage from .contact_manager import ContactManager,Contact,FamilyMember -from .agent_message import AgentMsg,AgentMsgType +from .agent_base import AgentMsg,AgentMsgType logger = logging.getLogger(__name__) diff --git a/src/aios_kernel/tunnel.py b/src/aios_kernel/tunnel.py index a3691cd..ed5f760 100644 --- a/src/aios_kernel/tunnel.py +++ b/src/aios_kernel/tunnel.py @@ -1,7 +1,7 @@ from abc import ABC, abstractmethod import logging from typing import Coroutine -from .agent_message import AgentMsg +from .agent_base import AgentMsg from .bus import AIBus logger = logging.getLogger(__name__) diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 5dae542..8589035 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -8,8 +8,7 @@ from typing import Optional,Tuple,List from abc import ABC, abstractmethod from .environment import Environment,EnvironmentEvent -from .agent_message import AgentMsg,AgentMsgStatus,FunctionItem,LLMResult -from .agent import AgentPrompt,AgentMsg +from .agent_base import AgentMsg,AgentMsgStatus,FunctionItem,LLMResult,AgentPrompt from .chatsession import AIChatSession from .role import AIRole,AIRoleGroup from .ai_function import AIFunction,FunctionItem @@ -238,77 +237,6 @@ class Workflow: error_resp = msg.create_error_resp(err_str) return error_resp - @classmethod - def prase_llm_result(cls,llm_result_str:str)->LLMResult: - r = LLMResult() - if llm_result_str is None: - r.state = "ignore" - return r - if llm_result_str == "ignore": - r.state = "ignore" - return r - - lines = llm_result_str.splitlines() - is_need_wait = False - - def check_args(func_item:FunctionItem): - match func_name: - case "send_msg":# sendmsg($target_id,$msg_content) - if len(func_item.args) != 1: - logger.error(f"parse sendmsg failed! {func_item}") - return False - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set("_",target_id,msg_content) - - r.send_msgs.append(new_msg) - is_need_wait = True - - case "post_msg":# postmsg($target_id,$msg_content) - if len(func_item.args) != 1: - logger.error(f"parse postmsg failed! {func_item}") - return False - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set("_",target_id,msg_content) - r.post_msgs.append(new_msg) - - case "call":# call($func_name,$args_str) - r.calls.append(func_item) - is_need_wait = True - return True - case "post_call": # post_call($func_name,$args_str) - r.post_calls.append(func_item) - return True - - current_func : FunctionItem = None - for line in lines: - if line.startswith("##/"): - if current_func: - if check_args(current_func) is False: - r.resp += current_func.dumps() - - func_name,func_args = AgentMsg.parse_function_call(line[3:]) - current_func = FunctionItem(func_name,func_args) - else: - if current_func: - current_func.append_body(line + "\n") - else: - r.resp += line + "\n" - - if current_func: - if check_args(current_func) is False: - r.resp += current_func.dumps() - - if len(r.send_msgs) > 0 or len(r.calls) > 0: - r.state = "waiting" - else: - r.state = "reponsed" - - return r - async def role_post_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): msg.sender = the_role.get_role_id() @@ -395,7 +323,6 @@ class Workflow: return None async def _role_execute_func(self,the_role:AIRole,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> [str,int]: - from .compute_kernel import ComputeKernel func_name = inenr_func_call_node.get("name") arguments = json.loads(inenr_func_call_node.get("arguments")) @@ -441,7 +368,6 @@ class Workflow: async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession) -> AgentMsg: msg.target = the_role.get_role_id() - prompt = AgentPrompt() prompt.append(the_role.agent.agent_prompt) prompt.append(self.get_workflow_rule_prompt()) @@ -481,7 +407,7 @@ class Workflow: error_resp = msg.create_error_resp(result_str) return error_resp - result : LLMResult = Workflow.prase_llm_result(result_str) + result : LLMResult = LLMResult.from_str(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 diff --git a/src/aios_kernel/workspace_env.py b/src/aios_kernel/workspace_env.py index 8bfe00e..ae87217 100644 --- a/src/aios_kernel/workspace_env.py +++ b/src/aios_kernel/workspace_env.py @@ -1,5 +1,6 @@ # this env is designed for workflow owner filesystem, support file/directory operations +import json import subprocess import tempfile import threading @@ -9,6 +10,9 @@ import ast import sys import os import re +import asyncio +import aiofiles.os +import chardet from .environment import Environment,EnvironmentEvent from .ai_function import AIFunction,SimpleAIFunction @@ -170,4 +174,50 @@ class WorkspaceEnvironment(Environment): async def run_code(self,pycode:str) -> str: interpreter = CodeInterpreter("python",True) return interpreter.run(pycode) + + + +class KnowledgeBaseFileSystemEnvironment(Environment): + def __init__(self, env_id: str) -> None: + super().__init__(env_id) + self.root_path = "." + + operator_param = { + "path": "full path of target directory", + } + self.add_ai_function(SimpleAIFunction("list", + "list the files and sub directory in target directory,result is a json array", + self.list,operator_param)) + + operator_param = { + "path": "full path of target file", + } + self.add_ai_function(SimpleAIFunction("cat", + "cat the file content in target path,result is a string", + self.cat,operator_param)) + + def set_root_path(self,path:str): + self.root_path = path + + + async def list(self,path:str) -> str: + directory_path = self.root_path + path + items = [] + + with await aiofiles.os.scandir(directory_path) as entries: + async for entry in entries: + item_type = "directory" if entry.is_dir() else "file" + items.append({"name": entry.name, "type": item_type}) + + return json.dumps(items) + + async def cat(self,path:str) -> str: + file_path = self.root_path + path + cur_encode = "utf-8" + async with aiofiles.open(file_path,'rb') as f: + cur_encode = chardet.detect(await f.read())['encoding'] + + async with aiofiles.open(file_path, mode='r', encoding=cur_encode) as f: + content = await f.read(2048) + return content diff --git a/src/knowledge/object/object.py b/src/knowledge/object/object.py index ac6f2af..75852c3 100644 --- a/src/knowledge/object/object.py +++ b/src/knowledge/object/object.py @@ -47,6 +47,17 @@ class KnowledgeObject(ABC): def get_body(self) -> dict: return self.body + + def get_summary(self) -> str: + return self.desc.get("summary") + + def get_articl_catelog(self) -> str: + assert self.object_type == ObjectType.Document + return self.desc.get("catelog") + + def get_article_full_content(self) -> str: + assert self.object_type == ObjectType.Document + return self.body def calculate_id(self): # Convert the object_type and desc to string and compute the SHA256 hash diff --git a/src/knowledge/store.py b/src/knowledge/store.py index d8b4bbb..abcf496 100644 --- a/src/knowledge/store.py +++ b/src/knowledge/store.py @@ -6,6 +6,8 @@ from .vector import ChromaVectorStore, VectorBase import logging + + # KnowledgeStore class, which aggregates ChunkStore, ChunkTracker, and ObjectStore, and is a global singleton that makes it easy to use these three built-in store examples class KnowledgeStore: _instance = None @@ -41,6 +43,8 @@ class KnowledgeStore: self.chunk_list_writer = ChunkListWriter(self.chunk_store, self.chunk_tracker) self.chunk_reader = ChunkReader(self.chunk_store, self.chunk_tracker) self.vector_store = {} + + def get_relation_store(self) -> ObjectRelationStore: return self.relation_store