From 5c2dd13ab2e8735e5673d05957be74ed7f241343 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 22 Sep 2023 00:09:21 +0800 Subject: [PATCH] Fix story maker bug --- rootfs/agents/fairy_tale_writer/agent.toml | 1 + rootfs/agents/manager/agent.toml | 2 + rootfs/agents/studio_director/agent.toml | 4 +- rootfs/workflows/story_maker/workflow.toml | 6 +- src/aios_kernel/agent.py | 78 +++++++------- src/aios_kernel/bus.py | 44 ++++---- src/aios_kernel/compute_kernel.py | 4 +- src/aios_kernel/open_ai_node.py | 23 +++-- src/aios_kernel/text_to_speech_function.py | 4 +- src/aios_kernel/workflow.py | 114 +++++++++++---------- 10 files changed, 147 insertions(+), 133 deletions(-) diff --git a/rootfs/agents/fairy_tale_writer/agent.toml b/rootfs/agents/fairy_tale_writer/agent.toml index 1f2256a..7955df6 100644 --- a/rootfs/agents/fairy_tale_writer/agent.toml +++ b/rootfs/agents/fairy_tale_writer/agent.toml @@ -1,6 +1,7 @@ instance_id = "fairy_tale_writer" fullname = "tracy wang" llm_model_name = "gpt-3.5-turbo-16k-0613" +enable_function = [] [[prompt]] role = "system" diff --git a/rootfs/agents/manager/agent.toml b/rootfs/agents/manager/agent.toml index 5bb2ba6..e4bba03 100644 --- a/rootfs/agents/manager/agent.toml +++ b/rootfs/agents/manager/agent.toml @@ -1,5 +1,7 @@ instance_id = "agent:xxxxxxabcde" fullname = "musk" +enable_function = [] + [[prompt]] role = "system" content = "你有丰富的管理技能，擅长将复杂工作拆解成简单的任务，让团队成员高效协作。" diff --git a/rootfs/agents/studio_director/agent.toml b/rootfs/agents/studio_director/agent.toml index eb61bca..65a7c07 100644 --- a/rootfs/agents/studio_director/agent.toml +++ b/rootfs/agents/studio_director/agent.toml @@ -1,7 +1,7 @@ instance_id = "studio_director" fullname = "tracy wang" -llm_model_name = "gpt-3.5-turbo-16k-0613" +enable_function = ["text_to_speech"] [[prompt]] role = "system" -content = "你是一个演播导演，请将下面故事改编成朗读剧本，提取旁白和角色台词，每个角色需要有性别、年龄、以及每句台词的语气。并调用text_to_speech function生成音频数据。" +content = "你是一个故事播音员，请将下面故事改编成播音剧本，提取旁白和角色台词，以及每个角色需要有性别、年龄、以及每句台词的语气，最后生成音频文件。" diff --git a/rootfs/workflows/story_maker/workflow.toml b/rootfs/workflows/story_maker/workflow.toml index 30a26d6..540665f 100644 --- a/rootfs/workflows/story_maker/workflow.toml +++ b/rootfs/workflows/story_maker/workflow.toml @@ -8,10 +8,12 @@ name = "story_maker" name = "manager" fullname = "总导演" agent="manager" +enable_function = [] + [[roles.manager.prompt]] role="system" content=""" -你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事。你的基本工作模式是： +你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事生成音频文件。你的基本工作模式是： 1. 收到客户的明确的指令后,让writer写出故事 2. 将writer写出的故事交给studio_director演播 3. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 @@ -25,6 +27,7 @@ content=""" name = "writer" agent = "fairy_tale_writer" fullname = "作家" +enable_function = [] [[roles.writer.prompt]] role="system" content="" @@ -32,6 +35,7 @@ content="" [roles.studio_director] name = "studio_director" agent = "studio_director" +enable_function = ["text_to_speech"] [[roles.studio_director.prompt]] role="system" content="" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index cc79637..653ac0a 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -26,7 +26,7 @@ class AgentPrompt: self.system_message = None def as_str(self)->str: - result_str = "" + result_str = "" if self.system_message: result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" if self.messages: @@ -34,18 +34,18 @@ class AgentPrompt: result_str += msg.get("role") + ":" + msg.get("content") + "\n" return result_str - + def to_message_list(self): result = [] if self.system_message: result.append(self.system_message) result.extend(self.messages) return result - + def append(self,prompt): if prompt is None: return - + if prompt.system_message is not None: if self.system_message is None: self.system_message = prompt.system_message @@ -99,9 +99,9 @@ class AIAgentTemplete: logger.error("load prompt from config failed!") return False - + return True - + class AIAgent: def __init__(self) -> None: @@ -111,7 +111,7 @@ class AIAgent: self.agent_id:str = None self.template_id:str = None self.fullname:str = None - self.powerby = None + self.powerby = None self.enable = True self.enable_kb = False self.enable_timestamp = False @@ -124,7 +124,7 @@ class AIAgent: self.owner_env : Environment = None self.owenr_bus = None self.enable_function_list = [] - + @classmethod def create_from_templete(cls,templete:AIAgentTemplete, fullname:str): # Agent just inherit from templete on craete,if template changed,agent will not change @@ -137,7 +137,7 @@ class AIAgent: 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!") @@ -188,7 +188,7 @@ class AIAgent: if llm_result_str == "ignore": r.state = "ignore" return r - + lines = llm_result_str.splitlines() is_need_wait = False @@ -205,7 +205,7 @@ class AIAgent: r.send_msgs.append(new_msg) is_need_wait = True - + case "post_msg":# postmsg($target_id,$msg_content) if len(func_args) != 1: logger.error(f"parse postmsg failed! {func_call}") @@ -215,22 +215,22 @@ class AIAgent: msg_content = func_item.body new_msg.set(self.agent_id,target_id,msg_content) r.post_msgs.append(new_msg) - + case "call":# call($func_name,$args_str) r.calls.append(func_item) is_need_wait = True return True case "post_call": # post_call($func_name,$args_str) r.post_calls.append(func_item) - return True - + return True + current_func : FunctionItem = None for line in lines: if line.startswith("##/"): if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + func_name,func_args = AgentMsg.parse_function_call(line[3:]) current_func = FunctionItem(func_name,func_args) else: @@ -238,11 +238,11 @@ class AIAgent: current_func.append_body(line + "\n") else: r.resp += line + "\n" - + if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + if len(r.send_msgs) > 0 or len(r.calls) > 0: r.state = "waiting" else: @@ -279,21 +279,22 @@ class AIAgent: def _get_inner_functions(self) -> dict: if self.owner_env is None: return None - + all_inner_function = self.owner_env.get_all_ai_functions() if all_inner_function is None: return None - + result_func = [] result_len = 0 for inner_func in all_inner_function: func_name = inner_func.get_name() - if self.enable_function_list: + if self.enable_function_list is not None: if len(self.enable_function_list) > 0: if func_name not in self.enable_function_list: logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}") continue - + else: + continue this_func = {} this_func["name"] = func_name this_func["description"] = inner_func.get_description() @@ -313,28 +314,29 @@ class AIAgent: func_node : AIFunction = self.owner_env.get_ai_function(func_name) if func_node is None: return "execute failed,function not found" - + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) try: result_str:str = await func_node.execute(**arguments) except Exception as e: - result_str = "call error:" + str(e) + result_str = "call error:" + str(e) logger.error(f"llm execute inner func:{func_name} error:{e}") - + + logger.info("llm execute inner func result:" + result_str) inner_functions,inner_function_len = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) - + ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() org_msg.inner_call_chain.append(ineternal_call_record) if stack_limit > 0: inner_func_call_node = task_result.result_message.get("function_call") - + if inner_func_call_node: - return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) + return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) else: return task_result.result_str @@ -344,7 +346,7 @@ class AIAgent: def _format_msg_by_env_value(self,prompt:AgentPrompt): if self.owner_env is None: return - + for msg in prompt.messages: old_content = msg.get("content") msg["content"] = old_content.format_map(self.owner_env) @@ -380,7 +382,7 @@ class AIAgent: system_prompt_len = prompt.get_prompt_token_len() input_len = len(msg.body) - + history_prmpt,history_token_len = await self._get_prompt_from_session(chatsession,system_prompt_len + function_token_len,input_len) prompt.append(history_prmpt) # chat context @@ -397,7 +399,7 @@ class AIAgent: if inner_func_call_node: #TODO to save more token ,can i use msg_prompt? final_result = await self._execute_func(inner_func_call_node,prompt,msg) - + llm_result : LLMResult = self._get_llm_result_type(final_result) is_ignore = False result_prompt_str = "" @@ -415,21 +417,21 @@ class AIAgent: agent_sesion = AIChatSession.get_session(self.agent_id,f"{sendmsg.target}#{sendmsg.topic}",self.chat_db) agent_sesion.append(sendmsg) agent_sesion.append(send_resp) - + final_result = llm_result.resp + result_prompt_str if is_ignore is not True: resp_msg = msg.create_resp_msg(final_result) chatsession.append(msg) chatsession.append(resp_msg) - + return resp_msg - + return None - + def get_id(self) -> str: return self.agent_id - + def get_fullname(self) -> str: return self.fullname @@ -438,14 +440,14 @@ class AIAgent: def get_llm_model_name(self) -> str: return self.llm_model_name - + def get_max_token_size(self) -> int: return self.max_token_size - + async def _get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len,is_groupchat=False) -> AgentPrompt: # TODO: get prompt from group chat is different from single chat history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len - messages = chatsession.read_history() # read + messages = chatsession.read_history() # read result_token_len = 0 result_prompt = AgentPrompt() read_history_msg = 0 diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 29fc186..16ede81 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -19,14 +19,14 @@ class AIBusHandler: async def handle_message(self,msg:AgentMsg) -> Any: if self.handler is None: return None - + resp_msg = await self.handler(msg) if self.enable_defualt_proc: if resp_msg is not None: await self.owner_bus.post_message(resp_msg,False) return resp_msg - + class AIBus: _instance = None @classmethod @@ -48,16 +48,16 @@ class AIBus: if msg.rely_msg_id is not None: handler.results[msg.rely_msg_id] = msg return None - + handler.queue.put_nowait(msg) self.start_process(target_id) return True - + if use_unhandle: if self.unhandle_handler is not None: if await self.unhandle_handler(self,msg): return await self.post_message(msg,False) - + logger.warn(f"post message to {msg.target} failed!,target not found") return False @@ -71,7 +71,7 @@ class AIBus: if sender_handler is None: logger.warn(f"sender {sender_id} not register on AI_BUS!") return None - + post_result = await self.post_message(msg) if post_result is False: return None @@ -84,37 +84,39 @@ class AIBus: msg.status = AgentMsgStatus.RESPONSED del sender_handler.results[msg.msg_id] return resp - + await asyncio.sleep(0.2) retry_times += 1 - if retry_times > 5*120: # default timeout is 120 sec + if retry_times > 5*240: # default timeout is 240 sec msg.status = AgentMsgStatus.ERROR return None - + return None - + def register_unhandle_message_handler(self,handler:Any) -> Queue: self.unhandle_handler = handler # means sub - def register_message_handler(self,handler_name:str,handler:Any) -> Queue: - handler_node = AIBusHandler(handler,self) + def register_message_handler(self,handler_name:str,handler:Any) -> Queue: + handler_node = AIBusHandler(handler,self) self.handlers[handler_name] = handler_node return handler_node.queue - + async def process_queue(self, handler:AIBusHandler): while True: # Wait for a message message = await handler.queue.get() - #try: + try: # Try to handle the message - await handler.handle_message(message) - #except Exception as e: + await handler.handle_message(message) + except Exception as e: # If an error occurs, put the message back into the queue - # logger.error(f"handle message {message.msg_id} failed! {e}") + logger.error(f"handle message {message.msg_id} failed! {e}") + logger.exception(e) + raise e #self.queues[name].put_nowait(message) - + return def start_process(self,target_name): @@ -122,12 +124,12 @@ class AIBus: if handler is None: logger.error(f"handler {target_name} not found!") return - + if handler.handler is None: return if handler.working_task is not None: logger.warn(f"handler {target_name} is already working!") return - - handler.working_task = asyncio.create_task(self.process_queue(handler)) \ No newline at end of file + + handler.working_task = asyncio.create_task(self.process_queue(handler)) diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 5859efe..fffc8ac 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -118,7 +118,7 @@ class ComputeKernel: if task_req.state == ComputeTaskState.ERROR: break - if check_times >= 20: + if check_times >= 120: task_req.state = ComputeTaskState.ERROR break @@ -129,7 +129,7 @@ class ComputeKernel: if task_req.state == ComputeTaskState.DONE: return task_req.result - return "error!" + raise Exception("error!") def text_embedding(self,input:str,model_name:Optional[str] = None): diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 475fa22..e610be7 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -20,7 +20,7 @@ class OpenAI_ComputeNode(ComputeNode): if cls._instance is None: cls._instance = OpenAI_ComputeNode() return cls._instance - + @classmethod def declare_user_config(cls): if os.getenv("OPENAI_API_KEY_") is None: @@ -46,7 +46,7 @@ class OpenAI_ComputeNode(ComputeNode): if self.openai_api_key is None: logger.error("openai_api_key is None!") return False - + openai.api_key = self.openai_api_key self.start() return True @@ -68,7 +68,7 @@ class OpenAI_ComputeNode(ComputeNode): resp = openai.Embedding.create(model=model_name, input=input) - + # resp = { # "object": "list", # "data": [ @@ -86,7 +86,7 @@ class OpenAI_ComputeNode(ComputeNode): logger.info(f"openai response: {resp}") - result = ComputeTaskResult() + result = ComputeTaskResult() result.set_from_task(task) result.worker_id = self.node_id result.result = resp["data"][0]["embedding"] @@ -100,23 +100,23 @@ class OpenAI_ComputeNode(ComputeNode): if max_token_size is None: max_token_size = 4000 - result_token = int(max_token_size * 0.4) - - logger.info(f"call openai {mode_name} prompts: {prompts}") + result_token = max_token_size if llm_inner_functions is None: + logger.info(f"call openai {mode_name} prompts: {prompts}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, max_tokens=result_token, temperature=0.7) else: + logger.info(f"call openai {mode_name} prompts: {prompts} functions: {json.dumps(llm_inner_functions)}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, functions=llm_inner_functions, max_tokens=result_token, temperature=0.7) # TODO: add temperature to task params? - + logger.info(f"openai response: {json.dumps(resp, indent=4)}") result = ComputeTaskResult() @@ -139,6 +139,7 @@ class OpenAI_ComputeNode(ComputeNode): result.result_message = resp["choices"][0]["message"] if token_usage: result.result_refers["token_usage"] = token_usage + logger.info(f"openai success response: {result.result_str}") return result case _: task.state = ComputeTaskState.ERROR @@ -148,7 +149,7 @@ class OpenAI_ComputeNode(ComputeNode): if self.is_start is True: return self.is_start = True - + async def _run_task_loop(): while True: task = await self.task_queue.get() @@ -171,13 +172,13 @@ class OpenAI_ComputeNode(ComputeNode): def is_support(self, task: ComputeTask) -> bool: - if task.task_type == ComputeTaskType.LLM_COMPLETION: + if task.task_type == ComputeTaskType.LLM_COMPLETION: if not task.params["model_name"]: return True model_name : str = task.params["model_name"] if model_name.startswith("gpt-"): return True - + if task.task_type == ComputeTaskType.TEXT_EMBEDDING: if task.params["model_name"] == "text-embedding-ada-002": return True diff --git a/src/aios_kernel/text_to_speech_function.py b/src/aios_kernel/text_to_speech_function.py index 3237974..2f7bff0 100644 --- a/src/aios_kernel/text_to_speech_function.py +++ b/src/aios_kernel/text_to_speech_function.py @@ -83,9 +83,9 @@ class TextToSpeechFunction(AIFunction): continue if audio is not None: - path = os.path.join(os.curdir, "{}.mp3".format(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))) + path = os.path.join(os.path.realpath(os.curdir), "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10)))) audio.export(path, format="mp3") - return "complete.file path:{}".format(path) + return "已经生成音频文件, 文件路径为{}".format(path) else: return "failed" diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 97893b6..00bb255 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -2,7 +2,7 @@ import logging import asyncio import json import os -import time +import time from asyncio import Queue from typing import Optional,Tuple,List from abc import ABC, abstractmethod @@ -32,7 +32,7 @@ class MessageFilter: # TODO: add more filter return None - + def load_from_config(self,config:dict) -> bool: self.filters = config return True @@ -68,8 +68,8 @@ class Workflow: def load_from_config(self,config:dict) -> bool: if config is None: return False - - if config.get("name") is None: + + if config.get("name") is None: logger.error("workflow config must have name") return False self.workflow_name = config.get("name") @@ -84,7 +84,7 @@ class Workflow: if self.rule_prompt.load_from_config(config.get("prompt")) is False: logger.error("Workflow load prompt failed") return False - + if config.get("roles") is None: logger.error("workflow config must have roles") return False @@ -106,13 +106,13 @@ class Workflow: self.env_db_file = self.owner_workflow.env_db_file self.workflow_env = WorkflowEnvironment(self.workflow_id,self.env_db_file) - env_ndoe = config.get("enviroment") + env_ndoe = config.get("enviroment") if env_ndoe is not None: if self._load_env_from_config(env_ndoe) is False: logger.error("Workflow load env failed") return False - connected_env_ndoe = config.get("connected_env") + connected_env_ndoe = config.get("connected_env") if connected_env_ndoe is not None: for _node in connected_env_ndoe: env_id = _node.get("env_id") @@ -124,13 +124,13 @@ class Workflow: logger.error(f"Workflow load connected_env failed, env {env_id} not found!") return False self.connect_to_environment(remote_env,_node.get("event2msg")) - + sub_workflows = config.get("sub_workflows") if sub_workflows is not None: if self._load_sub_workflows(sub_workflows) is False: logger.error("Workflow load sub workflows failed") return False - + return True def _load_env_from_config(self,config:dict) -> bool: @@ -147,7 +147,7 @@ class Workflow: return False self.sub_workflows[k] = sub_workflow return True - + def _parse_msg_target(self,s:str)->list[str]: return s.split(".") @@ -170,12 +170,12 @@ class Workflow: 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 - + def get_workflow_id_from_target(self,target:str) -> str: target_list = target.split(".") if len(target_list) == 0: @@ -203,11 +203,11 @@ class Workflow: #1. workflow start process message final_result = None - + # this is workflow's group_chat session session_topic = msg.sender + "#" + msg.topic chatsesssion = AIChatSession.get_session(self.workflow_id,session_topic,self.db_file) - + #2. find role by msg.mentions or workflow's selector logic if msg.mentions is not None: if not self.workflow_id in msg.mentions: @@ -219,20 +219,20 @@ class Workflow: this_role = self.role_group.get(mention) if this_role is not None: return await self.role_process_msg(msg,this_role,chatsesssion) - + if self.input_filter is not None: select_role_id = self.input_filter.select(msg) - if select_role_id is not None: + if select_role_id is not None: select_role = self.role_group.get(select_role_id) if select_role is None: logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group") return None - + return await self.role_process_msg(msg,select_role,chatsesssion) else: logger.error(f"input_filter return None for :{msg.body}") return None - + logger.error(f"{self.workflow_id}:no role can process this msg:{msg.body}") return final_result @@ -245,7 +245,7 @@ class Workflow: if llm_result_str == "ignore": r.state = "ignore" return r - + lines = llm_result_str.splitlines() is_need_wait = False @@ -262,7 +262,7 @@ class Workflow: r.send_msgs.append(new_msg) is_need_wait = True - + case "post_msg":# postmsg($target_id,$msg_content) if len(func_args) != 1: logger.error(f"parse postmsg failed! {func_call}") @@ -272,22 +272,22 @@ class Workflow: msg_content = func_item.body new_msg.set("_",target_id,msg_content) r.post_msgs.append(new_msg) - + case "call":# call($func_name,$args_str) r.calls.append(func_item) is_need_wait = True return True case "post_call": # post_call($func_name,$args_str) r.post_calls.append(func_item) - return True - + return True + current_func : FunctionItem = None for line in lines: if line.startswith("##/"): if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + func_name,func_args = AgentMsg.parse_function_call(line[3:]) current_func = FunctionItem(func_name,func_args) else: @@ -295,7 +295,7 @@ class Workflow: current_func.append_body(line + "\n") else: r.resp += line + "\n" - + if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() @@ -309,14 +309,14 @@ class Workflow: async def role_post_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): 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.msg_id} to inner role: {msg.target}") asyncio.create_task(self.role_process_msg(msg,target_role,workflow_chat_session)) return - + target_workflow = self.sub_workflows.get(msg.target) if target_workflow: msg.target = target_workflow.workflow_id @@ -341,7 +341,7 @@ class Workflow: # msg.target = target_workflow.workflow_id logger.info(f"{msg.sender} send message {msg.msg_id} to sub workflow: {msg.target}") return await target_workflow._process_msg(msg) - + logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}") return await self.get_bus().send_message(msg) @@ -352,8 +352,8 @@ class Workflow: func_node : AIFunction = self.workflow_env.get_ai_function(func_item.name) if func_node is None: return "execute failed,function not found" - - result_str:str = await func_node.execute(**arguments) + + result_str:str = await func_node.execute(**arguments) return result_str async def role_post_call(self,func_item:FunctionItem,the_role:AIRole): @@ -363,7 +363,7 @@ class Workflow: def _format_msg_by_env_value(self,prompt:AgentPrompt): if self.workflow_env is None: return - + for msg in prompt.messages: old_content = msg.get("content") msg["content"] = old_content.format_map(self.workflow_env) @@ -372,15 +372,17 @@ class Workflow: all_inner_function = self.workflow_env.get_all_ai_functions() if all_inner_function is None: return None - + result_func = [] for inner_func in all_inner_function: func_name = inner_func.get_name() - if the_role.enable_function_list: + if the_role.enable_function_list is not None: if len(the_role.enable_function_list) > 0: if func_name not in the_role.enable_function_list: - logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}") + logger.debug(f"agent {self.agent_id} ignore inner func:{func_name}") continue + else: + continue this_func = {} this_func["name"] = func_name this_func["description"] = inner_func.get_description() @@ -399,17 +401,17 @@ class Workflow: func_node : AIFunction = self.workflow_env.get_ai_function(func_name) if func_node is None: return "execute failed,function not found" - + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) result_str:str = await func_node.execute(**arguments) - - inner_functions = self._get_inner_functions() + + inner_functions = self._get_inner_functions(the_role) prompt.messages.append({"role":"function","content":result_str,"name":func_name}) task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt, the_role.agent.llm_model_name,the_role.agent.max_token_size, inner_functions) - + ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() org_msg.inner_call_chain.append(ineternal_call_record) @@ -419,13 +421,13 @@ class Workflow: return await self._role_execute_func(the_role,inner_func_call_node,prompt,org_msg,stack_limit-1) else: return task_result.result_str - + def _is_in_same_workflow(self,msg) -> bool: pass - async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): + async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): msg.target = the_role.get_role_id() - + prompt = AgentPrompt() prompt.append(the_role.agent.prompt) @@ -433,7 +435,7 @@ class Workflow: prompt.append(the_role.get_prompt()) # prompt.append(self._get_function_prompt(the_role.get_name())) # prompt.append(self._get_knowlege_prompt(the_role.get_name())) - + #support group chat, user content include sender name! prompt.append(await self._get_prompt_from_session(workflow_chat_session)) @@ -443,15 +445,15 @@ class Workflow: self._format_msg_by_env_value(prompt) inner_functions = self._get_inner_functions(the_role) - + async def _do_process_msg(): #TODO: send msg to agent might be better? task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size(),inner_functions) result_str = task_result.result_str logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}") - + inner_func_call_node = task_result.result_message.get("function_call") - + if inner_func_call_node: #TODO to save more token ,can i use msg_prompt? result_str = await self._role_execute_func(the_role,inner_func_call_node,prompt,msg) @@ -461,7 +463,7 @@ class Workflow: postmsg.prev_msg_id = msg.get_msg_id() # might be craete a new msg.topic for this postmsg postmsg.topic = msg.topic - + await self.role_post_msg(postmsg,the_role,workflow_chat_session) if not self._is_in_same_workflow(postmsg): role_sesion = AIChatSession.get_session(the_role.get_role_id(),f"{postmsg.target}#{msg.topic}",self.db_file) @@ -469,14 +471,14 @@ class Workflow: else: # message will be saved in role.process_message pass - - + + for post_call in result.post_calls: action_msg = msg.create_action_msg(post_call[0],post_call[1],the_role.get_role_id()) workflow_chat_session.append(action_msg) await self.role_post_call(post_call,the_role) #save post_call - + result_prompt_str = "" match result.state: case "ignore": @@ -506,11 +508,11 @@ class Workflow: else: # message will be saved in role.process_message pass - + for call in result.calls: action_msg = msg.create_action_msg(call[0],call[1],call_result,the_role.get_role_id) 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}" #save action @@ -522,7 +524,7 @@ class Workflow: prompt.append(result_prompt) r = await _do_process_msg() return r - + return await _do_process_msg() async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt: @@ -533,9 +535,9 @@ class Workflow: 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_knowlege_prompt(self,role_name:str) -> AgentPrompt: pass @@ -557,7 +559,7 @@ class Workflow: # if k == "role": # continue # else: - # + # # def _env_msg_handler(env_event:EnvironmentEvent) -> None: # the_msg:AgentMsg= self._env_event_to_msg(env_event) # self.role_post_msg