diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 79e67e6..269ba85 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -11,7 +11,7 @@ import datetime from .agent_message import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult from .chatsession import AIChatSession -from .compute_task import ComputeTaskResult +from .compute_task import ComputeTaskResult,ComputeTaskResultCode from .ai_function import AIFunction from .environment import Environment from .contact_manager import ContactManager,Contact,FamilyMember @@ -305,7 +305,7 @@ class AIAgent: return result_func,result_len - async def _execute_func(self,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> str: + async def _execute_func(self,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> [str,int]: from .compute_kernel import ComputeKernel func_name = inenr_func_call_node.get("name") @@ -314,21 +314,24 @@ 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) - logger.error(f"llm execute inner func:{func_name} error:{e}") + result_str = f"execute {func_name} error,function not found" + else: + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + try: + result_str:str = await func_node.execute(**arguments) + except Exception as e: + result_str = f"execute {func_name} 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) - + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + return task_result.error_str,1 + 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) @@ -339,7 +342,7 @@ class AIAgent: if inner_func_call_node: return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) else: - return task_result.result_str + return task_result.result_str,0 async def _get_agent_prompt(self) -> AgentPrompt: return self.prompt @@ -394,12 +397,20 @@ class AIAgent: logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + error_resp = msg.create_error_resp(task_result.error_str) + return error_resp + final_result = task_result.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? - final_result = await self._execute_func(inner_func_call_node,prompt,msg) + final_result,error_code = await self._execute_func(inner_func_call_node,prompt,msg) + if error_code != 0: + error_resp = msg.create_error_resp(final_result) + return error_resp llm_result : LLMResult = self._get_llm_result_type(final_result) is_ignore = False @@ -455,7 +466,7 @@ class AIAgent: for msg in reversed(messages): read_history_msg += 1 dt = datetime.datetime.fromtimestamp(float(msg.create_time)) - formatted_time = dt.strftime('%m-%d %H:%M:%S') + formatted_time = dt.strftime('%y-%m-%d %H:%M:%S') if msg.sender == self.agent_id: diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py index 1fcaf44..d597492 100644 --- a/src/aios_kernel/agent_message.py +++ b/src/aios_kernel/agent_message.py @@ -88,6 +88,18 @@ class AgentMsg: msg.sender = caller return msg + def create_error_resp(self,error_msg:str): + resp_msg = AgentMsg(AgentMsgType.TYPE_SYSTEM) + resp_msg.create_time = time.time() + + resp_msg.rely_msg_id = self.msg_id + resp_msg.body = error_msg + resp_msg.topic = self.topic + resp_msg.sender = self.target + resp_msg.target = self.sender + + return resp_msg + def create_resp_msg(self,resp_body): resp_msg = AgentMsg() resp_msg.create_time = time.time() diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 0caab10..bd3eb07 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -137,7 +137,6 @@ class ComputeKernel: return await self._send_task(task_req) - def text_embedding(self,input:str,model_name:Optional[str] = None): task_req = ComputeTask() task_req.set_text_embedding_params(input,model_name) diff --git a/src/aios_kernel/compute_task.py b/src/aios_kernel/compute_task.py index dfc58ad..05125ee 100644 --- a/src/aios_kernel/compute_task.py +++ b/src/aios_kernel/compute_task.py @@ -6,7 +6,8 @@ import time class ComputeTaskResultCode(Enum): OK = 0 TIMEOUT = 1 - NO_WORK = 2 + NO_WORKER = 2 + ERROR = 3 class ComputeTaskState(Enum): @@ -91,6 +92,7 @@ class ComputeTaskResult: self.task_id: str = None self.callchain_id: str = None self.worker_id: str = None + self.error_str : str = None self.result_code: int = 0 self.result_str: str = None # easy to use,can read from result self.result_message: dict = {} diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index bf226df..e16d76d 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -1,4 +1,3 @@ - import openai import os import asyncio @@ -6,7 +5,7 @@ from asyncio import Queue import logging import json -from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType +from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType,ComputeTaskResultCode from .compute_node import ComputeNode from .storage import AIStorage,UserConfig @@ -60,15 +59,26 @@ class OpenAI_ComputeNode(ComputeNode): def _run_task(self, task: ComputeTask): task.state = ComputeTaskState.RUNNING + + result = ComputeTaskResult() + result.result_code = ComputeTaskResultCode.ERROR + result.set_from_task(task) + match task.task_type: case ComputeTaskType.TEXT_EMBEDDING: model_name = task.params["model_name"] input = task.params["input"] logger.info(f"call openai {model_name} input: {input}") - - resp = openai.Embedding.create(model=model_name, + try: + resp = openai.Embedding.create(model=model_name, input=input) - + except Exception as e: + logger.error(f"openai run TEXT_EMBEDDING task error: {e}") + task.state = ComputeTaskState.ERROR + task.error_str = str(e) + result.error_str = str(e) + return result + # resp = { # "object": "list", # "data": [ @@ -85,9 +95,8 @@ class OpenAI_ComputeNode(ComputeNode): # } logger.info(f"openai response: {resp}") - - result = ComputeTaskResult() - result.set_from_task(task) + task.state = ComputeTaskState.DONE + result.result_code = ComputeTaskResultCode.OK result.worker_id = self.node_id result.result = resp["data"][0]["embedding"] @@ -101,27 +110,29 @@ class OpenAI_ComputeNode(ComputeNode): max_token_size = 4000 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, + try: + if llm_inner_functions is None: + logger.info(f"call openai {mode_name} prompts: {prompts}") + 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? - + 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? + except Exception as e: + logger.error(f"openai run LLM_COMPLETION task error: {e}") + task.state = ComputeTaskState.ERROR + task.error_str = str(e) + result.error_str = str(e) + return result logger.info(f"openai response: {json.dumps(resp, indent=4)}") - result = ComputeTaskResult() - result.set_from_task(task) - status_code = resp["choices"][0]["finish_reason"] token_usage = resp.get("usage") match status_code: @@ -132,8 +143,11 @@ class OpenAI_ComputeNode(ComputeNode): case _: task.state = ComputeTaskState.ERROR task.error_str = f"The status code was {status_code}." - return None + result.error_str = f"The status code was {status_code}." + result.result_code = ComputeTaskResultCode.ERROR + return result + result.result_code = ComputeTaskResultCode.OK result.worker_id = self.node_id result.result_str = resp["choices"][0]["message"]["content"] result.result_message = resp["choices"][0]["message"] @@ -143,6 +157,8 @@ class OpenAI_ComputeNode(ComputeNode): return result case _: task.state = ComputeTaskState.ERROR + task.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" + result.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" return None def start(self): diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index 442deeb..eb732bb 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/src/aios_kernel/tg_tunnel.py @@ -167,7 +167,7 @@ class TelegramTunnel(AgentTunnel): resp_msg = await self.ai_bus.send_message(agent_msg) logger.info(f"process message {agent_msg.msg_id} from {agent_msg.sender} to {agent_msg.target}") if resp_msg is None: - await update.message.reply_text(f"{self.target_id} process message error") + await update.message.reply_text(f"System Error: Timeout,{self.target_id} no resopnse! Please check logs/aios.log for more details!") else: if resp_msg.body_mime is None: if resp_msg.body is not None: diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index f85d153..ed79b74 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -14,7 +14,7 @@ from .chatsession import AIChatSession from .role import AIRole,AIRoleGroup from .ai_function import AIFunction,FunctionItem from .compute_kernel import ComputeKernel -from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState +from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskResultCode from .bus import AIBus from .workflow_env import WorkflowEnvironment @@ -392,25 +392,32 @@ class Workflow: return result_func return None - async def _role_execute_func(self,the_role:AIRole,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> str: + async def _role_execute_func(self,the_role:AIRole,inenr_func_call_node:dict,prompt:AgentPrompt,org_msg:AgentMsg,stack_limit = 5) -> [str,int]: from .compute_kernel import ComputeKernel func_name = inenr_func_call_node.get("name") arguments = json.loads(inenr_func_call_node.get("arguments")) - - 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) + func_node : AIFunction = self.workflow_env.get_ai_function(func_name) + result_str : str = "" + if func_node is None: + result_str = f"execute {func_name} failed,function not found" + else: + try: + result_str = await func_node.execute(**arguments) + except Exception as e: + result_str = f"execute {func_name} error:{str(e)}" + logger.error(f"llm execute inner func:{func_name} error:{e}") - result_str:str = await func_node.execute(**arguments) 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) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + return task_result.error_str,1 ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() @@ -420,7 +427,7 @@ class Workflow: if inner_func_call_node: return await self._role_execute_func(the_role,inner_func_call_node,prompt,org_msg,stack_limit-1) else: - return task_result.result_str + return task_result.result_str,0 def _is_in_same_workflow(self,msg) -> bool: pass @@ -449,6 +456,11 @@ class Workflow: 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) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + error_resp = msg.create_error_resp(task_result.error_str) + return error_resp + result_str = task_result.result_str logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}") @@ -456,7 +468,10 @@ class Workflow: 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) + result_str,r_code = await self._role_execute_func(the_role,inner_func_call_node,prompt,msg) + if r_code != 0: + error_resp = msg.create_error_resp(result_str) + return error_resp result = Workflow.prase_llm_result(result_str) for postmsg in result.post_msgs: diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index da5e7ee..9b8731a 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -601,11 +601,9 @@ async def main(): format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') is_daemon = False + logger.info(f"Check Host OS :{os.name}") if os.name != 'nt': - if os.getppid() == 1: - is_daemon = True - if not sys.stdout.isatty(): - is_daemon = True + is_daemon = os.fstat(0) != os.fstat(1) or os.fstat(0) != os.fstat(2) shell = AIOS_Shell("user") shell.declare_all_user_config()