gyt/opendan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved the system's exception handling logic, striving to let users see the cause of the error.

Browse Source
This commit is contained in:
Liu Zhicong
2023-09-26 22:50:50 -07:00
parent 857a7ed956
commit 943d0e3916
8 changed files with 109 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aios_kernel/agent.py
+25 -14
View File
@@ -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:
src/aios_kernel/agent_message.py
+12
View File
@@ -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()
src/aios_kernel/compute_kernel.py
-1
View File
@@ -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)
src/aios_kernel/compute_task.py
+3 -1
View File
@@ -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 = {}
src/aios_kernel/open_ai_node.py
+41 -25
View File
@@ -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):
src/aios_kernel/tg_tunnel.py
+1 -1
View File
@@ -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:
src/aios_kernel/workflow.py
+25 -10
View File
@@ -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: