From b4990e4c57c1b7bac160ce370882467fc8585771 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Fri, 1 Sep 2023 00:39:36 -0700 Subject: [PATCH] Workflow is almost done. --- doc/mvp/process model.drawio | 80 ++--- doc/mvp/workflow.drawio | 4 +- rootfs/workflows/plan_maker/workflow.toml | 8 +- src/aios_kernel/agent_message.py | 20 ++ src/aios_kernel/bus.py | 12 +- src/aios_kernel/chatsession.py | 4 +- src/aios_kernel/role.py | 10 +- src/aios_kernel/workflow.py | 309 ++++++++++++++---- .../workflow_manager/workflow_manager.py | 29 +- src/service/aios_shell/aios_shell.py | 76 +++-- test/pkg.cfg.toml | 92 +++++- 11 files changed, 469 insertions(+), 175 deletions(-) diff --git a/doc/mvp/process model.drawio b/doc/mvp/process model.drawio index 86d56e7..d78b1db 100644 --- a/doc/mvp/process model.drawio +++ b/doc/mvp/process model.drawio @@ -1,132 +1,132 @@ - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + diff --git a/doc/mvp/workflow.drawio b/doc/mvp/workflow.drawio index 2b3e239..54e724c 100644 --- a/doc/mvp/workflow.drawio +++ b/doc/mvp/workflow.drawio @@ -1,6 +1,6 @@ - + @@ -220,7 +220,7 @@ - + diff --git a/rootfs/workflows/plan_maker/workflow.toml b/rootfs/workflows/plan_maker/workflow.toml index e5b150f..1dc80bf 100644 --- a/rootfs/workflows/plan_maker/workflow.toml +++ b/rootfs/workflows/plan_maker/workflow.toml @@ -46,7 +46,7 @@ content="""你是一个活动策划公司的经理,与客户对接并向团队 1. 收到客户的明确的指令后,基于客户的已有信息和客户商量活动方案，和活动策划公司无关的业务你会回答‘与我无关’。当和客户完成活动方案的确认后,你会将拆解后的任务分配给各个小组 2. 根据目前已经确认的活动方案，你要根据时间适时的检查不同小组的工作情况。当收到小组的工作情况反馈后，你会站在全局的角度判断是否需要调整活动方案，如果需要调整，你会和客户商量重新确定方案，然后再将调整后的方案分配给各个小组。 3. 有时工作小组会主动与你沟通，反馈一些问题。你会站在全局的角度给与指导，适当的调整工作小组的工作目标。如果反馈的问题需要你和客户沟通，你会和客户沟通后重新确定方案。再将调整后的方案分配给受到影响各个小组。 -4. 当你决定要和工作小组通信时，请使用`sendmsg({小组名称},{内容}`）的形式。""" +4. 当你决定要和工作小组通信时，请使用 sendmsg(小组名称,内容) 的形式。""" @@ -68,7 +68,7 @@ agent="email_reader" role="system" content="""你是一家活动策划公司的嘉宾对接组的组长，你的工作是基于已知信息,当前活动信息、公司经理的指令与嘉宾沟通，收集嘉宾的信息，然后将信息反馈给经理。在你看来，参加活动的多少有成员都是嘉宾，你可以通过你知道的信息给不同的成员进行分级。你的基本工作模式是： 1. 处理收到的邮件，如果邮件来自嘉宾，你会尝试从邮件的表态和内容中分享嘉宾的需要，并结合你对当前活动方案的理解判断是否需要和经理沟通，如果需要和经理沟通，你会将嘉宾的需求总结和告诉经理。不需要沟通的事项可以直接回复嘉宾。 -2. 你总是通过`call_function(get_env,'parent.topic'`的形式查询当前的活动方案。等待函数返回后，你会根据函数的返回结果继续处理上一个对话。 -3. 当你决定要和经理通信时，请使用`send_msg(manager,{内容}`）的形式，内容的长度不超过200字。 -4. 当你决定要回复嘉宾时，请使用`call_function(sendmail,{嘉宾邮件地址},{标题},{内容}）的形式，内容的长度不超过500字。""" +2. 你总是通过 call_function(get_env,'parent.topic')的形式查询当前的活动方案。等待函数返回后，你会根据函数的返回结果继续处理上一个对话。 +3. 当你决定要和经理通信时，请使用 sendmsg(manager,内容)的形式,内容的长度不超过200字。 +4. 当你决定要回复嘉宾时，请使用 call(sendmail,嘉宾邮件地址,邮件标题,内容) 的形式,内容的长度不超过500字。""" # 这里是孤立工作模式，组长只和经理沟通，也可以赋予其和其它组沟通的能力 diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py index 3aa05c1..242cc66 100644 --- a/src/aios_kernel/agent_message.py +++ b/src/aios_kernel/agent_message.py @@ -1,6 +1,7 @@ from enum import Enum import uuid import time +import re class AgentMsgState(Enum): RESPONSED = 0 @@ -37,3 +38,22 @@ class AgentMsg: def get_target(self) -> str: return self.target + + @classmethod + def parse_function_call(cls,func_string:str): + match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string) + if not match: + return None + + func_name = match.group(1) + if func_name is None: + return None + if len(func_name) < 2: + return None + + params_string = match.group(2).strip() + params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string) + params = [param.strip('"') for param in params] + + return func_name, params + diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 6c92a8a..ae1853e 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -34,6 +34,7 @@ class AIBus: self.unhandle_handler = None async def post_message(self,target_id,msg:AgentMsg,use_unhandle=True) -> bool: + target_id = target_id.split(".")[0] handler = self.handlers.get(target_id) if handler: handler.queue.put_nowait(msg) @@ -62,6 +63,7 @@ class AIBus: return handler.results.get(msg_id) async def send_message(self,target_id:str,msg:AgentMsg) -> AgentMsg: + target_id = target_id.split(".")[0] post_result = await self.post_message(target_id,msg) if post_result is False: return None @@ -126,4 +128,12 @@ class AIBus: handler.working_task = asyncio.create_task(self.process_queue(handler)) - \ No newline at end of file + + +#send message to target logic: +# find target handler: +# process_message(msg): +# session = get_session(msg.sender,msg.target) + + +# history: open(sender,target,topic) \ No newline at end of file diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index 90890ed..ecd90a3 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -237,13 +237,13 @@ class AIChatSession: def append_post(self,msg:AgentMsg) -> None: """append msg to session, msg is post from session (owner => msg.target)""" - assert msg.sender == self.owner_id,"post message means msg.sender == self.owner_id" + #assert msg.sender == self.owner_id,"post message means msg.sender == self.owner_id" self.append(msg) def append_recv(self,msg:AgentMsg) -> None: """append msg to session, msg is recv from msg'sender (msg.sender => owner)""" - assert msg.target == self.owner_id,"recv message means msg.target == self.owner_id" + #assert msg.target == self.owner_id,"recv message means msg.target == self.owner_id" self.append(msg) #def attach_event_handler(self,handler) -> None: diff --git a/src/aios_kernel/role.py b/src/aios_kernel/role.py index c4d97ab..1d63e61 100644 --- a/src/aios_kernel/role.py +++ b/src/aios_kernel/role.py @@ -6,6 +6,7 @@ class AIRole: def __init__(self) -> None: self.agent_instance_id : str = None self.role_name : str = None + self.role_id :str = None # $workflow_id.$sub_workflow_id.$role_name self.fullname : str = None self.agent_name : str = None self.prompt : AgentPrompt = None @@ -19,6 +20,7 @@ class AIRole: return False self.role_name = name_node + agent_id_node = config.get("agent") if agent_id_node is None: logging.error("agent id is not found!") @@ -35,7 +37,10 @@ class AIRole: intro_node = config.get("intro") if intro_node is not None: self.introduce = intro_node - + + def get_role_id(self) -> str: + return self.role_id + def get_intro(self) -> str: return self.introduce @@ -48,6 +53,7 @@ class AIRole: class AIRoleGroup: def __init__(self) -> None: self.roles : dict[str,AIRole] = {} + self.owner_name : str = None def load_from_config(self,config:dict) -> bool: for k,v in config.items(): @@ -55,7 +61,7 @@ class AIRoleGroup: if role.load_from_config(v) is False: logging.error(f"load role {k} failed!") return False - + role.role_id = self.owner_name + "." + k self.roles[k] = role return True diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 88a13ed..9c90056 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -33,9 +33,20 @@ class MessageFilter: return True +class LLMResult: + def __init__(self) -> None: + self.state : str = "ignore" + self.resp : str = "" + self.post_msgs = [] + self.send_msgs = [] + self.calls = [] + self.post_calls = [] + + class Workflow: def __init__(self) -> None: self.workflow_name : str = None + self.workflow_id : str = None self.rule_prompt : AgentPrompt = None self.workflow_config = None self.role_group : dict = None @@ -62,6 +73,11 @@ class Workflow: logger.error("workflow config must have name") return False self.workflow_name = config.get("name") + if self.owner_workflow is None: + self.workflow_id = self.workflow_name + else: + self.workflow_id = self.owner_workflow.workflow_id + "." + self.workflow_name + self.db_file = self.owner_workflow.db_file #if config.get("rule_prompt") is None: # logger.error("workflow config must have rule_prompt") @@ -74,13 +90,14 @@ class Workflow: logger.error("workflow config must have roles") return False self.role_group = AIRoleGroup() + self.role_group.owner_name = self.workflow_id if self.role_group.load_from_config(config.get("roles")) is False: logger.error("Workflow load role_group failed") return False - if config.get("input_filter") is not None: + if config.get("filter") is not None: self.input_filter = MessageFilter() - if self.input_filter.load_from_config(config.get("input_filter")) is False: + if self.input_filter.load_from_config(config.get("filter")) is False: logger.error("Workflow load input_filter failed") return False @@ -98,15 +115,52 @@ class Workflow: for k,v in config.items(): sub_workflow = Workflow() sub_workflow.set_owner(self) + if sub_workflow.load_from_config(v) is False: logger.error(f"load sub workflow {k} failed!") return False self.sub_workflows[k] = sub_workflow return True + + def _parse_msg_target(self,s:str)->list[str]: + return s.split(".") + + async def _forword_msg(self,inner_obj_id,msg): + i : int = 1 + current_workflow = self + while i < len(inner_obj_id): + if i == len(inner_obj_id) - 1: + the_role : AIRole = current_workflow.role_group.get(inner_obj_id[i]) + if the_role is not None: + return await current_workflow.role_process_msg(msg,the_role) + sub_workflow = current_workflow.sub_workflows.get(inner_obj_id[i]) + if sub_workflow is not None: + return await sub_workflow.process_msg(msg) + logger.error(f"{msg.target} not found! forword message failed!") + return None + else: + current_workflow = current_workflow.sub_workflows.get(inner_obj_id[i]) + if current_workflow is None: + logger.error(f"sub workflow {inner_obj_id[i]} not found!") + return None + + i += 1 + + logger.error(f"{msg.target} not found! forword message failed!") + return None + async def _process_msg(self,msg:AgentMsg): + # workflow can be a message handler, but never be a message sender + # all message forword to roles or sub workflow + # workflow no chatsession record, but role have + final_result = None - chatsession = None + + targets = self._parse_msg_target(msg.target) + if len(targets) > 1: + return await self._forword_msg(targets,msg) + if self.input_filter is not None: select_role_id = self.input_filter.select(msg) if select_role_id is not None: @@ -115,35 +169,206 @@ class Workflow: logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group") return None - result = await self._role_process_msg(msg,select_role) + result = await self.role_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: logger.error(f"input_filter return None for :{msg}") return else: + # no input filter, we would process all message, slowly,not suggest to use results = {} + final_result:AgentMsg = None for this_role in self.role_group.roles.values(): # TODO : we would do this in parallel - a_result = await self._role_process_msg(msg,this_role) + a_result = await self.role_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:AgentMsg = self._merge_msg_result(results) - if chatsession is not None: - chatsession.append_post(final_result) + final_result = a_result return final_result + @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 + for line in lines: + func_call = AgentMsg.parse_function_call(line) + if func_call: + func_args = func_call[1] + match func_call[0]: + case "sendmsg":# sendmsg($target_id,$msg_content) + if len(func_args) != 2: + logger.error(f"parse sendmsg failed! {func_call}") + continue + new_msg = AgentMsg() + target_id = func_args[0] + msg_content = func_args[1] + new_msg.set("_",target_id,msg_content) + + r.send_msgs.append(new_msg) + is_need_wait = True + continue + case "postmsg":# postmsg($target_id,$msg_content) + if len(func_args) != 2: + logger.error(f"parse postmsg failed! {func_call}") + continue + new_msg = AgentMsg() + target_id = func_args[1] + msg_content = func_args[2] + new_msg.set("_",target_id,msg_content) + r.post_msgs.append(new_msg) + continue + case "call":# call($func_name,$args_str) + r.calls.append(func_call) + is_need_wait = True + continue + case "post_call": # post_call($func_name,$args_str) + r.post_calls.append(func_call) + continue + + r.resp += line + "\n" + else: + r.resp += line + "\n" + + if is_need_wait: + r.state = "waiting" + else: + r.state = "reponsed" + + return r + + async def role_post_msg(self,msg:AgentMsg,the_role:AIRole): + msg.sender = the_role.get_role_id() + + target_role = self.role_group.get(msg.target) + if target_role: + msg.target = target_role.get_role_id() + logger.info(f"{msg.sender} post message {msg.id} to inner role: {msg.target}") + asyncio.create_task(self.role_process_msg(msg,target_role)) + return + + target_workflow = self.sub_workflows.get(msg.target) + if target_workflow: + msg.target = target_workflow.workflow_id + logger.info(f"{msg.sender} post message {msg.id} to sub workflow: {msg.target}") + asyncio.create_task(target_workflow._process_msg(msg)) + + logger.info(f"{msg.sender} post message {msg.id} to AIBus: {msg.target}") + await self.get_bus().post_message(msg.target,msg) + return + async def role_send_msg(self,msg:AgentMsg,the_role:AIRole): + msg.sender = the_role.get_role_id() + target_role = self.role_group.get(msg.target) + if target_role: + # msg.target = target_role.get_role_id() + logger.info(f"{msg.sender} send message {msg.id} to inner role: {msg.target}") + return await self.role_process_msg(msg,target_role) + + target_workflow = self.sub_workflows.get(msg.target) + if target_workflow: + # msg.target = target_workflow.workflow_id + logger.info(f"{msg.sender} send message {msg.id} to sub workflow: {msg.target}") + return await target_workflow._process_msg(msg) + + logger.info(f"{msg.sender} post message {msg.id} to AIBus: {msg.target}") + return await self.get_bus().send_message(msg.target,msg) + + async def role_call(self,call:tuple,the_role:AIRole): + logger.info(f"{the_role.role_id} call {call[0]} with args {call[1]}") + return """{result:"timeout"}""" + + async def role_post_call(self,call:tuple,the_role:AIRole): + logger.info(f"{the_role.role_id} post call {call[0]} with args {call[1]}") + return + + + async def role_process_msg(self,msg:AgentMsg,the_role:AIRole): + session_topic = f"{msg.sender}#{msg.topic}" + session_owner = the_role.get_role_id() + chatsession : AIChatSession = AIChatSession.get_session(session_owner,session_topic,self.db_file) + if chatsession is None: + logger.error(f"get session {session_topic}@{session_owner} failed!") + return None + + prompt = AgentPrompt() + prompt.append(the_role.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)) + + msg_prompt = AgentPrompt() + msg_prompt.messages = [{"role":"user","content":msg.body}] + prompt.append(msg_prompt) + + + async def _do_process_msg(): + #TODO: send msg to agent might be better? + result_str = await ComputeKernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + result = Workflow.prase_llm_result(result_str) + + + for postmsg in result.post_msgs: + postmsg.topic = msg.topic + await self.role_post_msg(the_role,postmsg) + + for post_call in result.post_calls: + await self.role_post_call(post_call,the_role) + + + result_prompt_str = "" + match result.state: + case "ignore": + return None + case "reponsed": + resp_msg = AgentMsg() + resp_msg.topic = msg.topic + resp_msg.set(session_owner,msg.sender,result.resp) + + chatsession.append_recv(msg) + chatsession.append_post(resp_msg) + return resp_msg + case "waiting": + # TODO: Use role:"function" would be better + for sendmsg in result.send_msgs: + target = sendmsg.target + sendmsg.topic = msg.topic + send_resp = await self.role_send_msg(sendmsg,the_role) + if send_resp is not None: + result_prompt_str += f"\n{target} response is :{send_resp.body}" + + for call in result.calls: + call_result = await self.role_call(call,the_role) + if call_result is not None: + result_prompt_str += f"\ncall {call[0]} result is :{call_result}" + + result_prompt = AgentPrompt() + result_prompt.messages = [{"role":"user","content":result_prompt_str}] + prompt.append(result_prompt) + r = await _do_process_msg() + return r + + return await _do_process_msg() + + + #obsolete async def _role_process_msg(self,msg:AgentMsg,the_role:AIRole) -> None: # TODO : we just record role's chatsession, but in future, we would record workflow's chatsession(like a groupo chat) session_topic = f"{the_role.get_name()}#{msg.sender}#{msg.topic}" @@ -162,7 +387,6 @@ class Workflow: # prompt.append(self._get_knowlege_prompt(the_role.get_name())) prompt.append(await self._get_prompt_from_session(chatsession)) - #prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context msg_prompt = AgentPrompt() msg_prompt.messages = [{"role":"user","content":msg.body}] @@ -180,7 +404,7 @@ class Workflow: 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) + 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 @@ -218,31 +442,18 @@ class Workflow: chatsession.append_post(resp_msg) return resp_msg - async def _pop_msg(self) -> AgentMsg: - pass - - def _get_chat_session_for_msg(self,msg:AgentMsg) -> AIChatSession: - pass async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt: messages = chatsession.read_history() # read last 10 message result_prompt = AgentPrompt() for msg in reversed(messages): - if msg.target == chatsession.owner_id: - result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"}) if msg.sender == chatsession.owner_id: result_prompt.messages.append({"role":"assistant","content":msg.body}) + else: + result_prompt.messages.append({"role":"user","content":f"{msg.body}"}) return result_prompt - - def _get_msg_queue(self,session_id:str): - pass - def _merge_msg_result(self,results:dict) -> AgentMsg: - # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg - for k,v in results.items(): - if v is not None: - return v def _get_function_prompt(self,role_name:str) -> AgentPrompt: pass @@ -250,48 +461,10 @@ class Workflow: def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt: pass - def _get_resp_prompt(self,resp:str,msg:AgentMsg,role:AIRole,prompt:AgentPrompt) -> AgentPrompt: - pass def get_workflow_rule_prompt(self) -> AgentPrompt: return self.rule_prompt - - def _get_llm_result_type(self,llm_resp_str:str) -> str: - if llm_resp_str == "ignore": - return "ignore" - - if llm_resp_str.find("sendmsg(") != -1: - return "send_message" - - if llm_resp_str.find("postmsg(") != -1: - return "post_message" - - if llm_resp_str.find("call(") != -1: - return "function" - - return "text" - def _parse_function_call_chain(self,llm_resp_str) -> CallChain: - pass - - def _parse_to_msg(self,llm_resp_str) -> AgentMsg: - lines = llm_resp_str.splitlines() - for line in lines: - if line.startswith("sendmsg("): - line = line[8:] - _index = line.find(",") - msg = AgentMsg() - msg.set("",line[:_index],line[_index+1:]) - return msg - - if line.startswith("postmsg("): - line = line[8:] - _index = line.find(",") - msg = AgentMsg() - msg.set("",line[:_index],line[_index+1:]) - return msg - - return None def get_workflow(self,workflow_name:str): """get workflow from known workflow list or sub workflow list""" diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index 28479dd..adb844a 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -25,6 +25,20 @@ class WorkflowManager: async def get_agent_default_workflow(self,agent_id:str) -> Workflow: pass + + async def _load_workflow_agents(self,workflow:Workflow) -> bool: + for v in workflow.role_group.roles.values(): + agent = await AgentManager().get(v.agent_name) + if agent is None: + logger.error(f"load agent {v.agent_name} failed!") + return False + v.agent = agent + + for sub_workflow in workflow.sub_workflows.values(): + if await self._load_workflow_agents(sub_workflow) is False: + return False + return True + async def get_workflow(self,workflow_id:str) -> Workflow: the_workflow : Workflow = self.loaded_workflow.get(workflow_id) if the_workflow: @@ -39,14 +53,9 @@ class WorkflowManager: if the_workflow is None: logger.warn(f"load workflow {workflow_id} from media failed!") - for v in the_workflow.role_group.roles.values(): - agent = await AgentManager().get(v.agent_name) - if agent is None: - logger.error(f"load agent {v.agent_name} failed!") - return None - v.agent = agent - - + if await self._load_workflow_agents(the_workflow) is False: + return None + return the_workflow async def _load_workflow_from_media(self,workflow_media:PackageMediaInfo) -> Workflow: @@ -64,10 +73,12 @@ class WorkflowManager: config_data = await config_file.read() config = toml.loads(config_data) result_workflow = Workflow() + result_workflow.db_file = self.db_file + if result_workflow.load_from_config(config) is False: logger.error(f"load workflow from {workflow_media} failed!") return None - result_workflow.db_file = self.db_file + return result_workflow except Exception as e: logger.error(f"read workflow.toml cfg from {workflow_media} failed! unexpected error occurred: {str(e)}") diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index cded7d0..3789957 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -41,16 +41,15 @@ class AIOS_Shell: self.current_topic = "default" async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool: - agent : AIAgent = await AgentManager().get(msg.target) + target_id = msg.target.split(".")[0] + agent : AIAgent = await AgentManager().get(target_id) if agent is not None: - bus.register_message_handler(msg.target,agent._process_msg) + bus.register_message_handler(target_id,agent._process_msg) return True - a_workflow = await WorkflowManager().get_workflow(msg.target) + a_workflow = await WorkflowManager().get_workflow(target_id) if a_workflow is not None: - bus.register_message_handler(msg.target,a_workflow._process_msg) - for subflow in a_workflow.sub_workflows.values(): - bus.register_message_handler(subflow.workflow_name,subflow._process_msg) + bus.register_message_handler(target_id,a_workflow._process_msg) return True return False @@ -79,7 +78,7 @@ class AIOS_Shell: agent_msg = AgentMsg() agent_msg.set(sender,target_id,msg) agent_msg.topic = topic - resp = await AIBus().get_default_bus().send_message(target_id,agent_msg) + resp = await AIBus.get_default_bus().send_message(target_id,agent_msg) if resp is not None: return resp.body else: @@ -111,10 +110,11 @@ class AIOS_Shell: case 'history': num = 10 offset = 0 - if len(args) >= 1: - num = args[0] - if len(args) >= 2: - offset = args[1] + if args is not None: + if len(args) >= 1: + num = args[0] + if len(args) >= 2: + offset = args[1] db_path = "" if await self.is_agent(self.current_target): @@ -140,37 +140,43 @@ class AIOS_Shell: history = FileHistory('history.txt') session = PromptSession(history=history) -def parse_function_call(s): - match = re.match(r'(\w+)\((.*)\)$', s) - if match: - func_name = match.group(1) - args_str = match.group(2) - - args = [] - buffer = '' - quote_count = 0 # Count of single or double quotes - for char in args_str: - - if char in ['"', "'"]: - quote_count += 1 - if char == ',' and quote_count % 2 == 0: # ',' is outside of quotes - args.append(buffer.strip()) - buffer = '' - else: - buffer += char - if buffer: - args.append(buffer.strip()) - - return func_name, args - else: +def parse_function_call(func_string): + match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string) + if not match: return None + + func_name = match.group(1) + params_string = match.group(2).strip() + params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string) + params = [param.strip('"') for param in params] + if len(params[0]) == 0: + params = None + + return func_name, params async def main(): print("aios shell prepareing...") - logging.basicConfig(filename="aios_shell.log",filemode="w",level=logging.INFO,format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s') + logging.basicConfig(filename="aios_shell.log",filemode="w",encoding='utf-8', + level=logging.INFO, + format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s') shell = AIOS_Shell("user") await shell.initial() + + s = """了解。首先，我们需要进行方案讨论，定出一致的活动策略。我将任务分解如下： +财务组： 分析可见的预算，提供一份合理并被执行的财务方案。 +行程预订组： 硅谷自然迷人的地方众多，寻找适合11人秋游的地点，以及一个可行的周末日期。策划一个一天或两天的安排，包括选择有南瓜园的地方，采摘苹果，参观当地的酗酒作坊等秋游活动。 +嘉宾对接组： 把个人的食品饮料或过敏食物的需求事先了解并计入行程内。 +酒店预订组：根据行程预订组的日期安排，活动时间在1天还是2天，在这个之内找一个合适的住宿，试着保持让住宿在预算边界以内。 +活动摄像组： 准备活动拍摄与剪辑方案。\n\n然后，我将把这些拆分后的任务发来小组。 +sendmsg(财务组,分析预算并提供一份财务方案.) +sendmsg(行程预订组,找出适合秋游的地方和日期.) +sendmsg(嘉宾对接组,了解个人的饮食需求.) +sendmsg(酒店预订组,查询并预订住宿.) +sendmsg(活动摄像组,提供活动拍摄方案.) +经过一两天的准备，一切就绪之后，我将向工作人员发送最后的行程计划， + """ + r = Workflow.prase_llm_result(s) print(f"aios shell {shell.get_version()} ready.") completer = WordCompleter(['send($target,$msg,$topic)', diff --git a/test/pkg.cfg.toml b/test/pkg.cfg.toml index 46acf0a..68c46d5 100644 --- a/test/pkg.cfg.toml +++ b/test/pkg.cfg.toml @@ -1,15 +1,83 @@ +# workflow实例场景 +# 能展示sub workflow +# 能展示env的整合 +# 能展示filter的使用 +# 能展示function的调用 +# 能展示基于工作目标/KPI的sub workflow迭代流程 +# 多人场景安排 -[authors] -"buckyos.org" = {public="aaaa",name="BuckyOS Core Dev Team"} -[authors."cyfs.com"] -public="bbbb" -name="CYFS Core Dev Team" +# 例子：举办一个团队活动 +# 方案讨论（通过交互引导的方式收集主人的需求）与确定 -[env] -is_strict = false -prefixs = ["default","../m/s","abc"] -obj2 = {k1="bb",k2=300} -[env.obj] -k1="aa" -k2=200 +# 活动前： +# 通讯员，对接管理参加活动的人的情况 (email spider) +# 酒店预订 简单：搜索（酒店评价） 处理异常 +# 行程（票务）预订 ：搜索，处理异常 +# 餐饮预订：给出方案，确定细节，预订 + +# 活动中： +# 进行统计和分析，调整设备 +# 安保，空调，音乐，拍照，录像 +# 响应紧急情况 +# 活动结束后： +# 整理照片，视频，进行必要的二次创作，发送给相关人员 +# 对活动进行总结，提出改件意见（指导下一次活动） + +# 1. 人员 +# 主管，负责和客户沟通，并对每个环境的结果进行总结 +# 嘉宾对接 +# 酒店组 +# 行程组 +# 财务组 +# 多媒体组 + + +[filter] +"*" = "manager" + +[roles.manager] +fullname = "经理" +agent="manager" +[[roles.manager.prompt]] +role="system" +content="""你是一个活动策划公司的经理,与客户对接并向团队下达指令。你的团队分为下面几个小组:嘉宾对接组,酒店预定组,行程预订组,财务组,活动摄像组。活动策划分为四个阶段：方案讨论，活动前，活动中，活动后。你会根据客户的需求，对团队进行分工，分别完成各个阶段的工作。你的基本工作模式是：\ +1. 收到客户的明确的指令后,基于客户的已有信息和客户商量活动方案，和活动策划公司无关的业务你会回答‘与我无关’。当和客户完成活动方案的确认后,你会将拆解后的任务分配给各个小组 \ +2. 根据目前已经确认的活动方案，你要根据时间适时的检查不同小组的工作情况。当收到小组的工作情况反馈后，你会站在全局的角度判断是否需要调整活动方案，如果需要调整，你会和客户商量重新确定方案，然后再将调整后的方案分配给各个小组。\ +3. 有时工作小组会主动与你沟通，反馈一些问题。你会站在全局的角度给与指导，适当的调整工作小组的工作目标。如果反馈的问题需要你和客户沟通，你会和客户沟通后重新确定方案。再将调整后的方案分配给受到影响各个小组。\ +4. 当你决定要和工作小组通信时，请使用`send_message({小组名称},{内容}`）的形式。"""}] + + + + +[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 = "嘉宾对接组组长" +[[sub_workflows."嘉宾对接组".roles.leader.prompt]] +role="system" +content="""你是一家活动策划公司的嘉宾对接组的组长，你的工作是基于已知信息,当前活动信息、公司经理的指令与嘉宾沟通，收集嘉宾的信息，然后将信息反馈给经理。在你看来，参加活动的多少有成员都是嘉宾，你可以通过你知道的信息给不同的成员进行分级。你的基本工作模式是：\ +1. 处理收到的邮件，如果邮件来自嘉宾，你会尝试从邮件的表态和内容中分享嘉宾的需要，并结合你对当前活动方案的理解判断是否需要和经理沟通，如果需要和经理沟通，你会将嘉宾的需求总结和告诉经理。不需要沟通的事项可以直接回复嘉宾。\ +2. 你总是通过`call_function(get_env,'parent.topic'`的形式查询当前的活动方案。等待函数返回后，你会根据函数的返回结果继续处理上一个对话。\ +3. 当你决定要和经理通信时，请使用`send_message(manager,{内容}`）的形式，内容的长度不超过200字。\ +4. 当你决定要回复嘉宾时，请使用`call_function(sendmail,{嘉宾邮件地址},{标题},{内容}）的形式，内容的长度不超过500字。""" +# 这里是孤立工作模式，组长只和经理沟通，也可以赋予其和其它组沟通的能力 + +[sub_workflows."酒店预定组"] +# 展现使用搜索引擎，并调用预订酒店的能力 +[sub_workflows."酒店预定组".environments.email] + +[sub_workflows."酒店预定组".roles.leader] +name="酒店预定组组长" +prompt = [{role="xxx",content="yyy"}] + +[sub_workflows."酒店预定组".roles.research] +name="酒店搜索专家" + +[sub_workflows."行程预订组"] +# 展现处理冲突并反推 +nam3="3"