From ccbef2104b0c1aa2324072244075da060cea8a88 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sun, 27 Aug 2023 18:07:33 -0700 Subject: [PATCH] framework code has been completed basicly. Through the use of aios_shell, we are now able to get agents run able (at openai compute node) --- doc/mvp/demo_workflow/workflow.toml | 81 ++++++ doc/mvp/package_manager.md | 2 +- doc/mvp/workflow.drawio | 4 +- rootfs/agents/agent_1/agent.toml | 5 + rootfs/agents/agents.cfg | 2 + rootfs/templetes/templetes.cfg | 2 + src/aios_kernel/__init__.py | 2 +- src/aios_kernel/agent.py | 251 ++++++++++++++++-- src/aios_kernel/aios_shell.py | 14 - src/aios_kernel/compute_kernel.py | 91 +++++-- src/aios_kernel/compute_node.py | 9 +- src/aios_kernel/compute_task.py | 60 +++++ src/aios_kernel/environment.py | 2 - src/aios_kernel/open_ai_node.py | 104 +++++++- src/aios_kernel/workflow.py | 9 +- src/component/agent_manager/__init__.py | 4 +- src/component/agent_manager/agent_manager.py | 40 ++- src/component/ndn_client/cid.py | 3 + src/component/ndn_client/ndn_client.py | 110 +++++++- src/component/package_manager/env.py | 232 ++++++++-------- src/component/package_manager/media_reader.py | 19 +- src/component/package_manager/pkg.py | 23 +- src/component/workflow_manager/__init__.py | 1 + .../workflow_manager/workflow_manager.py | 11 + src/service/aios_shell/aios_shell.py | 128 +++++++++ 25 files changed, 1011 insertions(+), 198 deletions(-) create mode 100644 doc/mvp/demo_workflow/workflow.toml create mode 100644 rootfs/agents/agent_1/agent.toml create mode 100644 rootfs/agents/agents.cfg create mode 100644 rootfs/templetes/templetes.cfg delete mode 100644 src/aios_kernel/aios_shell.py create mode 100644 src/aios_kernel/compute_task.py create mode 100644 src/component/workflow_manager/__init__.py create mode 100644 src/component/workflow_manager/workflow_manager.py create mode 100644 src/service/aios_shell/aios_shell.py diff --git a/doc/mvp/demo_workflow/workflow.toml b/doc/mvp/demo_workflow/workflow.toml new file mode 100644 index 0000000..66a3735 --- /dev/null +++ b/doc/mvp/demo_workflow/workflow.toml @@ -0,0 +1,81 @@ +# workflow实例场景 +# 能展示sub workflow +# 能展示env的整合 +# 能展示filter的使用 +# 能展示function的调用 +# 能展示基于工作目标/KPI的sub workflow迭代流程 +# 多人场景安排 + +# 例子：举办一个团队活动 +# 方案讨论（通过交互引导的方式收集主人的需求）与确定 + +# 活动前： +# 通讯员，对接管理参加活动的人的情况 (email spider) +# 酒店预订 简单：搜索（酒店评价） 处理异常 +# 行程（票务）预订 ：搜索，处理异常 +# 餐饮预订：给出方案，确定细节，预订 + +# 活动中： +# 进行统计和分析，调整设备 +# 安保，空调，音乐，拍照，录像 +# 响应紧急情况 + +# 活动结束后： +# 整理照片，视频，进行必要的二次创作，发送给相关人员 +# 对活动进行总结，提出改件意见（指导下一次活动） + +# 1. 人员 +# 主管，负责和客户沟通，并对每个环境的结果进行总结 +# 嘉宾对接 +# 酒店组 +# 行程组 +# 财务组 +# 多媒体组 + + +[filter] +"*" = "manager" + +[roles.manager] +name = "经理" +prompt = "你是一个活动策划公司的经理,与客户对接并向团队下达指令。你的团队分为下面几个小组:嘉宾对接组,酒店预定组,行程预订组,财务组,活动摄像组。活动策划分为四个阶段：方案讨论，活动前，活动中，活动后。你会根据客户的需求，对团队进行分工，分别完成各个阶段的工作。你的基本工作模式是： +1. 收到客户的明确的指令后,基于客户的已有信息和客户商量活动方案，和活动策划公司无关的业务你会回答‘与我无关’。当和客户完成活动方案的确认后,你会将拆解后的任务分配给各个小组 +2. 根据目前已经确认的活动方案，你要根据时间适时的检查不同小组的工作情况。当收到小组的工作情况反馈后，你会站在全局的角度判断是否需要调整活动方案，如果需要调整，你会和客户商量重新确定方案，然后再将调整后的方案分配给各个小组。 +3. 有时工作小组会主动与你沟通，反馈一些问题。你会站在全局的角度给与指导，适当的调整工作小组的工作目标。如果反馈的问题需要你和客户沟通，你会和客户沟通后重新确定方案。再将调整后的方案分配给受到影响各个小组。 +4. 当你决定要和工作小组通信时，请使用`send_message({小组名称},{内容}`）的形式。 +" +agent="agent.manager" + + +[sub_workflows] +[sub_workflows."嘉宾对接组"] +# 展现读取email和发送email与嘉宾沟通的能力 +[sub_workflows."嘉宾对接组".environments.email] +new_mail = "收到来自{event.data.from},标题为{event.data.subject}的邮件，内容为{event.data.content}的电子邮件" # 这里将new_mail事件转换为了一个来自环境的message + +[sub_workflows."嘉宾对接组".roles.leader] +name = "嘉宾对接组组长" +prompt = "你是一家活动策划公司的嘉宾对接组的组长，你的工作是基于已知信息,当前活动信息、公司经理的指令与嘉宾沟通，收集嘉宾的信息，然后将信息反馈给经理。在你看来，参加活动的多少有成员都是嘉宾，你可以通过你知道的信息给不同的成员进行分级。你的基本工作模式是： +1. 处理收到的邮件，如果邮件来自嘉宾，你会尝试从邮件的表态和内容中分享嘉宾的需要，并结合你对当前活动方案的理解判断是否需要和经理沟通，如果需要和经理沟通，你会将嘉宾的需求总结和告诉经理。不需要沟通的事项可以直接回复嘉宾。 +2. 你总是通过`call_function(get_env,'parent.topic'`的形式查询当前的活动方案。等待函数返回后，你会根据函数的返回结果继续处理上一个对话。 +3. 当你决定要和经理通信时，请使用`send_message(manager,{内容}`）的形式，内容的长度不超过200字。 +4. 当你决定要回复嘉宾时，请使用`call_function(sendmail,{嘉宾邮件地址},{标题},{内容}）的形式，内容的长度不超过500字。 +" +# 这里是孤立工作模式，组长只和经理沟通，也可以赋予其和其它组沟通的能力 +agent="agent.email.leader" + +[sub_workflows."酒店预定组"] +# 展现使用搜索引擎，并调用预订酒店的能力 +[sub_workflows."酒店预定组".environments.email] + +[sub_workflows."酒店预定组".roles.leader] +name="酒店预定组组长" + +[sub_workflows."酒店预定组".roles.research] +name="酒店搜索专家" + +[sub_workflows."行程预订组"] +# 展现处理冲突并反推 +nam3="3" + + diff --git a/doc/mvp/package_manager.md b/doc/mvp/package_manager.md index 787d259..33d9ce7 100644 --- a/doc/mvp/package_manager.md +++ b/doc/mvp/package_manager.md @@ -15,7 +15,7 @@ Let's start by introducing the two important processes. ## Install Package [![](./install_package.png)](pkg_procedure.drawio) -注意安装的依赖检测流程，按照流程允许将确实的package安装到当前env +Note that the dependency check during installation allows for the missing packages to be installed into the current environment. # Some Basic Concepts - ***env***:A target environment consisting of a series of configuration files, where packages can be loaded/installed. diff --git a/doc/mvp/workflow.drawio b/doc/mvp/workflow.drawio index 88b61cc..e49374a 100644 --- a/doc/mvp/workflow.drawio +++ b/doc/mvp/workflow.drawio @@ -1,6 +1,6 @@ - + @@ -220,7 +220,7 @@ - + diff --git a/rootfs/agents/agent_1/agent.toml b/rootfs/agents/agent_1/agent.toml new file mode 100644 index 0000000..9b13b5f --- /dev/null +++ b/rootfs/agents/agent_1/agent.toml @@ -0,0 +1,5 @@ +instance_id = "agent:xxxxxxabcde" +fullname = "tracy wang" +[[prompt]] +role = "system" +content = "你是我的私人英文老师，和我用地道的美式英语进行交流。你会在和我交流的同时，调整我的输入成为更地道的美式句子，并根据你对我英文水平的预测，对可能发错英的单词标上音标。如果我给你发中文，说明我不知道这句话用美式英语怎么说，你依旧按上述规则回应我。" diff --git a/rootfs/agents/agents.cfg b/rootfs/agents/agents.cfg new file mode 100644 index 0000000..c9f8d63 --- /dev/null +++ b/rootfs/agents/agents.cfg @@ -0,0 +1,2 @@ +main = "./" +cache = "./.agents" \ No newline at end of file diff --git a/rootfs/templetes/templetes.cfg b/rootfs/templetes/templetes.cfg new file mode 100644 index 0000000..9cff26f --- /dev/null +++ b/rootfs/templetes/templetes.cfg @@ -0,0 +1,2 @@ +main = "./" +cache = "./.templetes" \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 5eff80e..04d4cc5 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,5 +1,5 @@ from .environment import Environment,EnvironmentEvent -from .agent import AgentMsg,AIAgent,AIAgentTemplete +from .agent import AgentMsg,AIAgent,AIAgentTemplete,AgentMsgState,AgentPrompt,AgentMsgState from .compute_kernel import ComputeKernel,ComputeTask 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 6b0e5fc..8346649 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -1,14 +1,37 @@ from typing import Optional - +from enum import Enum +from asyncio import Queue +import asyncio import logging +import uuid +import time + logger = logging.getLogger(__name__) + +class AgentMsgState(Enum): + RESPONSED = 0 + INIT = 1 + SENDING = 2 + PROCESSING = 3 + ERROR = 4 + class AgentMsg: def __init__(self) -> None: - self.sender = None - self.target = None - self.body = None + self.create_time = 0 + self.sender:str = None + self.target:str = None + self.body:str = None + self.state = AgentMsgState.INIT + + self.resp_msg = None + + def set(self,sender:str,target:str,body:str) -> None: + self.sender = sender + self.target = target + self.body = body + self.create_time = time.time() def get_msg_id(self) -> str: pass @@ -25,22 +48,35 @@ class AgentMsg: class AgentPrompt: def __init__(self) -> None: - pass + self.messages = [] def as_str(self)->str: - pass + result_str = "" + if self.messages: + for msg in self.messages: + result_str += msg.get("role") + ":" + msg.get("content") + "\n" - def append(self,prompt) -> None: - pass + return result_str + + def append(self,prompt): + self.messages.extend(prompt.messages) + + 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 = config + return True # chat session store the chat history between owner and agent # chat session might be large, so can read / write at stream mode. class AIChatSession: - def __init__(self) -> None: - pass + def __init__(self,owner_id) -> None: + self.owner_id = owner_id def get_owner_id(self) -> str: - pass + return self.owner_id def append_post(self,msg:AgentMsg) -> None: """append msg to session, msg is post from session (owner => msg.target)""" @@ -58,18 +94,185 @@ class AIChatSession: class AIAgentTemplete: def __init__(self) -> None: - pass + self.llm_model_name:str = "gpt-4-0613" + self.max_token_size:int = 0 + self.template_id:str = None + self.introduce:str = None + self.author:str = None + self.prompt:AgentPrompt = None + + def load_from_config(self,config:dict) -> bool: + if config.get("llm_model_name") is not None: + self.llm_model_name = config["llm_model_name"] + if config.get("max_token_size") is not None: + self.max_token_size = config["max_token_size"] + if config.get("template_id") is not None: + self.template_id = config["template_id"] + if config.get("prompt") is not None: + self.prompt = AgentPrompt() + if self.prompt.load_from_config(config["prompt"]) is False: + logger.error("load prompt from config failed!") + return False + + return True + class AIAgent: def __init__(self) -> None: - self.chat_sessions = None - self.llm_model_name = None - self.max_token_size = 0 - self.instance_id = None - self.template_id = None + self.prompt:AgentPrompt = None + self.llm_model_name:str = None + self.max_token_size:int = 0 + self.instance_id:str = None + self.template_id:str = None + self.fullname:str = None + self.powerby = None + self.enable = True + self.chat_sessions = {} + self.unread_msg = Queue() # msg from other agent + + @classmethod + def create_from_templete(cls,templete:AIAgentTemplete, fullname:str): + # Agent just inherit from templete on craete,if template changed,agent will not change + result_agent = AIAgent() + result_agent.llm_model_name = templete.llm_model_name + result_agent.max_token_size = templete.max_token_size + result_agent.template_id = templete.template_id + result_agent.instance_id = "agent#" + uuid.uuid4().hex + result_agent.fullname = fullname + result_agent.powerby = templete.author + result_agent.prompt = templete.prompt + return result_agent + + def load_from_config(self,config:dict) -> bool: + if config.get("instance_id") is None: + logger.error("agent instance_id is None!") + return False + self.instance_id = config["instance_id"] + + if config.get("fullname") is None: + logger.error(f"agent {self.instance_id} fullname is None!") + return False + self.fullname = config["fullname"] + + if config.get("prompt") is not None: + self.prompt = AgentPrompt() + self.prompt.load_from_config(config["prompt"]) + + if config.get("powerby") is not None: + self.powerby = config["powerby"] + if config.get("template_id") is not None: + self.template_id = config["template_id"] + if config.get("llm_model_name") is not None: + self.llm_model_name = config["llm_model_name"] + if config.get("max_token_size") is not None: + self.max_token_size = config["max_token_size"] + + return True + + def post_msg(self,msg:AgentMsg) -> None: + # TODO: drop same msg already processed + msg.state = AgentMsgState.SENDING + self.unread_msg.put_nowait(msg) + + def start(self) -> None: + async def _process_msg_loop(): + while True: + msg = await self.unread_msg.get() + if msg is None: + continue + msg.state = AgentMsgState.PROCESSING + resp_msg = await self._process_msg(msg) + if resp_msg is None: + msg.state = AgentMsgState.ERROR + continue + else: + msg.state = AgentMsgState.RESPONSED + msg.resp_msg = resp_msg + + asyncio.create_task(_process_msg_loop()) + + def _get_llm_result_type(self,result:str) -> str: + if result == "ignore": + return "ignore" + + return "text" + + async def _process_msg(self,msg:AgentMsg) -> AgentMsg: + from .compute_kernel import ComputeKernel + + prompt = AgentPrompt() + prompt.append(self.prompt) + msg_prompt = AgentPrompt() + msg_prompt.messages = [{"role":msg.sender,"content":msg.body}] + prompt.append(msg_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 ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size) + final_result = result + result_type : str = self._get_llm_result_type(result) + is_ignore = False + + match result_type: + # case "function": + # callchain:CallChain = self._parse_function_call_chain(result) + # resp = await callchain.exec() + # if callchain.have_result(): + # # generator proc resp prompt with WAITING state + # proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) + # final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + # return final_result + + + # case "send_message": + # # send message to other / sub workflow + # next_msg:AgentMsg = self._parse_to_msg(result) + # if next_msg is not None: + # # TODO: Next Target can be another role in workflow + # next_workflow:Workflow = self.get_workflow(next_msg.get_target()) + # inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) + + # inner_chat_session.append_post(next_msg) + # resp = await next_workflow.send_msg(next_msg) + # inner_chat_session.append_recv(resp) + # # generator proc resp prompt with WAITING state + # proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) + # final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + + # return final_result + + #case "post_message": + # # post message to other / sub workflow + # next_msg:AgentMsg = self._parse_to_msg(result) + # if next_msg is not None: + # next_workflow:Workflow = self.get_workflow(next_msg.get_target()) + # inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) + # inner_chat_session.append_post(next_msg) + # next_workflow.post_msg(next_msg) + + case "ignore": + is_ignore = True + + if is_ignore is not True: + # TODO : how to get inner chat session? + chatsession = self.get_chat_session(msg.sender) + resp_msg = AgentMsg() + resp_msg.set(self.instance_id,msg.sender,final_result) + if chatsession is not None: + chatsession.append_recv(msg) + chatsession.append_post(final_result) + + return resp_msg + + return None + def get_id(self) -> str: return self.instance_id + + def get_fullname(self) -> str: + return self.fullname def get_template_id(self) -> str: return self.template_id @@ -77,8 +280,18 @@ class AIAgent: def get_chat_session_for_msg(self,msg:AgentMsg) -> AIChatSession: pass - def get_chat_session(self,sender:str,session_id:str) -> AIChatSession: - pass + def get_chat_session(self,remote:str,topic_name:str=None) -> AIChatSession: + if topic_name is None: + topic_name = "_" + + result_session = self.chat_sessions.get(topic_name + "@" + remote) + if result_session is not None: + return result_session + + result_session = AIChatSession(self) + self.chat_sessions[topic_name + "@" + remote] = result_session + return result_session + def get_llm_model_name(self) -> str: return self.llm_model_name diff --git a/src/aios_kernel/aios_shell.py b/src/aios_kernel/aios_shell.py deleted file mode 100644 index 610d5a9..0000000 --- a/src/aios_kernel/aios_shell.py +++ /dev/null @@ -1,14 +0,0 @@ -# aiso shell like bash of linux -from .workflow import Workflow - - -class AIOS_Shell: - 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: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 d05bd02..84351a7 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -2,32 +2,39 @@ from abc import ABC, abstractmethod from typing import Optional import logging import asyncio +from asyncio import Queue from .agent import AgentPrompt from .compute_node import ComputeNode +from .compute_task import ComputeTask,ComputeTaskState,ComputeTaskResult 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, # is the CORE GOAL of the entire computing task schedule system (aios_kernel). -class ComputeTask(ABC): - @abstractmethod - def display(self) -> str: - pass - - class ComputeKernel: _instance = None def __new__(cls): if cls._instance is None: - cls._instance = super(ComputeKernel, cls).__new__(cls) + cls._instance = super().__new__(cls) + cls._instance.is_start = False + else: + print("ComputeKernel is already created!") return cls._instance def __init__(self) -> None: - self.task_queue = [] + if self.is_start is True: + print("ComputeKernel is already start!") + return + + print("init ComputeKernel!!!") + self.is_start = True + self.task_queue = Queue() self.is_start = False - pass + self.compute_nodes = {} + + self.start() def run(self,task:ComputeTask) -> None: # check there is compute node can support this task @@ -35,7 +42,7 @@ class ComputeKernel: logger.error(f"task {task.display()} is not support by any compute node") return # add task to working_queue - self.task_queue.append(task) + self.task_queue.put_nowait(task) def start(self): @@ -46,32 +53,72 @@ class ComputeKernel: self.is_start = True async def _run_task_loop(): while True: - task = self.task_queue.pop(0) - c_node:ComputeNode= await self._schedule(task) - c_node.push_task(task) + logger.info("compute_kernel is waiting for task...") + task = await self.task_queue.get() + logger.info(f"compute_kernel get task: {task.display()}") + c_node:ComputeNode = self._schedule(task) + await c_node.push_task(task) + + logger.warn("compute_kernel is stoped!") asyncio.create_task(_run_task_loop()) - async def _schedule(self,task) -> ComputeNode: - pass + def _schedule(self,task) -> ComputeNode: + for node in self.compute_nodes.values(): + if node.is_support(task) is True: + return node + logger.warning(f"task {task.display()} is not support by any compute node") + return None def add_compute_node(self,node:ComputeNode): - pass + if self.compute_nodes.get(node.node_id) is not None: + logger.warn(f"compute_node {node.display()} already in compute_kernel") + return + self.compute_nodes[node.node_id] = node + logger.info(f"add compute_node {node.display()} to compute_kernel") - def disable_compute_node(self,): - pass + def disable_compute_node(self,node_id:str): + node = self.compute_nodes.get(node_id) + if node is None: + logger.warn(f"compute_node {node_id} not in compute_kernel") + return + node.enable = False def is_task_support(self,task:ComputeTask) -> bool: - pass + return True # friendly interface for use: - def llm_completion(self,prompt:AgentPrompt,mode_name:Optional[str] = None,max_token:int = 0) -> ComputeTask: + def llm_completion(self,prompt:AgentPrompt,mode_name:Optional[str] = None,max_token:int = 0): # 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 + task_req = ComputeTask() + task_req.set_llm_params(prompt,mode_name,max_token) + self.run(task_req) + return task_req async def do_llm_completion(self,prompt:AgentPrompt,mode_name:Optional[str] = None,max_token:int = 0) -> str: - pass + task_req = self.llm_completion(prompt,mode_name,max_token) + async def check_timer(): + check_times = 0 + while True: + if task_req.state == ComputeTaskState.DONE: + break + + if task_req.state == ComputeTaskState.ERROR: + break + + if check_times >= 20: + task_req.state = ComputeTaskState.ERROR + break + + await asyncio.sleep(0.5) + check_times += 1 + + await asyncio.create_task(check_timer()) + if task_req.state == ComputeTaskState.DONE: + return task_req.result.result_str + + return "error!" \ No newline at end of file diff --git a/src/aios_kernel/compute_node.py b/src/aios_kernel/compute_node.py index acf32a6..6fe0059 100644 --- a/src/aios_kernel/compute_node.py +++ b/src/aios_kernel/compute_node.py @@ -1,11 +1,17 @@ from abc import ABC, abstractmethod -from .compute_kernel import ComputeTask +from .compute_task import ComputeTask + class ComputeNode(ABC): + def __init__(self) -> None: + self.node_id = "default" + self.enable = True + @abstractmethod async def push_task(self,task:ComputeTask,proiority:int = 0): pass + @abstractmethod async def remove_task(self,task_id:str): pass @@ -29,7 +35,6 @@ class ComputeNode(ABC): def is_local(self) -> bool: pass - @abstractmethod def is_trusted(self) -> bool: return True diff --git a/src/aios_kernel/compute_task.py b/src/aios_kernel/compute_task.py new file mode 100644 index 0000000..f2253e0 --- /dev/null +++ b/src/aios_kernel/compute_task.py @@ -0,0 +1,60 @@ + +from enum import Enum +import uuid +import time + +class ComputeTaskState(Enum): + DONE = 0 + INIT = 1 + RUNNING = 2 + ERROR = 3 + PENDING = 4 + + +class ComputeTask: + def __init__(self) -> None: + self.task_type = "llm_completion" + self.create_time = None + + self.task_id:str = None + self.callchain_id:str = None + self.params:dict = {} + self.refers:dict = None + self.pading_data:bytearray = None + + self.state = ComputeTaskState.INIT + self.result = None + self.error_str = None + + def set_llm_params(self,prompts,model_name,max_token_size,callchain_id = None): + self.task_type = "llm_completion" + self.create_time = time.time() + self.task_id = uuid.uuid4().hex + self.callchain_id = callchain_id + self.params["prompts"] = prompts.messages + if model_name is not None: + self.params["model_name"] = model_name + else: + self.params["model_name"] = "gpt-4-0613" + self.params["max_token_size"] = max_token_size + + def display(self) -> str: + return f"ComputeTask: {self.task_id} {self.task_type} {self.state}" + + +class ComputeTaskResult: + def __init__(self) -> None: + self.create_time = None + self.task_id:str = None + self.callchain_id:str = None + self.worker_id:str = None + self.result_code:int = 0 + self.result_str:str = None + + self.result:dict = {} + self.result_refers:dict = None + self.pading_data:bytearray = None + + def set_from_task(self,task:ComputeTask): + self.task_id = task.task_id + self.callchain_id = task.callchain_id diff --git a/src/aios_kernel/environment.py b/src/aios_kernel/environment.py index 29de5b8..64c6481 100644 --- a/src/aios_kernel/environment.py +++ b/src/aios_kernel/environment.py @@ -4,8 +4,6 @@ from abc import ABC, abstractmethod from typing import Callable -from .agent import AgentMsg - class EnvironmentEvent(ABC): @abstractmethod def display(self) -> str: diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 69279d9..cf88815 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -1,8 +1,110 @@ + +import openai +import os +import asyncio +from asyncio import Queue +import logging + +from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState from .compute_node import ComputeNode +logger = logging.getLogger(__name__) + class OpenAI_ComputeNode(ComputeNode): + _instance = None + def __new__(cls): + if cls._instance is None: + cls._instance = super(OpenAI_ComputeNode, cls).__new__(cls) + cls._instance.is_start = False + return cls._instance + + def __init__(self) -> None: + super().__init__() + if self.is_start is True: + logger.warn("OpenAI_ComputeNode is already start") + return + + self.is_start = True + #openai.organization = "org-AoKrOtF2myemvfiFfnsSU8rF" #buckycloud + self.openai_api_key = "" + self.node_id = "openai_node" + + self.task_queue = Queue() + + if os.getenv("OPENAI_API_KEY") is not None: + openai.api_key = os.getenv("OPENAI_API_KEY") + else: + openai.api_key = self.openai_api_key + + self.start() + + async def push_task(self,task:ComputeTask,proiority:int = 0): + logger.info(f"openai_node push task: {task.display()}") + self.task_queue.put_nowait(task) + + async def remove_task(self,task_id:str): + pass + + def _run_task(self,task:ComputeTask): + task.state = ComputeTaskState.RUNNING + mode_name = task.params["model_name"] + # max_token_size = task.params["max_token_size"] + prompts = task.params["prompts"] + + logger.info(f"call openai {mode_name} prompts: {prompts}") + resp = openai.ChatCompletion.create(model=mode_name, + messages=prompts, + max_tokens=2000, + temperature=1.2) + logger.info(f"openai response: {resp}") + + status_code = resp["choices"][0]["finish_reason"] + if status_code != "stop": + task.state = ComputeTaskState.ERROR + task.error_str =f"The status code was {status_code}." + return None + + result = ComputeTaskResult() + result.set_from_task(task) + result.worker_id = self.node_id + result.result_str = resp["choices"][0]["message"]["content"] + result.result = resp["choices"][0]["message"] + + return result + + def start(self): + async def _run_task_loop(): + while True: + logger.info("openai_node is waiting for task...") + task = await self.task_queue.get() + logger.info(f"openai_node get task: {task.display()}") + result = self._run_task(task) + if result is not None: + task.state = ComputeTaskState.DONE + task.result = result + + asyncio.create_task(_run_task_loop()) + def display(self) -> str: - return super().display() + return f"OpenAI_ComputeNode: {self.node_id}" + + def get_task_state(self,task_id:str): + pass + + + def get_capacity(self): + pass + + + def is_support(self,task_type:str) -> bool: + return True + + + def is_local(self) -> bool: + return False + + + \ No newline at end of file diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 2192b09..1c842b8 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -1,7 +1,9 @@ import logging import asyncio +from asyncio import Queue from typing import Optional,Tuple +from abc import ABC, abstractmethod from .environment import Environment,EnvironmentEvent from .agent import AgentPrompt,AgentMsg,AIChatSession @@ -17,13 +19,14 @@ class MessageFilter: def select(self,msg:AgentMsg) -> AIRole: pass + class Workflow: def __init__(self) -> None: self.rule_prompt : AgentPrompt = None self.workflow_config = None self.role_group = None self.input_filter : MessageFilter= None - self.msg_queue = [] + self.msg_queue = Queue() self.connected_environment = {} def load_from_disk(self,config_path:str,context_dir_path) -> int: @@ -34,7 +37,7 @@ class Workflow: # 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:AgentMsg) -> None: - self.msg_queue.append(msg) + self.msg_queue.put_nowait(msg) return async def send_msg(self,msg:AgentMsg) -> str: @@ -180,7 +183,7 @@ class Workflow: def _parse_to_msg(self,llm_resp_str) -> AgentMsg: pass - def get_workflow(self,workflow_name:str) -> Workflow: + def get_workflow(self,workflow_name:str): """get workflow from known workflow list or sub workflow list""" pass diff --git a/src/component/agent_manager/__init__.py b/src/component/agent_manager/__init__.py index 7313a5a..18058c7 100644 --- a/src/component/agent_manager/__init__.py +++ b/src/component/agent_manager/__init__.py @@ -1,3 +1 @@ -from .agent_manager import AgentManager,AgentManagerClient -from .agent import ai_agent -from .templete import ai_agent_templete \ No newline at end of file +from .agent_manager import AgentManager diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index af39c88..3d1284f 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -22,13 +22,13 @@ class AgentManager: def initial(self,root_dir:str) -> None: - self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}templetes/agent_templetes.cfg") + self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}templetes/templetes.cfg") self.agent_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}agents/agents.cfg") if self.agent_templete_env is None: raise Exception("agent_manager initial failed") - def get(self,agent_id:str) -> AIAgent: + async def get(self,agent_id:str) -> AIAgent: the_agent = self.loaded_agent_instance.get(agent_id) if the_agent: return the_agent @@ -38,17 +38,17 @@ class AgentManager: if agent_media_info is None: return None - the_agent : AIAgent = self._load_agent_from_media(agent_media_info) + the_agent : AIAgent = await self._load_agent_from_media(agent_media_info) if the_agent is None: logger.warn(f"load agent {agent_id} from media failed!") - + else: + the_agent.start() return the_agent - def remove(self,agent_id:str)->int: pass - def get_templete(self,templete_id) -> AIAgentTemplete: + async def get_templete(self,templete_id) -> AIAgentTemplete: template_media_info = self.agent_templete_env.get(templete_id) if template_media_info is None: return None @@ -61,11 +61,33 @@ class AgentManager: def uninstall(self,templete_id) -> int: pass - def _load_templete_from_media(self,templete_media:PackageMediaInfo) -> AIAgentTemplete: + async def _load_templete_from_media(self,templete_media:PackageMediaInfo) -> AIAgentTemplete: pass - def _load_agent_from_media(self,agent_media:PackageMediaInfo) -> AIAgent: - pass + async def _load_agent_from_media(self,agent_media:PackageMediaInfo) -> AIAgent: + reader = self.agent_env._create_media_loader(agent_media) + if reader is None: + logger.error(f"create media loader for {agent_media} failed!") + return None + + try: + config_file = await reader.read("agent.toml","r") + if config_file is None: + logger.error(f"read agent config from {agent_media} failed!") + return None + + config_data = await config_file.read() + config = toml.loads(config_data) + result_agent = AIAgent() + if result_agent.load_from_config(config) is False: + logger.error(f"load agent from {agent_media} failed!") + return None + return result_agent + except Exception as e: + logger.error(f"read agent.toml cfg from {agent_media} failed! unexpected error occurred: {str(e)}") + return None + + def create(self,template,agent_name,agent_last_name,agent_introduce) -> AIAgent: pass diff --git a/src/component/ndn_client/cid.py b/src/component/ndn_client/cid.py index da89ed4..8bb6764 100644 --- a/src/component/ndn_client/cid.py +++ b/src/component/ndn_client/cid.py @@ -4,6 +4,9 @@ class ContentId: def __init__(self) -> None: pass + def as_str(self) -> str: + pass + @staticmethod def create_from_str(cid_str:str): pass \ No newline at end of file diff --git a/src/component/ndn_client/ndn_client.py b/src/component/ndn_client/ndn_client.py index fe46f6b..e9c46a6 100644 --- a/src/component/ndn_client/ndn_client.py +++ b/src/component/ndn_client/ndn_client.py @@ -1,13 +1,117 @@ +import asyncio,aiofiles,aiohttp +import logging +from typing import Optional from .cid import ContentId +logger = logging.getLogger(__name__) + +NDN_GET_TASK_STATE_INIT = 0 +NDN_GET_TAKS_CONNECTING = 1 +NDN_GET_TASK_STATE_DOWNLOADING = 2 +NDN_GET_TASK_STATE_VERIFYING = 3 +NDN_GET_TASK_STATE_DONE = 4 +NDN_GET_TASK_STATE_ERROR = 5 + class NDN_GetTask: def __init__(self) -> None: + self.cid:str = None + self.target_path:str = None + self.urls:[str] = None + self.options:Optional[dict] = None + + self.working_task = None + self.state = NDN_GET_TASK_STATE_INIT + self.total_size = 0 + self.recv_bytes = 0 + self.write_bytes = 0 + self.error_str = None + self.chunk_queue = None + + self.retry_count = 0 + self.used_urls = [] + self.hash_update = None + + + def select_url(self,index:int)->str: + return self.urls[0] + + def get_chunk_for_download(self)->bytes: pass class NDN_Client: def __init__(self): - pass + self.cache_dir = "" + self.default_ndn_http_gateway = "" + self.all_task = {} + self.memory_chunk_size = 1024*1024*2 + self.chunk_queue_size = 16 - def get_file(self,cid:ContentId,target_path:str,urls:{}=None,options:{}=None): - pass \ No newline at end of file + def load_config(self,config:dict): + if config.get("cache_dir"): + self.cache_dir = config.get("cache_dir") + if config.get("dndn_gateway"): + self.default_ndn_http_gateway = config.get("ndn_gateway") + + def get_file(self,cid:ContentId,target_path:str,urls:{}=None,options:{}=None)->NDN_GetTask: + get_task = self.all_task.get(cid.as_str()) + if get_task: + return get_task + else: + get_task = NDN_GetTask() + self.all_task[cid.as_str()] = get_task + + get_task.cid = cid + get_task.target_path = target_path + get_task.urls = urls + get_task.options = options + if get_task.urls is None: + get_task.urls = [f"{self.default_ndn_http_gateway}/{cid.as_str()}"] + logger.info(f"get_file {cid.as_str()} urls is None, use {get_task.urls[0]} as default") + + + async def get_file_async(): + target_file = aiofiles.open(target, 'wb') + # if file exist, check hash first + + http_session = aiohttp.ClientSession() + resp = http_session.get(get_task.select_url(0)) + if resp.status != 200: + get_task.error_str = f"get_file {cid.as_str()} failed,http status:{resp.status}" + return + get_task.total_size = resp.content_length + + async def write_file_async(): + while True: + chunk = await get_task.chunk_queue.pop() + chunk_size = len(chunk) + if not chunk or chunk_size == 0: + break + get_task.hash_update.update(chunk) + await target_file.write(chunk) + get_task.write_bytes += chunk_size + + #verify + get_task.state = NDN_GET_TASK_STATE_VERIFYING + await target_file.close() + return + + write_task = asyncio.create_task(write_file_async()) + while True: + await get_task.chunk_queue.pop() + chunk = resp.content.read(self.memory_chunk_size) + chunk_size = len(chunk) + if not chunk or chunk_size == 0: + break + + get_task.recv_bytes += len(chunk) + get_task.chunk_queue.push(chunk) + + + get_task.state = NDN_GET_TASK_STATE_DONE + await write_task + + get_task.working_task = asyncio.create_task(get_file_async()) + return get_task + + \ No newline at end of file diff --git a/src/component/package_manager/env.py b/src/component/package_manager/env.py index b71813d..0ef9cfe 100644 --- a/src/component/package_manager/env.py +++ b/src/component/package_manager/env.py @@ -1,13 +1,137 @@ import logging import toml +import os -from .pkg import PackageInfo,PackageMediaInfo, -from .env import PackageEnv -from .installer import PackageInstaller,PackageInstallTask +from .pkg import PackageInfo,PackageMediaInfo +from .media_reader import MediaReader logger = logging.getLogger(__name__) + +class PackageEnv: + def __init__(self,cfg_path:str) -> None: + self.pkg_dir : str = "./pkgs/" + self.pkg_obj_dir : str = "./.pkgs/" + + self.locked_index : str = "./pkg.lock" + self.is_strict : bool = True + self.parent_envs : list[PackageEnv] = None + self.index_dbs = None + + self.env_dir = None + self.cfg_path = cfg_path + self._load_pkg_cfg(cfg_path) + pass + + def load_from_config(self,config:dict) -> bool: + if config.get("main") is not None: + self.pkg_dir = os.path.abspath(self.env_dir + "/" + config["main"]) + + if config.get("cache") is not None: + self.pkg_obj_dir = os.path.abspath(self.env_dir + "/ " + config["cache"]) + + def load(self,pkg_name:str,search_parent=True) -> PackageMediaInfo: + pkg_path = None + pkg_id,verion_str,cid = PackageInfo.parse_pkg_name(pkg_name) + + if cid is None: + if verion_str is None: + pkg_path = f"{self.pkg_dir}/{pkg_id}" + else: + #TODO fix bug about channel here + channel:str = self.get_pkg_channel_from_version(verion_str) + the_version:str = self.get_exact_version_from_installed(verion_str) + if the_version is None: + logger.warn(f"load {pkg_name} failed: no match version from {verion_str}") + return None + if channel is None: + pkg_path = f"{self.pkg_dir}/{pkg_id}#{the_version}" + else: + pkg_path = f"{self.pkg_dir}/{pkg_id}#{channel}#{the_version}" + else: + pkg_path = f"{self.pkg_obj_dir}/.{pkg_id}/{cid}" + + media_info:PackageMediaInfo = self.try_load_pkg_media_info(pkg_path) + if media_info is None: + if search_parent is True and self.parent_envs is not None: + for parent_env in self.parent_envs: + media_info = parent_env.load(pkg_id,cid,False) + if media_info is not None: + return media_info + + if media_info is None: + logger.warn(f"pkg_load {pkg_id}, cid:{cid} error,not found ,search_parent={search_parent}") + + return media_info + + def get_exact_version_from_installed(self,verion_str:str) -> str: + pass + + def get_pkg_channel_from_version(self,pkg_version:str) -> str: + args = pkg_version.split("~") + if len(args) == 1: + return None + else: + return args[0] + + + def get_pkg_media_info(self,pkg_name:str)->PackageMediaInfo: + pass + + def try_load_pkg_media_info(self,pkg_full_path:str) -> PackageMediaInfo: + the_result : PackageMediaInfo = None + logger.debug(f"try load pkng from:{pkg_full_path}") + if os.path.isdir(pkg_full_path): + the_result = PackageMediaInfo(pkg_full_path,"dir") + + return the_result + + def _create_media_loader(self,media_info:PackageMediaInfo) -> MediaReader: + match media_info.media_type: + case "dir": + from .media_reader import FolderMediaReader + return FolderMediaReader(media_info.full_path) + + logger.error(f"create media loader for {media_info} failed!") + return None + + def get_installed_pkg_info(self,pkg_name:str) -> PackageInfo: + pass + + def lookup(self,pkg_id:str,version_str:str) -> PackageInfo: + # to make sure pkg.cid is correct, we MUST verfiy eveything here + pass + + @classmethod + def is_valied_media(pkg_full_path:str) -> bool: + pass + + def do_pkg_media_trans(self,pkg_info:PackageInfo,source_path:str,target_path:str) -> bool: + pass + + def _load_pkg_cfg(self,cfg_path:str): + if cfg_path is None: + return + + cfg = None + if len(cfg_path) < 1: + return + try: + cfg = toml.load(cfg_path) + self.env_dir = os.path.abspath(os.path.dirname(cfg_path)) + self.cfg_path = os.path.abspath(cfg_path) + except Exception as e: + logger.error(f"read pkg cfg from {cfg_path} failed! unexpected error occurred: {str(e)}") + return + + return self.load_from_config(cfg) + + + + def _preprocess_prefixs(self,prefixs): + pass + class PackageEnvManager: _instance = None def __new__(cls): @@ -33,105 +157,3 @@ class PackageEnvManager: def get_system_env(self) -> PackageEnv: pass - -class PackageEnv: - def __init__(self,cfg_path:str) -> None: - self.pkg_dir : str = "" - self.pkg_obj_dir : str = "" - self.is_strict : bool = True - self.parent_envs : list[PackageEnv] = None - self.index_dbs = None - - self.cfg_path = cfg_path - self._load_pkg_cfg(cfg_path) - pass - - def load(self,pkg_name:str,search_parent=True) -> PackageMediaInfo: - pkg_path = None - pkg_id,verion_str,cid = PackageInfo.parse_pkg_name(pkg_name) - - if cid is None: - if verion_str is None: - channel:str = self.get_pkg_channel_from_version(verion_str) - if channel is None: - pkg_path = f"{self.pkg_dir}{pkg_id}" - else: - pkg_path = f"{self.pkg_dir}{pkg_id}#{channel}" - else: - channel:str = self.get_pkg_channel_from_version(verion_str) - the_version:str = self.get_exact_version_from_installed(verion_str) - if the_version is None: - logger.warn(f"load {pkg_name} failed: no match version from {verion_str}") - return None - if channel is None: - pkg_path = f"{self.pkg_dir}{pkg_id}#{the_version}" - else: - pkg_path = f"{self.pkg_dir}{pkg_id}#{channel}#{the_version}" - else: - pkg_path = f"{self.pkg_obj_dir}.{pkg_id}/{cid}" - - media_info:PackageMediaInfo = self.try_load_pkg_media_info(pkg_path) - if media_info is None: - if search_parent: - for parent_env in self.parent_envs: - media_info = parent_env.load(pkg_id,cid,False) - if media_info is not None: - return media_info - - logger.warn(f"load {pkg_id}#{cid} error,not found ,search_parent={search_parent}") - return None - - def get_exact_version_from_installed(self,verion_str:str) -> str: - pass - - def get_pkg_channel_from_version(self,pkg_version:str) -> str: - pass - - def get_pkg_media_info(self,pkg_name:str)->PackageMediaInfo: - pass - - def try_load_pkg_media_info(self,pkg_full_path:str) -> PackageMediaInfo: - pass - - - def get_installed_pkg_info(self,pkg_name:str) -> PackageInfo: - pass - - def lookup(self,pkg_id:str,version_str:str) -> PackageInfo: - # to make sure pkg.cid is correct, we MUST verfiy eveything here - pass - - def get_installer(self) -> PackageInstaller: - pass - - @classmethod - def is_valied_media(pkg_full_path:str) -> bool: - pass - - def do_pkg_media_trans(self,pkg_info:PackageInfo,source_path:str,target_path:str) -> bool: - pass - - def _load_pkg_cfg(self,cfg_path:str): - if cfg_path is None: - return - - cfg = None - if len(cfg_path) < 1: - return - try: - cfg = toml.load(cfg_path) - except Exception as e: - logger.error(f"read pkg cfg from {cfg_path} failed! unexpected error occurred: {str(e)}") - return - - if cfg: - if cfg.env: - if cfg.env.is_strict is not None: - self.is_strict = cfg.env.is_strict - - if cfg.env.prefixs is not None: - self.prefixs = self._preprocess_prefixs(cfg.env.prefixs) - - - def _preprocess_prefixs(self,prefixs): - pass \ No newline at end of file diff --git a/src/component/package_manager/media_reader.py b/src/component/package_manager/media_reader.py index d1b941d..9c7f72e 100644 --- a/src/component/package_manager/media_reader.py +++ b/src/component/package_manager/media_reader.py @@ -1,9 +1,18 @@ -class MediaReader: - def __init__(self): +from abc import ABC, abstractmethod +import aiofiles + +class MediaReader(ABC): + @abstractmethod + async def read(self, inner_path:str,mode:str): pass - def load_from_media(self,media_info:str) -> int: + +class FolderMediaReader(MediaReader): + def __init__(self, root_dir:str) -> None: + self.root_dir = root_dir pass - def read(self, inner_path:str): - raise NotImplementedError \ No newline at end of file + async def read(self, inner_path:str,mode:str): + full_path = self.root_dir + "/" + inner_path + result_file = await aiofiles.open(full_path, mode,encoding='utf-8') + return result_file \ No newline at end of file diff --git a/src/component/package_manager/pkg.py b/src/component/package_manager/pkg.py index 9944103..7ed8431 100644 --- a/src/component/package_manager/pkg.py +++ b/src/component/package_manager/pkg.py @@ -11,10 +11,20 @@ class PackageInfo: self.target_media_type = "dir" self.source_media_type = "7z" - @classmethod - def parse_pkg_name(cls,pkg_name:str) -> Tuple[str, str, str]: - """parse pkg name like test-pkg#nightly#>0.2.31#sha1:323423423 to test-pkg,nightly#>0.2.31,sha1:323423423""" - pass + @staticmethod + def parse_pkg_name(pkg_name:str) -> Tuple[str, str, str]: + """parse pkg name like test-pkg#nightly~>0.2.31#sha1:323423423 to test-pkg,nightly#>0.2.31,sha1:323423423""" + args = pkg_name.split("#") + if len(args) == 1: + return args[0],None,None + elif len(args) == 2: + return args[0],None,arg[2] + elif len(args) == 3: + return args[0],args[1],args[2] + else: + logger.error(f"parse pkg name {pkg_name} failed!") + return None,None,None + @@ -23,8 +33,9 @@ class PackageInfo: return self.cid class PackageMediaInfo: - def __init__(self) -> None: - pass + def __init__(self,full_path,media_type) -> None: + self.media_type = media_type + self.full_path = full_path diff --git a/src/component/workflow_manager/__init__.py b/src/component/workflow_manager/__init__.py new file mode 100644 index 0000000..de62e89 --- /dev/null +++ b/src/component/workflow_manager/__init__.py @@ -0,0 +1 @@ +from .workflow_manager import WorkflowManager \ No newline at end of file diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py new file mode 100644 index 0000000..2a6a68d --- /dev/null +++ b/src/component/workflow_manager/workflow_manager.py @@ -0,0 +1,11 @@ +from aios_kernel import Workflow + +class WorkflowManager: + def __init__(self) -> None: + pass + + def initial(self,root_dir:str) -> None: + pass + + def get_workflow(self,workflow_id:str) -> Workflow: + pass \ No newline at end of file diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py new file mode 100644 index 0000000..7d989ca --- /dev/null +++ b/src/service/aios_shell/aios_shell.py @@ -0,0 +1,128 @@ +# aiso shell like bash for linux +import asyncio +import sys +import os +import logging + +from typing import Any, Optional, TypeVar, Tuple, Sequence +import argparse + +from prompt_toolkit.formatted_text.base import AnyFormattedText +from prompt_toolkit import Application, PromptSession, prompt +from prompt_toolkit.selection import SelectionState +from prompt_toolkit.history import FileHistory +from prompt_toolkit.auto_suggest import AutoSuggestFromHistory +from prompt_toolkit.completion import WordCompleter + + +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../') +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode + +sys.path.append(directory + '/../../component/') +from agent_manager import AgentManager +from workflow_manager import WorkflowManager + +class AIOS_Shell: + def __init__(self,username:str) -> None: + self.username = username + self.user_chatsession = {} + + async def initial(self) -> bool: + AgentManager().initial(directory + "/../../../rootfs/") + WorkflowManager().initial(directory + "/../../../../rootfs/workflows/workflows.cfg") + open_ai_node = OpenAI_ComputeNode() + open_ai_node.start() + ComputeKernel().add_compute_node(open_ai_node) + return True + + + def get_version(self) -> str: + return "0.0.1" + + async def send_msg(self,msg:str,target_id:str,sender:str = None) -> str: + agent_msg = AgentMsg() + agent_msg.set(sender,target_id,msg) + + agent : AIAgent = await AgentManager().get(target_id) + if agent is not None: + agent.post_msg(agent_msg) + + a_workflow = WorkflowManager().get_workflow(target_id) + if a_workflow is not None: + a_workflow.post_msg(agent_msg) + + + async def check_timer(): + check_times = 0 + while True: + if agent_msg.state == AgentMsgState.RESPONSED: + break + + if agent_msg.state == AgentMsgState.ERROR: + break + + if check_times >= 20: + agent_msg.state = AgentMsgState.ERROR + break + + await asyncio.sleep(0.5) + check_times += 1 + + await asyncio.create_task(check_timer()) + if agent_msg.state == AgentMsgState.RESPONSED: + return agent_msg.resp_msg.body + + return "error!" + + + async def install_workflow(self,workflow_id:Workflow) -> None: + pass + + +####################################################################################### +def proc_input_by_agent(): + pass + +def show_help(): + print("this is help") + + +async def main(): + print("aios shell prepareing...") + logging.basicConfig(level=logging.INFO, + format='%(asctime)s %(name)s %(levelname)s %(message)s') + + shell = AIOS_Shell("user") + await shell.initial() + print(f"aios shell {shell.get_version()} ready.") + + + completer = WordCompleter(['list agent', 'list workflow', 'exit', 'help'], ignore_case=True) + #history = FileHistory('history.txt') + while True: + user_input = await PromptSession().prompt_async('>>> ',completer=completer) + match user_input: + case "list agent": + print(AgentManager().list_agent()) + case "list workflow": + print(WorkflowManager().list_workflow()) + case "help": + show_help() + case "exit": + break + + + if user_input.startswith("send"): + args = user_input.split(" ") + if len(args) < 3: + print("send msg failed, usage: send target_id msg_content") + continue + target_id = args[1] + msg_content = args[2] + resp = await shell.send_msg(msg_content,target_id,shell.username) + print(f"<<< {resp}") + +if __name__ == "__main__": + asyncio.run(main()) +