From 760b0871cd66c202edea324b31e13c8fd7daf9db Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Tue, 22 Aug 2023 17:11:20 -0700 Subject: [PATCH] The framework design of the aios kernel has been basically completed, as well as the key logic code centered on workflow. --- doc/mvp/pkg_install_procedure.drawio | 4 +- doc/mvp/workflow.drawio | 8 +- src/aios_kernel/README | 4 +- src/aios_kernel/__init__.py | 7 + src/aios_kernel/agent.py | 88 +++++++---- src/aios_kernel/agent_group.py | 20 --- src/aios_kernel/ai_function.py | 32 ++++ src/aios_kernel/aios_shell.py | 12 +- src/aios_kernel/compute_kernel.py | 54 ++++++- src/aios_kernel/compute_node.py | 37 ++++- src/aios_kernel/environment.py | 17 +- src/aios_kernel/llm_kernel.py | 14 -- src/aios_kernel/open_ai_node.py | 4 +- src/aios_kernel/role.py | 25 +++ src/aios_kernel/workflow.py | 196 ++++++++++++++++++++++-- src/component/agent_manager/templete.py | 3 - test/agent_test.py | 3 +- test/async_test.py | 66 ++++++++ 18 files changed, 496 insertions(+), 98 deletions(-) delete mode 100644 src/aios_kernel/agent_group.py create mode 100644 src/aios_kernel/ai_function.py delete mode 100644 src/aios_kernel/llm_kernel.py create mode 100644 src/aios_kernel/role.py delete mode 100644 src/component/agent_manager/templete.py create mode 100644 test/async_test.py diff --git a/doc/mvp/pkg_install_procedure.drawio b/doc/mvp/pkg_install_procedure.drawio index b77bb77..7c2b49d 100644 --- a/doc/mvp/pkg_install_procedure.drawio +++ b/doc/mvp/pkg_install_procedure.drawio @@ -1,4 +1,4 @@ - + @@ -241,7 +241,7 @@ - + diff --git a/doc/mvp/workflow.drawio b/doc/mvp/workflow.drawio index cc154e7..dfdce99 100644 --- a/doc/mvp/workflow.drawio +++ b/doc/mvp/workflow.drawio @@ -1,6 +1,6 @@ - + - + @@ -211,7 +211,7 @@ - + @@ -251,7 +251,7 @@ - + diff --git a/src/aios_kernel/README b/src/aios_kernel/README index 30404ce..eee6354 100644 --- a/src/aios_kernel/README +++ b/src/aios_kernel/README @@ -1 +1,3 @@ -TODO \ No newline at end of file +TODO: +Embading Pipline +Knowlege Base \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index e69de29..515431f 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -0,0 +1,7 @@ +from .environment import environment,environment_event +from .agent import agent_msg,ai_agent,ai_agent_templete +from .compute_kernel import compute_kernel,compute_task +from .compute_node import compute_node,local_compute_node +from .open_ai_node import open_ai_compute_node +from .role import ai_role,ai_role_group +from .workflow import ai_workflow \ No newline at end of file diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index a8c15b9..b4738a9 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -1,58 +1,88 @@ from typing import Optional + import logging -import llm_kernel,llm_work_task logger = logging.getLogger(__name__) - - class agent_msg: def __init__(self) -> None: + self.sender = None + self.target = None + self.body = None + + def get_msg_id(self) -> str: + pass + + def get_sender(self) -> str: + return self.sender + + def get_target(self) -> str: + return self.target + + # return workflow_name, role_name, session_id + def parser_target(self,target:str) -> None: pass class agent_prompt: def __init__(self) -> None: pass - def as_str()->str: + def as_str(self)->str: pass -class agent_chat_session: + def append(self,prompt) -> None: + pass + +# chat session store the chat history between owner and agent +# chat session might be large, so can read / write at stream mode. +class ai_chat_session: def __init__(self) -> None: - self.llm_model_name = None - self.llm_instance = None - self.max_token_size = 0 - self.chat_msg_list = None - self.enable_function = True - pass - def chat(self,message:str) -> None: + def get_owner_id(self) -> str: pass - # Key functions, let the AI Agent try to run. - def completion(self)->llm_work_task: - if self.llm_instance is None: - self.llm_instance = llm_kernel.craete(self.llm_model_name) - if self.llm_instance is None: - logger.fatal(f"cann't get llm_kerenel : {self.llm_model_name}") - return - - llm_work_task = self.llm_instance.completion(self._get_prompt(),self.max_token_size) - return llm_work_task - def _get_prompt(str) -> str: + def append_post(self,msg:agent_msg) -> None: + """append msg to session, msg is post from session (owner => msg.target)""" + pass + + def append_recv(self,msg:agent_msg) -> None: + """append msg to session, msg is recv from msg'sender (msg.sender => owner)""" pass + def attach_event_handler(self,handler) -> None: + """chat session changed event handler""" + pass + + #TODO : add iterator interface for read chat history + +class ai_agent_templete: + def __init__(self) -> None: + pass + class ai_agent: def __init__(self) -> None: - pass + self.chat_sessions = None + self.llm_model_name = None + self.max_token_size = 0 + self.instance_id = None + self.template_id = None - def get_chat_session(self,chat_user_name:str,session_id:Optional[str]) -> agent_chat_session: + def get_id(self) -> str: + return self.instance_id + + def get_template_id(self) -> str: + return self.template_id + + def get_chat_session_for_msg(self,msg:agent_msg) -> ai_chat_session: pass + def get_chat_session(self,sender:str,session_id:str) -> ai_chat_session: + pass + def get_llm_model_name(self) -> str: + return self.llm_model_name + + def get_max_token_size(self) -> int: + return self.max_token_size - #chat_session = agent.get_default_chat_session("master"); - #chat_session.chat("给我讲一个英文笑话!"); - #chat_session.completion(); - #print(chat_session.last_msg()); \ No newline at end of file diff --git a/src/aios_kernel/agent_group.py b/src/aios_kernel/agent_group.py deleted file mode 100644 index edb031d..0000000 --- a/src/aios_kernel/agent_group.py +++ /dev/null @@ -1,20 +0,0 @@ -from abc import ABC, abstractmethod -from typing import Any -from .agent import agent,agent_msg - -class ai_role: - def __init__(self) -> None: - pass - -class agent_group: - def __init__(self) -> None: - self.roles = None - pass - - def add_role(self,role_name:str,agent_id:str) -> None: - pass - - def send_msg(self,role_name:str,msg:agent_msg) -> None: - pass - - \ No newline at end of file diff --git a/src/aios_kernel/ai_function.py b/src/aios_kernel/ai_function.py new file mode 100644 index 0000000..b3367fb --- /dev/null +++ b/src/aios_kernel/ai_function.py @@ -0,0 +1,32 @@ +class ai_function: + def __init__(self) -> None: + self.intro : str = None + + def load_from_config(self,config:dict) -> bool: + pass + + def is_local(self) -> bool: + pass + + def is_in_zone(self) -> bool: + pass + + def is_readyonly(self) -> bool: + pass + + def get_intro(self) -> str: + return self.intro + + async def execute(self): + pass + +# call chain is a combination of ai_function,group of ai_function. +class call_chain: + def __init__(self) -> None: + pass + + def load_from_config(self,config:dict) -> bool: + pass + + async def execute(self): + pass \ No newline at end of file diff --git a/src/aios_kernel/aios_shell.py b/src/aios_kernel/aios_shell.py index c6e1890..5599573 100644 --- a/src/aios_kernel/aios_shell.py +++ b/src/aios_kernel/aios_shell.py @@ -1,6 +1,14 @@ +# aiso shell like bash of linux from .workflow import ai_workflow -from agent import agent_msg + + class aios_shell: - def send_msg(self,msg:agent_msg,target:ai_workflow) -> None: + def __init__(self,username:str) -> None: + pass + + async def send_msg(self,msg:str,target_workflow:str) -> str: + pass + + async def install_workflow(self,workflow_id:ai_workflow) -> None: pass \ No newline at end of file diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index be21341..07dedde 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -1,5 +1,12 @@ -from compute_node import compute_node from abc import ABC, abstractmethod +from typing import Optional +import logging +import asyncio + +from .agent import agent_prompt +from .compute_node import compute_node + +logger = logging.getLogger(__name__) # How to dispatch different computing tasks (some tasks may contain a large amount of state for correct execution) # to suitable computing nodes, achieving a balance of speed, cost, and power consumption, @@ -11,19 +18,60 @@ class compute_task(ABC): class compute_kernel: + _instance = None + def __new__(cls): + if cls._instance is None: + cls._instance = super(compute_kernel, cls).__new__(cls) + return cls._instance + def __init__(self) -> None: + self.task_queue = [] + self.is_start = False pass def run(self,task:compute_task) -> None: # check there is compute node can support this task + if self.is_task_support(task) is False: + logger.error(f"task {task.display()} is not support by any compute node") + return # add task to working_queue - pass + self.task_queue.append(task) + def start(self): + if self.is_start is True: + logger.warn("compute_kernel is already start") + return + + self.is_start = True + async def _run_task_loop(): + while True: + task = self.task_queue.pop(0) + c_node:compute_node= await self._schedule(task) + c_node.push_task(task) + + asyncio.create_task(_run_task_loop()) + + + async def _schedule(self,task) -> compute_node: pass def add_compute_node(self,node:compute_node): pass def disable_compute_node(self,): - pass \ No newline at end of file + pass + + def is_task_support(self,task:compute_task) -> bool: + pass + + + # friendly interface for use: + def llm_completion(self,prompt:agent_prompt,mode_name:Optional[str] = None,max_token:int = 0) -> compute_task: + # craete a llm_work_task ,push on queue's end + # then task_schedule would run this task.(might schedule some work_task to another host) + pass + + async def do_llm_completion(self,prompt:agent_prompt,mode_name:Optional[str] = None,max_token:int = 0) -> str: + pass + \ No newline at end of file diff --git a/src/aios_kernel/compute_node.py b/src/aios_kernel/compute_node.py index ef3c16c..2e05fca 100644 --- a/src/aios_kernel/compute_node.py +++ b/src/aios_kernel/compute_node.py @@ -1,13 +1,48 @@ from abc import ABC, abstractmethod - +from .compute_kernel import compute_task class compute_node(ABC): + @abstractmethod + async def push_task(self,task:compute_task,proiority:int = 0): + pass + + async def remove_task(self,task_id:str): + pass + + @abstractmethod + def get_task_state(self,task_id:str): + pass + @abstractmethod def display(self) -> str: pass + @abstractmethod + def get_capacity(self): + pass + + @abstractmethod + def is_support(self,task_type:str) -> bool: + pass + + @abstractmethod + def is_local(self) -> bool: + pass + + @abstractmethod + def is_trusted(self) -> bool: + return True + + def get_fee_type(self) -> str: + return "free" + + + class local_compute_node(compute_node): def display(self) -> str: return super().display() + def is_local(self) -> bool: + return True + diff --git a/src/aios_kernel/environment.py b/src/aios_kernel/environment.py index 2524c36..60b2625 100644 --- a/src/aios_kernel/environment.py +++ b/src/aios_kernel/environment.py @@ -1,12 +1,23 @@ +# basic environment class +# we have some built-in environment: Calender(include timer),Home(connect to IoT device in your home), ,KnwoledgeBase,FileSystem, + +from abc import ABC, abstractmethod +from typing import Callable + +from .agent import agent_msg + class environment_event(ABC): @abstractmethod def display(self) -> str: pass - class environment: def __init__(self) -> None: pass - def event_to_msg(self,) -> environment_event: - pass \ No newline at end of file + def get_id(self) -> str: + pass + + def attach_event_handler(self,event_id:str,handler:Callable) -> None: + pass + diff --git a/src/aios_kernel/llm_kernel.py b/src/aios_kernel/llm_kernel.py deleted file mode 100644 index 992db7f..0000000 --- a/src/aios_kernel/llm_kernel.py +++ /dev/null @@ -1,14 +0,0 @@ - -class llm_work_task: - def __init__(self) -> None: - pass - - -class llm_kernel: - def __init__(self) -> None: - pass - def completion(self,prompt:str,max_token:int) -> llm_work_task: - # craete a llm_work_task ,push on queue's end - # then task_schedule would run this task.(might schedule some work_task to another host) - pass - \ No newline at end of file diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 17f3a83..f362ed8 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -1,6 +1,8 @@ -from compute_node import compute_node +from .compute_node import compute_node class open_ai_compute_node(compute_node): def display(self) -> str: return super().display() + + \ No newline at end of file diff --git a/src/aios_kernel/role.py b/src/aios_kernel/role.py new file mode 100644 index 0000000..2be34f5 --- /dev/null +++ b/src/aios_kernel/role.py @@ -0,0 +1,25 @@ +from .agent import ai_agent + +class ai_role: + def __init__(self) -> None: + self.agent_instance_id : str = None + self.role_name : str = None + self.agent : ai_agent = None + self.introduce : str = None + + def load_from_config(self,config:dict) -> bool: + pass + + def get_intro(self) -> str: + return self.introduce + + def get_name(self) -> str: + return self.role_name + +class ai_role_group: + def __init__(self) -> None: + self.roles : dict[str,str]= None + pass + + + \ No newline at end of file diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 3a514ac..05e1faa 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -1,38 +1,206 @@ -import environment -import agent_prompt,agent_msg + +import logging +import asyncio +from typing import Optional,Tuple + +from .environment import environment,environment_event +from .agent import agent_prompt,agent_msg,ai_chat_session +from .role import ai_role +from .ai_function import call_chain +from .compute_kernel import compute_kernel + +logger = logging.getLogger(__name__) + +class ai_message_filter: + def __init__(self) -> None: + pass + def select(self,msg:agent_msg) -> ai_role: + pass class ai_workflow: def __init__(self) -> None: self.rule_prompt : agent_prompt = None self.workflow_config = None - self.context = None - + self.role_group = None + self.input_filter : ai_message_filter= None + self.msg_queue = [] + self.connected_environment = {} + def load_from_disk(self,config_path:str,context_dir_path) -> int: pass - def send_msg(self,msg:agent_msg,target_group:str = None) -> None: - if target_group is None: - target_group = self.get_default_group() + #workflow is asynchronous. + # When processing one message, it can process another message at the same time. + # chatsession is synchronous, it has to wait for the previous message to finish processing before it can process the next message. + # Therefore, post a message needs to specify the session_id explicitly, if not specified it will be automatically created by workflow. + def post_msg(self,msg:agent_msg) -> None: + self.msg_queue.append(msg) + return + + async def send_msg(self,msg:agent_msg) -> str: + pass + + async def run(self): + # TODO add tracking design of msg processing + while True: + the_msg = await self._pop_msg() + chatsession:ai_chat_session = self._get_chat_session_for_msg(the_msg) + if chatsession is None: + logger.error(f"get_chat_session_for_msg return None for :{the_msg}") + continue + + chatsession.append_recv(the_msg) + + async def _process_msg(msg:agent_msg,the_role) -> None: + # prompt generat progress is most important part of workflow(app) develope + prompt = agent_prompt() + prompt.append(the_role.get_prompt()) + prompt.append(self.get_workflow_rule_prompt()) + prompt.append(self._get_function_prompt(the_role.get_name())) + prompt.append(self._get_knowlege_prompt(the_role.get_name())) + prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context + + result = await compute_kernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + final_result = result + result_type : str = self._get_llm_result_type(result) + is_ignore = False + match result_type: + case "function": + callchain:call_chain = self._parse_function_call_chain(result) + resp = await callchain.exec() + if callchain.have_result(): + # generator proc resp prompt with WAITING state + proc_resp_prompt:agent_prompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) + final_result = await compute_kernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + return final_result + + + case "send_message": + # send message to other / sub workflow + next_msg:agent_msg = self._parse_to_msg(result) + if next_msg is not None: + # TODO: Next Target can be another role in workflow + next_workflow:ai_workflow = self.get_workflow(next_msg.get_target()) + inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) + + inner_chat_session.append_post(next_msg) + resp = await next_workflow.send_msg(next_msg) + inner_chat_session.append_recv(resp) + # generator proc resp prompt with WAITING state + proc_resp_prompt:agent_prompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) + final_result = await compute_kernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + + return final_result + + case "post_message": + # post message to other / sub workflow + next_msg:agent_msg = self._parse_to_msg(result) + if next_msg is not None: + next_workflow:ai_workflow = self.get_workflow(next_msg.get_target()) + inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) + inner_chat_session.append_post(next_msg) + next_workflow.post_msg(next_msg) + + case "ignore": + is_ignore = True + + if is_ignore is not True: + # TODO : how to get inner chat session? + inner_chat_session = the_role.agent.get_chat_session_for_msg(msg) + if inner_chat_session is not None: + inner_chat_session.append_input(msg) + inner_chat_session.append_result(final_result) + + return result + + async def _workflow_process_msg(msg:agent_msg) -> None: + final_result = None + if self.input_filter is not None: + select_role = self.input_filter.select(msg) + if select_role is not None: + result = await _process_msg(msg,select_role) + if result is None: + logger.error(f"_process_msg return None for :{msg}") + return + if chatsession is not None: + chatsession.append_post(result) + final_result = result + + else: + results = {} + for this_role in self.role_group.roles: + a_result = asyncio.create_task(_process_msg(msg,this_role)) + results[this_role.get_name()] = a_result + + # merge result from all roles + # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg + final_result:agent_msg = self._merge_msg_result(results) + if chatsession is not None: + chatsession.append_post(final_result) + + if final_result is not None: + # TODO post message to source + pass + + asyncio.create_task(_workflow_process_msg(the_msg)) + + async def _pop_msg(self) -> agent_msg: pass - def run(self): - + def _get_chat_session_for_msg(self,msg:agent_msg) -> ai_chat_session: pass - def _pop_msg(self) -> Tuple[agent_msg,str]: + async def _get_prompt_from_session(self,chatsession:ai_chat_session,role_name:str) -> agent_prompt: + pass + + def _get_msg_queue(self,session_id:str): pass - def get_default_group(self) -> agent_group: + def _merge_msg_result(self,results:dict) -> agent_msg: pass - def get_group(self,group_name:str) -> agent_group: + def _get_function_prompt(self,role_name:str) -> agent_prompt: + pass + + def _get_knowlege_prompt(self,role_name:str) -> agent_prompt: + pass + + def _get_resp_prompt(self,resp:str,msg:agent_msg,role:ai_role,prompt:agent_prompt,chatsession:ai_chat_session) -> agent_prompt: pass def get_workflow_rule_prompt(self) -> agent_prompt: return self.rule_prompt + + def _get_llm_result_type(self,llm_resp_str:str) -> str: + pass - def get_inner_environment(self) -> environment: + def _parse_function_call_chain(self,llm_resp_str) -> call_chain: + pass + + def _parse_to_msg(self,llm_resp_str) -> agent_msg: + pass + + def get_workflow(self,workflow_name:str) -> ai_workflow: + """get workflow from known workflow list or sub workflow list""" + pass + + + def _env_event_to_msg(self,env_event:environment_event) -> agent_msg: + pass + + def get_inner_environment(self,env_id:str) -> environment: pass def connect_to_environment(self,env:environment) -> None: - pass \ No newline at end of file + the_env = self.connected_environment.get(env.get_id()) + if the_env is None: + self.connected_environment[env.get_id()] = env + def _env_msg_handler(env_event:environment_event) -> None: + the_msg:agent_msg= self._env_event_to_msg(env_event) + self.post_msg(the_msg) + + # register all event handler + the_env.attach_event_handler(None,_env_msg_handler) + else: + logger.warn(f"environment {env.get_id()} already connected!") + diff --git a/src/component/agent_manager/templete.py b/src/component/agent_manager/templete.py deleted file mode 100644 index 87540bf..0000000 --- a/src/component/agent_manager/templete.py +++ /dev/null @@ -1,3 +0,0 @@ -class ai_agent_templete: - def __init__(self) -> None: - pass \ No newline at end of file diff --git a/test/agent_test.py b/test/agent_test.py index e0fd3e3..78941d4 100644 --- a/test/agent_test.py +++ b/test/agent_test.py @@ -23,8 +23,9 @@ def test_agent(): agent = am.create(agent_templete,"Tracy","Wang","Tracy Wang is my english teacher") print("Agent Tracy Wang load success!"); + print(agent.get_introduce()); - #print(agent.get_introduce()); + #chat_session = agent.get_default_chat_session("master"); #chat_session.chat("给我讲一个英文笑话!"); diff --git a/test/async_test.py b/test/async_test.py new file mode 100644 index 0000000..64cda58 --- /dev/null +++ b/test/async_test.py @@ -0,0 +1,66 @@ +import asyncio +import aiohttp +import aiofiles +STATE_DONE = 0 +STATE_DOWNLOADING = 1 +class install_task: + def __init__(self) -> None: + self.download_task = None + self.state = STATE_DOWNLOADING + self.recv_bytes = 0 + self.total_bytes = 0 + +class install_task_mgr: + def __init__(self) -> None: + self.all_tasks = {} + + def create_install_task(self,url:str,target:str): + owner = self + this_task = self.all_tasks.get(url) + if this_task is not None: + return this_task + + this_task = install_task() + self.all_tasks[url] = this_task + async def down_and_write(): + async with aiofiles.open(target, 'wb') as file: + async with aiohttp.ClientSession() as session: + async with session.get(url) as response: + while True: + chunk = await response.content.read(1024*1024*16) + this_task.recv_bytes += len(chunk) + if not chunk: + break + await file.write(chunk) + print(f"download task {url} done!") + this_task.state = STATE_DONE + del owner.all_tasks[url] + + this_task.download_task = asyncio.create_task(down_and_write()) + return this_task + + + +async def test_wait_download(mgr): + this_task = mgr.create_install_task("https://www.cyfs.com/download/beta/cyberchat/android/latest","test.pkg") + await this_task.download_task + +def test_timer_download(mgr): + this_task = mgr.create_install_task("https://www.cyfs.com/download/beta/cyberchat/android/latest","test.pkg") + # start timer + async def check_timer(): + while this_task.state == STATE_DOWNLOADING: + await r = asyncio.sleep(1) + print(f"download bytes:{this_task.recv_bytes}") + print("download complete!") + + asyncio.create_task(check_timer()) + +async def test_main(): + mgr = install_task_mgr() + test_timer_download(mgr) + await test_wait_download(mgr) + await asyncio.sleep(1) + +if __name__ == "__main__": + asyncio.run(test_main()) \ No newline at end of file