From 8739bf6a7628006f77a8822fcadc57fde455dae6 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Wed, 6 Dec 2023 13:31:05 -0800 Subject: [PATCH] 1) Do some rename refactor ,prepare for LLMProcess refactor 2) Fix merge bugs. --- .gitignore | 1 + doc/promps/Check TODO.md | 2 + doc/promps/Do TODO.md | 3 + src/[44 => doc/promps/Introspection.md | 0 doc/promps/Review Task.md | 5 + doc/promps/process message.md | 2 + doc/promps/self-learn.md | 0 doc/workflow.drawio | 395 -------------- rootfs/agents/Jarvis/agent.toml | 1 - rootfs/agents/TextSummary/agent.py | 12 +- src/aios/__init__.py | 9 +- src/aios/agent/agent.py | 92 ++-- src/aios/agent/agent_base.py | 415 +-------------- src/aios/agent/behavior.py | 4 - src/aios/agent/llm_process.py | 149 ++++++ src/aios/agent/role.py | 8 +- src/aios/agent/workflow.py | 30 +- src/aios/environment/asr_function.py | 2 +- .../environment/code_interpreter_function.py | 2 +- .../duckduckgo_text_search_function.py | 2 +- src/aios/environment/environment.py | 2 +- src/aios/environment/image_2_text_function.py | 2 +- .../environment/script_to_speech_function.py | 2 +- src/aios/environment/sql_database_function.py | 2 +- .../environment/text_to_speech_function.py | 2 +- src/aios/environment/workflow_env.py | 4 +- src/aios/environment/workspace_env.py | 14 +- src/aios/frame/compute_kernel.py | 12 +- src/aios/proto/agent_msg.py | 7 +- src/aios/proto/agent_task.py | 221 ++++++++ src/aios/{agent => proto}/ai_function.py | 2 +- src/aios/proto/compute_task.py | 171 +++++- src/aios_kernel/code_interpreter.py | 426 --------------- src/aios_kernel/code_interpreter_function.py | 41 -- .../duckduckgo_text_search_function.py | 52 -- src/aios_kernel/sql_database.py | 493 ------------------ src/aios_kernel/sql_database_function.py | 112 ---- src/component/common_environment/shell.py | 2 +- src/component/mail_environment/issue.py | 4 +- src/component/openai_node/open_ai_node.py | 5 +- src/component/tg_tunnel.py | 8 +- .../workflow_manager/workflow_manager.py | 2 +- src/requirements.txt | 8 +- src/service/aios_shell/aios_shell.py | 8 +- 44 files changed, 693 insertions(+), 2043 deletions(-) create mode 100644 doc/promps/Check TODO.md create mode 100644 doc/promps/Do TODO.md rename src/[44 => doc/promps/Introspection.md (100%) create mode 100644 doc/promps/Review Task.md create mode 100644 doc/promps/process message.md create mode 100644 doc/promps/self-learn.md delete mode 100644 doc/workflow.drawio delete mode 100644 src/aios/agent/behavior.py create mode 100644 src/aios/agent/llm_process.py create mode 100644 src/aios/proto/agent_task.py rename src/aios/{agent => proto}/ai_function.py (98%) delete mode 100644 src/aios_kernel/code_interpreter.py delete mode 100644 src/aios_kernel/code_interpreter_function.py delete mode 100644 src/aios_kernel/duckduckgo_text_search_function.py delete mode 100644 src/aios_kernel/sql_database.py delete mode 100644 src/aios_kernel/sql_database_function.py diff --git a/.gitignore b/.gitignore index 3bb2bad..09bdba0 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ math_school_env.db workflows.db +rootfs/test_doc/ diff --git a/doc/promps/Check TODO.md b/doc/promps/Check TODO.md new file mode 100644 index 0000000..d164691 --- /dev/null +++ b/doc/promps/Check TODO.md @@ -0,0 +1,2 @@ +# Check (TODO) +目的是根据Todo Log, 结合自己的角色检查TODO是否争取完成（非客观性TODO给出是否有所改进的评价） diff --git a/doc/promps/Do TODO.md b/doc/promps/Do TODO.md new file mode 100644 index 0000000..7e13f04 --- /dev/null +++ b/doc/promps/Do TODO.md @@ -0,0 +1,3 @@ +# Do (TODO) + 目标是结合 角色定义，手头的工具，已知知识 完成一个确定的任务。 + 完成任务时应使用ReAct的方法：应在给出执行动作前，先自言自语的输出一个计划，然后在动作（这个自言自语会变成TODO Logs） \ No newline at end of file diff --git a/src/[44 b/doc/promps/Introspection.md similarity index 100% rename from src/[44 rename to doc/promps/Introspection.md diff --git a/doc/promps/Review Task.md b/doc/promps/Review Task.md new file mode 100644 index 0000000..30072db --- /dev/null +++ b/doc/promps/Review Task.md @@ -0,0 +1,5 @@ +# Review (Task/Todo) +目的是结合已知信息（重点是已经进行操作的记录），对失败的，完成的不好的任务进行思考，尝试给出更好的解决方案 +1. 管理学方法：更换负责人 +2. 管理学方法：拆分 +3. 给出建议（该建议可以在下次一次DO-Check）循环中被使用 \ No newline at end of file diff --git a/doc/promps/process message.md b/doc/promps/process message.md new file mode 100644 index 0000000..6d44b34 --- /dev/null +++ b/doc/promps/process message.md @@ -0,0 +1,2 @@ +# Process Message +处理消息的首要是目的是分析消息的意图，并给予回复 \ No newline at end of file diff --git a/doc/promps/self-learn.md b/doc/promps/self-learn.md new file mode 100644 index 0000000..e69de29 diff --git a/doc/workflow.drawio b/doc/workflow.drawio deleted file mode 100644 index 0cc08b1..0000000 --- a/doc/workflow.drawio +++ /dev/null @@ -1,395 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/rootfs/agents/Jarvis/agent.toml b/rootfs/agents/Jarvis/agent.toml index fd4ab09..efee242 100644 --- a/rootfs/agents/Jarvis/agent.toml +++ b/rootfs/agents/Jarvis/agent.toml @@ -5,7 +5,6 @@ max_token_size = 128000 enable_timestamp = "true" owner_prompt = "I am your master{name}" contact_prompt = "I am your master's friend{name}" -owner_env = "calender" [[prompt]] role = "system" diff --git a/rootfs/agents/TextSummary/agent.py b/rootfs/agents/TextSummary/agent.py index 831b786..1f01e81 100644 --- a/rootfs/agents/TextSummary/agent.py +++ b/rootfs/agents/TextSummary/agent.py @@ -1,6 +1,6 @@ import copy -from aios.agent.agent_base import CustomAIAgent, AgentPrompt +from aios.agent.agent_base import CustomAIAgent, LLMPrompt from aios.knowledge.data.writer import split_text from aios.proto.agent_msg import AgentMsg, AgentMsgType from aios.proto.compute_task import ComputeTaskResultCode @@ -19,10 +19,10 @@ class TextSummaryAgent(CustomAIAgent): chunks = split_text(msg.body, separators=["\n\n", "\n"], chunk_size=4000, chunk_overlap=200, length_function=len) - prompt = AgentPrompt() - prompt.system_message = {"role":"system","content":"Your job is to generate a summary based on the input."} + prompt = LLMPrompt() + prompt.system_message = "Your job is to generate a summary based on the input." if len(chunks) == 1: - prompt.append(AgentPrompt(chunks[0])) + prompt.append(LLMPrompt(chunks[0])) resp = await self.do_llm_complection(prompt) if resp.result_code != ComputeTaskResultCode.OK: return msg.create_error_resp(resp.error_str) @@ -31,14 +31,14 @@ class TextSummaryAgent(CustomAIAgent): segments = [] for i, chunk in enumerate(chunks): seg_prompt = copy.deepcopy(prompt) - seg_prompt.append(AgentPrompt(chunk)) + seg_prompt.append(LLMPrompt(chunk)) resp = await self.do_llm_complection(seg_prompt) if resp.result_code != ComputeTaskResultCode.OK: return msg.create_error_resp(resp.error_str) segments.append(resp.result_str) segments_str = "\n".join(segments) - prompt.append(AgentPrompt(f"以下文本分段之后的各段摘要，请合并生成一个完整摘要：\n{segments_str}")) + prompt.append(LLMPrompt(f"以下文本分段之后的各段摘要，请合并生成一个完整摘要：\n{segments_str}")) resp = await self.do_llm_complection(prompt) if resp.result_code != ComputeTaskResultCode.OK: return msg.create_error_resp(resp.error_str) diff --git a/src/aios/__init__.py b/src/aios/__init__.py index d99ae12..aa01d69 100644 --- a/src/aios/__init__.py +++ b/src/aios/__init__.py @@ -1,13 +1,14 @@ from .proto.agent_msg import * from .proto.compute_task import * +from .proto.ai_function import * +from .proto.agent_task import * -from .agent.agent_base import AgentPrompt,CustomAIAgent, AgentTodo +from .agent.agent_base import * from .agent.chatsession import AIChatSession from .agent.agent import AIAgent,AIAgentTemplete, BaseAIAgent from .agent.role import AIRole,AIRoleGroup -# from .agent.workflow import Workflow -from .agent.ai_function import SimpleAIFunction, SimpleAIOperation +from .agent.workflow import Workflow from .frame.compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType from .frame.compute_node import ComputeNode,LocalComputeNode @@ -20,7 +21,7 @@ from .environment.environment import BaseEnvironment,SimpleEnvironment,Composite # from .environment.workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent,PaintEnvironment from .environment.text_to_speech_function import TextToSpeechFunction from .environment.image_2_text_function import Image2TextFunction -from .environment.workspace_env import WorkspaceEnvironment,TodoListEnvironment,TodoListType +from .environment.workspace_env import WorkspaceEnvironment from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index d0693af..e79d374 100644 --- a/src/aios/agent/agent.py +++ b/src/aios/agent/agent.py @@ -13,10 +13,12 @@ import copy import sys from ..proto.agent_msg import AgentMsg +from ..proto.ai_function import * +from ..proto.agent_task import * +from ..proto.compute_task import * from .agent_base import * from .chatsession import * -from .ai_function import * from ..environment.workspace_env import WorkspaceEnvironment, TodoListType from ..frame.contact_manager import ContactManager,Contact,FamilyMember @@ -69,7 +71,7 @@ class AIAgentTemplete: self.template_id:str = None self.introduce:str = None self.author:str = None - self.prompt:AgentPrompt = None + self.prompt:LLMPrompt = None def load_from_config(self,config:dict) -> bool: if config.get("llm_model_name") is not None: @@ -79,7 +81,7 @@ class AIAgentTemplete: if config.get("template_id") is not None: self.template_id = config["template_id"] if config.get("prompt") is not None: - self.prompt = AgentPrompt() + self.prompt = LLMPrompt() if self.prompt.load_from_config(config["prompt"]) is False: logger.error("load prompt from config failed!") return False @@ -90,9 +92,9 @@ class AIAgentTemplete: class AIAgent(BaseAIAgent): def __init__(self) -> None: - self.role_prompt:AgentPrompt = None - self.agent_prompt:AgentPrompt = None - self.agent_think_prompt:AgentPrompt = None + self.role_prompt:LLMPrompt = None + self.agent_prompt:LLMPrompt = None + self.agent_think_prompt:LLMPrompt = None self.llm_model_name:str = None self.max_token_size:int = 128000 self.agent_energy = 15 @@ -149,26 +151,26 @@ class AIAgent(BaseAIAgent): self.enable_thread = bool(config["enable_thread"]) if config.get("prompt") is not None: - self.agent_prompt = AgentPrompt() + self.agent_prompt = LLMPrompt() self.agent_prompt.load_from_config(config["prompt"]) if config.get("think_prompt") is not None: - self.agent_think_prompt = AgentPrompt() + self.agent_think_prompt = LLMPrompt() self.agent_think_prompt.load_from_config(config["think_prompt"]) def load_todo_config(todo_type:str) -> bool: todo_config = config.get(todo_type) if todo_config is not None: if todo_config.get("do") is not None: - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.load_from_config(todo_config["do"]) self.todo_prompts[todo_type]["do"] = prompt if todo_config.get("check") is not None: - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.load_from_config(todo_config["check"]) self.todo_prompts[todo_type]["check"] = prompt if todo_config.get("review_prompt") is not None: - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.load_from_config(todo_config["review_prompt"]) self.todo_prompts[todo_type]["review"] = prompt @@ -224,16 +226,16 @@ class AIAgent(BaseAIAgent): def get_max_token_size(self) -> int: return self.max_token_size - def get_agent_role_prompt(self) -> AgentPrompt: + def get_agent_role_prompt(self) -> LLMPrompt: return self.role_prompt - def _get_remote_user_prompt(self,remote_user:str) -> AgentPrompt: + def _get_remote_user_prompt(self,remote_user:str) -> LLMPrompt: cm = ContactManager.get_instance() contact = cm.find_contact_by_name(remote_user) if contact is None: #create guest prompt if self.guest_prompt_str is not None: - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.system_message = {"role":"system","content":self.guest_prompt_str} return prompt return None @@ -241,25 +243,25 @@ class AIAgent(BaseAIAgent): if contact.is_family_member: if self.owner_promp_str is not None: real_str = self.owner_promp_str.format_map(contact.to_dict()) - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.system_message = {"role":"system","content":real_str} return prompt else: if self.contact_prompt_str is not None: real_str = self.contact_prompt_str.format_map(contact.to_dict()) - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.system_message = {"role":"system","content":real_str} return prompt return None - def get_agent_prompt(self) -> AgentPrompt: + def get_agent_prompt(self) -> LLMPrompt: return self.agent_prompt - async def _get_agent_think_prompt(self) -> AgentPrompt: + async def _get_agent_think_prompt(self) -> LLMPrompt: return self.agent_think_prompt - def _format_msg_by_env_value(self,prompt:AgentPrompt): + def _format_msg_by_env_value(self,prompt:LLMPrompt): for msg in prompt.messages: old_content = msg.get("content") msg["content"] = old_content.format_map(self.agent_workspace) @@ -284,7 +286,7 @@ class AIAgent(BaseAIAgent): return image_path async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: - msg_prompt = AgentPrompt() + msg_prompt = LLMPrompt() if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: need_process = False if msg.is_image_msg(): @@ -378,7 +380,7 @@ class AIAgent(BaseAIAgent): workspace = self.get_workspace_by_msg(msg) - prompt = AgentPrompt() + prompt = LLMPrompt() if workspace: prompt.append(workspace.get_prompt()) prompt.append(workspace.get_role_prompt(self.agent_id)) @@ -390,7 +392,7 @@ class AIAgent(BaseAIAgent): if self.need_session_summmary(msg,chatsession): # get relate session(todos) summary summary = self.llm_select_session_summary(msg,chatsession) - prompt.append(AgentPrompt(summary)) + prompt.append(LLMPrompt(summary)) known_info_str = "# Known information\n" have_known_info = False @@ -399,7 +401,7 @@ class AIAgent(BaseAIAgent): have_known_info = True known_info_str += f"## todo\n{todos_str}\n" inner_functions,function_token_len = BaseAIAgent.get_inner_functions(self.agent_workspace) - system_prompt_len = self.token_len(prompt=prompt) + system_prompt_len = ComputeKernel.llm_num_tokens(prompt) input_len = len(msg.body) if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: history_str,history_token_len = await self._get_prompt_from_session_for_groupchat(chatsession,system_prompt_len + function_token_len,input_len) @@ -410,7 +412,7 @@ class AIAgent(BaseAIAgent): known_info_str += history_str if have_known_info: - known_info_prompt = AgentPrompt(known_info_str) + known_info_prompt = LLMPrompt(known_info_str) prompt.append(known_info_prompt) # chat context prompt.append(msg_prompt) @@ -436,7 +438,7 @@ class AIAgent(BaseAIAgent): final_result = llm_result.resp - await workspace.exec_op_list(llm_result.op_list,self.agent_id) + await workspace.exec_op_list(llm_result.action_list,self.agent_id) is_ignore = False result_prompt_str = "" @@ -471,12 +473,12 @@ class AIAgent(BaseAIAgent): return None - async def _get_history_prompt_for_think(self,chatsession:AIChatSession,summary:str,system_token_len:int,pos:int)->(AgentPrompt,int): + async def _get_history_prompt_for_think(self,chatsession:AIChatSession,summary:str,system_token_len:int,pos:int)->(LLMPrompt,int): history_len = (self.max_token_size * 0.7) - system_token_len messages = chatsession.read_history(self.history_len,pos,"natural") # read result_token_len = 0 - result_prompt = AgentPrompt() + result_prompt = LLMPrompt() have_summary = False if summary is not None: if len(summary) > 1: @@ -511,7 +513,7 @@ class AIAgent(BaseAIAgent): history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len messages = chatsession.read_history(self.history_len) # read result_token_len = 0 - result_prompt = AgentPrompt() + result_prompt = LLMPrompt() read_history_msg = 0 for msg in reversed(messages): read_history_msg += 1 @@ -569,13 +571,13 @@ class AIAgent(BaseAIAgent): async def _llm_read_report(self,report:AgentReport,worksapce:WorkspaceEnvironment): work_summary = worksapce.get_work_summary(self.agent_id) - prompt : AgentPrompt = AgentPrompt() + prompt : LLMPrompt = LLMPrompt() prompt.append(self.agent_prompt) prompt.append(worksapce.get_role_prompt(self.agent_id)) prompt.append(self.read_report_prompt) # report is a message from other agent(human) about work - prompt.append(AgentPrompt(work_summary)) - prompt.append(AgentPrompt(report.content)) + prompt.append(LLMPrompt(work_summary)) + prompt.append(LLMPrompt(report.content)) task_result:ComputeTaskResult = await self.do_llm_complection(prompt) @@ -606,7 +608,7 @@ class AIAgent(BaseAIAgent): do_prompts = self._can_do_todo(todo_list_type, todo) if do_prompts: - prompt : AgentPrompt = AgentPrompt() + prompt : LLMPrompt = LLMPrompt() prompt.append(self.agent_prompt) prompt.append(workspace.get_role_prompt(self.agent_id)) prompt.append(do_prompts) @@ -635,13 +637,13 @@ class AIAgent(BaseAIAgent): check_prompts = self._can_check_todo(todo_list_type, todo) if check_prompts: - prompt : AgentPrompt = AgentPrompt() + prompt : LLMPrompt = LLMPrompt() prompt.append(self.agent_prompt) prompt.append(workspace.get_role_prompt(self.agent_id)) prompt.append(check_prompts) if todo.last_check_result: - prompt.append(AgentPrompt(todo.last_check_result)) + prompt.append(LLMPrompt(todo.last_check_result)) prompt.append(todo.detail) prompt.append(todo.result) @@ -669,7 +671,7 @@ class AIAgent(BaseAIAgent): prompt.append(review_prompts) todo_tree = todo_list.get_todo_tree("/") - prompt.append(AgentPrompt(todo_tree)) + prompt.append(LLMPrompt(todo_tree)) do_result : AgentTodoResult = await self._llm_review_todo(todo, prompt, workspace) todo.last_review_time = datetime.datetime.now().timestamp() @@ -690,7 +692,7 @@ class AIAgent(BaseAIAgent): logger.info(f"agent {self.agent_id} ,check:{check_count} todo,do:{do_count} todo.") - def _can_review_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> AgentPrompt: + def _can_review_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> LLMPrompt: do_prompts = self.todo_prompts[todo_list_type].get("review") if not do_prompts: return None @@ -701,7 +703,7 @@ class AIAgent(BaseAIAgent): return do_prompts - def _can_check_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> AgentPrompt: + def _can_check_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> LLMPrompt: do_prompts = self.todo_prompts[todo_list_type].get("check") if not do_prompts: return None @@ -720,7 +722,7 @@ class AIAgent(BaseAIAgent): return do_prompts - def _can_do_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> AgentPrompt: + def _can_do_todo(self, todo_list_type: TodoListType, todo:AgentTodo) -> LLMPrompt: do_prompts = self.todo_prompts[todo_list_type].get("do") if not do_prompts: return None @@ -739,7 +741,7 @@ class AIAgent(BaseAIAgent): return do_prompts - async def _llm_do_todo(self, todo: AgentTodo, prompt: AgentPrompt, workspace: WorkspaceEnvironment) -> AgentTodoResult: + async def _llm_do_todo(self, todo: AgentTodo, prompt: LLMPrompt, workspace: WorkspaceEnvironment) -> AgentTodoResult: result = AgentTodoResult() task_result:ComputeTaskResult = await self.do_llm_complection(prompt, is_json_resp=True) @@ -763,7 +765,7 @@ class AIAgent(BaseAIAgent): resp = await AIBus.get_default_bus().post_message(msg) logging.info(f"agent {self.agent_id} send msg to {msg.target} result:{resp}") - result_str, have_error = await workspace.exec_op_list(llm_result.op_list, self.agent_id) + result_str, have_error = await workspace.exec_op_list(llm_result.action_list, self.agent_id) if have_error: result.result_code = AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR #result.error_str = error_str @@ -771,7 +773,7 @@ class AIAgent(BaseAIAgent): result.result_str = result_str return result - async def _llm_check_todo(self, todo: AgentTodo, prompt: AgentPrompt, workspace: WorkspaceEnvironment) -> AgentTodoResult: + async def _llm_check_todo(self, todo: AgentTodo, prompt: LLMPrompt, workspace: WorkspaceEnvironment) -> AgentTodoResult: result = AgentTodoResult() inner_functions,_ = BaseAIAgent.get_inner_functions(workspace) @@ -786,7 +788,7 @@ class AIAgent(BaseAIAgent): todo.last_check_result = task_result.result_str return result - async def _llm_review_todo(self, todo:AgentTodo, prompt: AgentPrompt, workspace: WorkspaceEnvironment): + async def _llm_review_todo(self, todo:AgentTodo, prompt: LLMPrompt, workspace: WorkspaceEnvironment): inner_functions,_ = BaseAIAgent.get_inner_functions(workspace) task_result:ComputeTaskResult = await self.do_llm_complection(prompt,inner_functions=inner_functions) @@ -842,10 +844,10 @@ class AIAgent(BaseAIAgent): while True: cur_pos = chatsession.summarize_pos summary = chatsession.summary - prompt:AgentPrompt = AgentPrompt() + prompt:LLMPrompt = LLMPrompt() #prompt.append(self._get_agent_prompt()) prompt.append(await self._get_agent_think_prompt()) - system_prompt_len = self.token_len(prompt=prompt) + system_prompt_len = ComputeKernel.llm_num_tokens(prompt) #think env? history_prompt,next_pos = await self._get_history_prompt_for_think(chatsession,summary,system_prompt_len,cur_pos) prompt.append(history_prompt) @@ -864,7 +866,7 @@ class AIAgent(BaseAIAgent): chatsession.update_think_progress(next_pos,new_summary) return - async def get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len) -> AgentPrompt: + async def get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len) -> LLMPrompt: # TODO: get prompt from group chat is different from single chat if self.enable_thread: return None diff --git a/src/aios/agent/agent_base.py b/src/aios/agent/agent_base.py index 562a069..f9e0b5b 100644 --- a/src/aios/agent/agent_base.py +++ b/src/aios/agent/agent_base.py @@ -11,400 +11,15 @@ import shlex import json from typing import List, Tuple -from .ai_function import FunctionItem, AIFunction -from ..proto.agent_msg import AgentMsg, AgentMsgType -from ..proto.compute_task import ComputeTaskResult,ComputeTaskResultCode -from ..environment.environment import BaseEnvironment +from ..proto.ai_function import * +from ..proto.agent_msg import * +from ..proto.compute_task import * +from ..environment.environment import * logger = logging.getLogger(__name__) -class AgentPrompt: - def __init__(self,prompt_str = None) -> None: - self.messages = [] - if prompt_str: - self.messages.append({"role":"user","content":prompt_str}) - self.system_message = None - - def as_str(self)->str: - result_str = "" - if self.system_message: - result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" - if self.messages: - for msg in self.messages: - 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 = copy.deepcopy(prompt.system_message) - else: - self.system_message["content"] += prompt.system_message.get("content") - - 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 = [] - for msg in config: - if msg.get("content"): - if msg.get("role") == "system": - self.system_message = msg - else: - self.messages.append(msg) - else: - logger.error("prompt message has no content!") - return True - -class LLMResult: - def __init__(self) -> None: - self.state : str = "ignore" - self.resp : str = "" - self.raw_resp = None - self.paragraphs : dict[str,FunctionItem] = [] - - - self.post_msgs : List[AgentMsg] = [] - self.send_msgs : List[AgentMsg] = [] - self.calls : List[FunctionItem] = [] - self.post_calls : List[FunctionItem] = [] - self.op_list : List[FunctionItem] = [] # op_list is a optimize design for saving token - @classmethod - def from_json_str(self,llm_json_str:str) -> 'LLMResult': - r = LLMResult() - if llm_json_str is None: - r.state = "ignore" - return r - if llm_json_str == "ignore": - r.state = "ignore" - return r - - llm_json = json.loads(llm_json_str) - r.state = llm_json.get("state") - r.resp = llm_json.get("resp") - r.raw_resp = llm_json - - post_msgs = llm_json.get("post_msg") - r.post_msgs = [] - if post_msgs: - for msg in post_msgs: - new_msg = AgentMsg() - target_id = msg.get("target") - msg_content = msg.get("content") - new_msg.set("",target_id,msg_content) - r.post_msgs.append(new_msg) - #new_msg.msg_type = AgentMsgType.TYPE_MSG - - r.calls = llm_json.get("calls") - r.post_calls = llm_json.get("post_calls") - r.op_list = llm_json.get("op_list") - - return r - - @classmethod - def from_str(self,llm_result_str:str,valid_func:List[str]=None) -> 'LLMResult': - r = LLMResult() - - if llm_result_str is None: - r.state = "ignore" - return r - if llm_result_str == "ignore": - r.state = "ignore" - return r - - if llm_result_str[0] == "{": - return LLMResult.from_json_str(llm_result_str) - # if llm_result_str.startswith("json"): - # return LLMResult.from_json_str(llm_result_str[4:]) - - lines = llm_result_str.splitlines() - is_need_wait = False - - def check_args(func_item:FunctionItem): - match func_name: - case "send_msg":# /send_msg $target_id - if len(func_args) != 1: - return False - - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set("",target_id,msg_content) - - r.send_msgs.append(new_msg) - is_need_wait = True - return True - - case "post_msg":# /post_msg $target_id - if len(func_args) != 1: - return False - - new_msg = AgentMsg() - target_id = func_item.args[0] - msg_content = func_item.body - new_msg.set("",target_id,msg_content) - r.post_msgs.append(new_msg) - return True - - 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 - case _: - if valid_func is not None: - if func_name in valid_func: - r.paragraphs[func_name] = func_item - return True - - return False - - - 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: - if current_func: - 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: - r.state = "reponsed" - - return r - -class AgentReport: - def __init__(self): - pass - -class AgentTodoResult: - TODO_RESULT_CODE_OK = 0, - TODO_RESULT_CODE_LLM_ERROR = 1, - TODO_RESULT_CODE_EXEC_OP_ERROR = 2 - - - def __init__(self) -> None: - self.result_code = AgentTodoResult.TODO_RESULT_CODE_OK - self.result_str = None - self.error_str = None - self.op_list = None - - def to_dict(self) -> dict: - result = {} - result["result_code"] = self.result_code - result["result_str"] = self.result_str - result["error_str"] = self.error_str - result["op_list"] = self.op_list - return result - - -class AgentTodo: - TODO_STATE_WAIT_ASSIGN = "wait_assign" - TODO_STATE_INIT = "init" - - TODO_STATE_PENDING = "pending" - TODO_STATE_WAITING_CHECK = "wait_check" - TODO_STATE_EXEC_FAILED = "exec_failed" - TDDO_STATE_CHECKFAILED = "check_failed" - - TODO_STATE_CANCEL = "cancel" - TODO_STATE_DONE = "done" - TODO_STATE_REVIEWED = "reviewed" - TODO_STATE_EXPIRED = "expired" - - def __init__(self): - self.todo_id = "todo#" + uuid.uuid4().hex - self.title = None - self.detail = None - self.todo_path = None # get parent todo,sub todo by path - #self.parent = None - self.create_time = time.time() - - self.state = "wait_assign" - self.worker = None - self.checker = None - self.createor = None - - self.need_check = True - self.due_date = time.time() + 3600 * 24 * 2 - self.last_do_time = None - self.last_check_time = None - self.last_review_time = None - - self.depend_todo_ids = [] - self.sub_todos = {} - - self.result : AgentTodoResult = None - self.last_check_result = None - self.retry_count = 0 - self.raw_obj = None - - @classmethod - def from_dict(cls,json_obj:dict) -> 'AgentTodo': - todo = AgentTodo() - if json_obj.get("id") is not None: - todo.todo_id = json_obj.get("id") - - todo.title = json_obj.get("title") - todo.state = json_obj.get("state") - create_time = json_obj.get("create_time") - if create_time: - todo.create_time = datetime.fromisoformat(create_time).timestamp() - - todo.detail = json_obj.get("detail") - due_date = json_obj.get("due_date") - if due_date: - todo.due_date = datetime.fromisoformat(due_date).timestamp() - - last_do_time = json_obj.get("last_do_time") - if last_do_time: - todo.last_do_time = datetime.fromisoformat(last_do_time).timestamp() - last_check_time = json_obj.get("last_check_time") - if last_check_time: - todo.last_check_time = datetime.fromisoformat(last_check_time).timestamp() - last_review_time = json_obj.get("last_review_time") - if last_review_time: - todo.last_review_time = datetime.fromisoformat(last_review_time).timestamp() - - todo.depend_todo_ids = json_obj.get("depend_todo_ids") - todo.need_check = json_obj.get("need_check") - #todo.result = json_obj.get("result") - #todo.last_check_result = json_obj.get("last_check_result") - todo.worker = json_obj.get("worker") - todo.checker = json_obj.get("checker") - todo.createor = json_obj.get("createor") - if json_obj.get("retry_count"): - todo.retry_count = json_obj.get("retry_count") - - todo.raw_obj = json_obj - - return todo - - def to_dict(self) -> dict: - if self.raw_obj: - result = self.raw_obj - else: - result = {} - - result["id"] = self.todo_id - #result["parent_id"] = self.parent_id - result["title"] = self.title - result["state"] = self.state - result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() - result["detail"] = self.detail - result["due_date"] = datetime.fromtimestamp(self.due_date).isoformat() - result["last_do_time"] = datetime.fromtimestamp(self.last_do_time).isoformat() if self.last_do_time else None - result["last_check_time"] = datetime.fromtimestamp(self.last_check_time).isoformat() if self.last_check_time else None - result["last_review_time"] = datetime.fromtimestamp(self.last_review_time).isoformat() if self.last_review_time else None - result["depend_todo_ids"] = self.depend_todo_ids - result["need_check"] = self.need_check - result["worker"] = self.worker - result["checker"] = self.checker - result["createor"] = self.createor - result["retry_count"] = self.retry_count - - return result - - def to_prompt(self) -> AgentPrompt: - json_str = json.dumps(self.raw_obj) - return AgentPrompt(json_str) - - def can_review(self) -> bool: - if self.state != AgentTodo.TODO_STATE_DONE: - return False - - now = datetime.now().timestamp() - if self.last_review_time: - time_diff = now - self.last_review_time - if time_diff < 60*15: - logger.info(f"todo {self.title} is already reviewed, ignore") - return False - - return True - - def can_check(self)->bool: - if self.state != AgentTodo.TODO_STATE_WAITING_CHECK: - return False - - now = datetime.now().timestamp() - if self.last_check_time: - time_diff = now - self.last_check_time - if time_diff < 60*15: - logger.info(f"todo {self.title} is already checked, ignore") - return False - - return True - - def can_do(self) -> bool: - match self.state: - case AgentTodo.TODO_STATE_DONE: - logger.info(f"todo {self.title} is done, ignore") - return False - case AgentTodo.TODO_STATE_CANCEL: - logger.info(f"todo {self.title} is cancel, ignore") - return False - case AgentTodo.TODO_STATE_EXPIRED: - logger.info(f"todo {self.title} is expired, ignore") - return False - case AgentTodo.TODO_STATE_EXEC_FAILED: - if self.retry_count > 3: - logger.info(f"todo {self.title} retry count ({self.retry_count}) is too many, ignore") - return False - - now = datetime.now().timestamp() - time_diff = self.due_date - now - if time_diff < 0: - logger.info(f"todo {self.title} is expired, ignore") - self.state = AgentTodo.TODO_STATE_EXPIRED - return False - - if time_diff > 7*24*3600: - logger.info(f"todo {self.title} is far before due date, ignore") - return False - - if self.last_do_time: - time_diff = now - self.last_do_time - if time_diff < 60*15: - logger.info(f"todo {self.title} is already do ignore") - return False - - logger.info(f"todo {self.title} can do.") - return True - - -class AgentWorkLog: - def __init__(self) -> None: - pass class BaseAIAgent(abc.ABC): @@ -420,19 +35,9 @@ class BaseAIAgent(abc.ABC): def get_max_token_size(self) -> int: pass - def token_len(self, text:str=None, prompt:AgentPrompt=None) -> int: - from ..frame.compute_kernel import ComputeKernel - if text: - return ComputeKernel.llm_num_tokens_from_text(text,self.get_llm_model_name()) - elif prompt: - result = 0 - if prompt.system_message: - result += ComputeKernel.llm_num_tokens_from_text(prompt.system_message.get("content"),self.get_llm_model_name()) - for msg in prompt.messages: - result += ComputeKernel.llm_num_tokens_from_text(msg.get("content"),self.get_llm_model_name()) - return result - else: - return 0 + @abstractmethod + async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: + pass @classmethod def get_inner_functions(cls, env:BaseEnvironment) -> (dict,int): @@ -458,7 +63,7 @@ class BaseAIAgent(abc.ABC): async def do_llm_complection( self, - prompt:AgentPrompt, + prompt:LLMPrompt, org_msg:AgentMsg=None, env:BaseEnvironment=None, inner_functions=None, @@ -503,7 +108,7 @@ class BaseAIAgent(abc.ABC): inner_func_call_node = result_message.get("function_call") if inner_func_call_node: - call_prompt : AgentPrompt = copy.deepcopy(prompt) + call_prompt : LLMPrompt = copy.deepcopy(prompt) func_msg = copy.deepcopy(result_message) del func_msg["tool_calls"] call_prompt.messages.append(func_msg) @@ -515,7 +120,7 @@ class BaseAIAgent(abc.ABC): self, env: BaseEnvironment, inner_func_call_node: dict, - prompt: AgentPrompt, + prompt: LLMPrompt, inner_functions: dict, org_msg:AgentMsg, stack_limit = 5 diff --git a/src/aios/agent/behavior.py b/src/aios/agent/behavior.py deleted file mode 100644 index 64b8834..0000000 --- a/src/aios/agent/behavior.py +++ /dev/null @@ -1,4 +0,0 @@ -# TODO: let agent develolp custmized behavior easily -class AgentBehavior: - def __init__(self) -> None: - pass \ No newline at end of file diff --git a/src/aios/agent/llm_process.py b/src/aios/agent/llm_process.py new file mode 100644 index 0000000..9c1245a --- /dev/null +++ b/src/aios/agent/llm_process.py @@ -0,0 +1,149 @@ +# Old name is behavior, I belive new name "llm_process" is better +from abc import ABC,abstractmethod +import copy +import json +import shlex +from typing import Any, Callable, Optional,Dict,Awaitable,List +from enum import Enum + +from ..proto.compute_task import * +from ..proto.ai_function import * +from ..frame.compute_kernel import * + +import logging +logger = logging.getLogger(__name__) + +MIN_PREDICT_TOKEN_LEN = 32 + + +class BaseLLMProcess: + def __init__(self) -> None: + self.enable_json_resp = False + self.model_name = "gpt-4" + self.max_token = 2000 # include input prompt + self.timeout = 1800 # 30 min + + @abstractmethod + async def prepare_prompt(self) -> LLMPrompt: + pass + + @abstractmethod + async def get_inner_function(self,func_name:str) -> AIFunction: + pass + + async def _execute_inner_func(self,inner_func_call_node,prompt: LLMPrompt,stack_limit = 5) -> ComputeTaskResult: + arguments = None + try: + func_name = inner_func_call_node.get("name") + arguments = json.loads(inner_func_call_node.get("arguments")) + logger.info(f"LLMProcess execute inner func:{func_name} :\n\t {json.dumps(arguments)}") + + func_node : AIFunction = await self.get_inner_function(func_name) + if func_node is None: + result_str:str = f"execute {func_name} error,function not found" + else: + result_str:str = await func_node.execute(**arguments) + except Exception as e: + result_str = f"execute {func_name} error:{str(e)}" + logger.error(f"LLMProcess execute inner func:{func_name} error:\n\t{e}") + + logger.info("LLMProcess execute inner func result:" + result_str) + + prompt.messages.append({"role":"function","content":result_str,"name":func_name}) + if self.enable_json_resp: + resp_mode = "json" + else: + resp_mode = "text" + + max_result_token = self.max_token - ComputeKernel.llm_num_tokens(prompt) + if max_result_token < MIN_PREDICT_TOKEN_LEN: + task_result = ComputeTaskResult() + task_result.result_code = ComputeTaskResultCode.ERROR + task_result.error_str = f"prompt too long,can not predict" + return task_result + + task_result: ComputeTaskResult = await (ComputeKernel.get_instance().do_llm_completion( + prompt, + resp_mode=resp_mode, + mode_name=self.model_name, + max_token=max_result_token, + inner_functions=prompt.inner_functions, + timeout=self.timeout)) + + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + return task_result + + inner_func_call_node = None + if stack_limit > 0: + result_message : dict = task_result.result.get("message") + if result_message: + inner_func_call_node = result_message.get("function_call") + if inner_func_call_node: + func_msg = copy.deepcopy(result_message) + del func_msg["tool_calls"]#TODO: support tool_calls? + prompt.messages.append(func_msg) + else: + logger.error(f"inner function call stack limit reached") + task_result.result_code = ComputeTaskResultCode.ERROR + task_result.error_str = "inner function call stack limit reached" + return task_result + + if inner_func_call_node: + return await self._execute_inner_func(inner_func_call_node,prompt,stack_limit-1) + else: + return task_result + + async def process(self) -> LLMResult: + if self.enable_json_resp: + resp_mode = "json" + else: + resp_mode = "text" + + prompt = await self.prepare_prompt() + max_result_token = self.max_token - ComputeKernel.llm_num_tokens(prompt) + if max_result_token < MIN_PREDICT_TOKEN_LEN: + return LLMResult.from_error_str(f"prompt too long,can not predict") + + task_result: ComputeTaskResult = await (ComputeKernel.get_instance().do_llm_completion( + prompt, + resp_mode=resp_mode, + mode_name=self.model_name, + max_token=max_result_token, + inner_functions=prompt.inner_functions, + timeout=self.timeout)) + + if task_result.result_code != ComputeTaskResultCode.OK: + err_str = f"do_llm_completion error:{task_result.error_str}" + logger.error(err_str) + return LLMResult.from_error_str(err_str) + + result_message = task_result.result.get("message") + inner_func_call_node = None + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + call_prompt : LLMPrompt = copy.deepcopy(prompt) + func_msg = copy.deepcopy(result_message) + del func_msg["tool_calls"] + call_prompt.messages.append(func_msg) + task_result = await self._execute_inner_func(inner_func_call_node,call_prompt) + + # parse task_result to LLM Result + if self.enable_json_resp: + llm_result = LLMResult.from_json_str(task_result.result_str) + else: + llm_result = LLMResult.from_str(task_result.result_str) + + # execute op_list in LLM Result? + + return llm_result + +#class LLMProcess + + + + + + diff --git a/src/aios/agent/role.py b/src/aios/agent/role.py index 272815f..d08049d 100644 --- a/src/aios/agent/role.py +++ b/src/aios/agent/role.py @@ -1,6 +1,6 @@ import logging -from .agent_base import AgentPrompt +from .agent_base import LLMPrompt class AIRole: def __init__(self) -> None: @@ -9,7 +9,7 @@ class AIRole: 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 + self.prompt : LLMPrompt = None self.introduce : str = None self.agent = None self.enable_function_list : list[str] = None @@ -31,7 +31,7 @@ class AIRole: prompt_node = config.get("prompt") if prompt_node: - self.prompt = AgentPrompt() + self.prompt = LLMPrompt() if self.prompt.load_from_config(prompt_node) is False: logging.error("load prompt failed!") return False @@ -56,7 +56,7 @@ class AIRole: def get_name(self) -> str: return self.role_name - def get_prompt(self) -> AgentPrompt: + def get_prompt(self) -> LLMPrompt: return self.prompt class AIRoleGroup: diff --git a/src/aios/agent/workflow.py b/src/aios/agent/workflow.py index d0bd585..ce245c7 100644 --- a/src/aios/agent/workflow.py +++ b/src/aios/agent/workflow.py @@ -9,11 +9,11 @@ from abc import ABC, abstractmethod from ..proto.compute_task import * from ..proto.agent_msg import * +from ..proto.ai_function import * from .agent_base import * from .chatsession import AIChatSession from .role import AIRole,AIRoleGroup -from .ai_function import AIFunction,FunctionItem from ..frame.compute_kernel import ComputeKernel from ..frame.bus import AIBus @@ -48,7 +48,7 @@ class Workflow: def __init__(self) -> None: self.workflow_name : str = None self.workflow_id : str = None - self.rule_prompt : AgentPrompt = None + self.rule_prompt : LLMPrompt = None self.workflow_config = None self.role_group : dict = None self.input_filter : MessageFilter= None @@ -83,7 +83,7 @@ class Workflow: self.db_file = self.owner_workflow.db_file if config.get("prompt") is not None: - self.rule_prompt = AgentPrompt() + self.rule_prompt = LLMPrompt() if self.rule_prompt.load_from_config(config.get("prompt")) is False: logger.error("Workflow load prompt failed") return False @@ -279,7 +279,7 @@ class Workflow: logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}") return await self.get_bus().send_message(msg) - async def role_call(self,func_item:FunctionItem,the_role:AIRole): + async def role_call(self,func_item:ActionItem,the_role:AIRole): logger.info(f"{the_role.role_id} call {func_item.name} ") arguments = func_item.args @@ -290,11 +290,11 @@ class Workflow: result_str:str = await func_node.execute(**arguments) return result_str - async def role_post_call(self,func_item:FunctionItem,the_role:AIRole): + async def role_post_call(self,func_item:ActionItem,the_role:AIRole): logger.info(f"{the_role.role_id} post call {func_item.name} ") return await self.role_call(func_item,the_role) - def _format_msg_by_env_value(self,prompt:AgentPrompt): + def _format_msg_by_env_value(self,prompt:LLMPrompt): if self.workflow_env is None: return @@ -326,7 +326,7 @@ 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,int]: + async def _role_execute_func(self,the_role:AIRole,inenr_func_call_node:dict,prompt:LLMPrompt,org_msg:AgentMsg,stack_limit = 5) -> [str,int]: func_name = inenr_func_call_node.get("name") arguments = json.loads(inenr_func_call_node.get("arguments")) @@ -372,7 +372,7 @@ class Workflow: async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession) -> AgentMsg: msg.target = the_role.get_role_id() - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.append(the_role.agent.agent_prompt) prompt.append(self.get_workflow_rule_prompt()) prompt.append(the_role.get_prompt()) @@ -382,7 +382,7 @@ class Workflow: #support group chat, user content include sender name! prompt.append(await self._get_prompt_from_session(the_role,workflow_chat_session)) - msg_prompt = AgentPrompt() + msg_prompt = LLMPrompt() msg_prompt.messages = [{"role":"user","content":f"user name is {msg.sender}, his question is :{msg.body}"}] prompt.append(msg_prompt) @@ -461,20 +461,20 @@ class Workflow: # message will be saved in role.process_message pass - this_llm_resp_prompt = AgentPrompt() + this_llm_resp_prompt = LLMPrompt() this_llm_resp_prompt.messages = [{"role":"assistant","content":result_str}] prompt.append(this_llm_resp_prompt) - result_prompt = AgentPrompt() + result_prompt = LLMPrompt() result_prompt.messages = [{"role":"user","content":result_prompt_str}] prompt.append(result_prompt) return await _do_process_msg() return await _do_process_msg() - async def _get_prompt_from_session(self,the_role:AIRole,chatsession:AIChatSession) -> AgentPrompt: + async def _get_prompt_from_session(self,the_role:AIRole,chatsession:AIChatSession) -> LLMPrompt: messages = chatsession.read_history(the_role.history_len) # read last 10 message - result_prompt = AgentPrompt() + result_prompt = LLMPrompt() for msg in reversed(messages): if msg.sender == the_role.role_id: @@ -484,10 +484,10 @@ class Workflow: return result_prompt - def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt: + def _get_knowlege_prompt(self,role_name:str) -> LLMPrompt: pass - def get_workflow_rule_prompt(self) -> AgentPrompt: + def get_workflow_rule_prompt(self) -> LLMPrompt: return self.rule_prompt # def _env_event_to_msg(self,env_event:EnvironmentEvent) -> AgentMsg: diff --git a/src/aios/environment/asr_function.py b/src/aios/environment/asr_function.py index 9b97006..81f8dc6 100644 --- a/src/aios/environment/asr_function.py +++ b/src/aios/environment/asr_function.py @@ -2,7 +2,7 @@ import logging from typing import Dict from ..frame.compute_kernel import ComputeKernel -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * logger = logging.getLogger(__name__) diff --git a/src/aios/environment/code_interpreter_function.py b/src/aios/environment/code_interpreter_function.py index 734af71..5913ba4 100644 --- a/src/aios/environment/code_interpreter_function.py +++ b/src/aios/environment/code_interpreter_function.py @@ -1,6 +1,6 @@ from typing import Dict -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * from .code_interpreter import execute_code diff --git a/src/aios/environment/duckduckgo_text_search_function.py b/src/aios/environment/duckduckgo_text_search_function.py index 5418095..22632eb 100644 --- a/src/aios/environment/duckduckgo_text_search_function.py +++ b/src/aios/environment/duckduckgo_text_search_function.py @@ -1,7 +1,7 @@ import json from typing import Dict -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * from duckduckgo_search import AsyncDDGS diff --git a/src/aios/environment/environment.py b/src/aios/environment/environment.py index 1af1813..72b5ee2 100644 --- a/src/aios/environment/environment.py +++ b/src/aios/environment/environment.py @@ -4,7 +4,7 @@ from abc import ABC, abstractmethod from typing import Any, Callable, Optional,Dict,Awaitable,List import logging -from ..agent.ai_function import AIFunction, AIOperation +from ..proto.ai_function import * logger = logging.getLogger(__name__) diff --git a/src/aios/environment/image_2_text_function.py b/src/aios/environment/image_2_text_function.py index 8760f56..422917f 100644 --- a/src/aios/environment/image_2_text_function.py +++ b/src/aios/environment/image_2_text_function.py @@ -2,7 +2,7 @@ import logging from typing import Dict from ..frame.compute_kernel import ComputeKernel -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * logger = logging.getLogger(__name__) diff --git a/src/aios/environment/script_to_speech_function.py b/src/aios/environment/script_to_speech_function.py index 6c71229..043bc0d 100644 --- a/src/aios/environment/script_to_speech_function.py +++ b/src/aios/environment/script_to_speech_function.py @@ -5,7 +5,7 @@ import random from pathlib import Path from typing import Dict -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * from ..frame.compute_kernel import ComputeKernel from ..storage.storage import AIStorage diff --git a/src/aios/environment/sql_database_function.py b/src/aios/environment/sql_database_function.py index 989b82b..c28b0e2 100644 --- a/src/aios/environment/sql_database_function.py +++ b/src/aios/environment/sql_database_function.py @@ -3,7 +3,7 @@ from typing import Dict from cachetools import TLRUCache, cached -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * from .sql_database import SQLDatabase, get_from_env diff --git a/src/aios/environment/text_to_speech_function.py b/src/aios/environment/text_to_speech_function.py index e93a568..878bed5 100644 --- a/src/aios/environment/text_to_speech_function.py +++ b/src/aios/environment/text_to_speech_function.py @@ -5,7 +5,7 @@ import random from pathlib import Path from typing import Dict -from ..agent.ai_function import AIFunction +from ..proto.ai_function import * from ..frame.compute_kernel import ComputeKernel from ..storage.storage import AIStorage diff --git a/src/aios/environment/workflow_env.py b/src/aios/environment/workflow_env.py index df1003f..c48bbfe 100644 --- a/src/aios/environment/workflow_env.py +++ b/src/aios/environment/workflow_env.py @@ -10,7 +10,7 @@ from typing import Optional import aiosqlite from ..proto.compute_task import * -from ..agent.ai_function import SimpleAIFunction +from ..proto.ai_function import * from ..frame.compute_kernel import ComputeKernel from ..frame.contact_manager import ContactManager,Contact,FamilyMember from ..storage.storage import AIStorage @@ -302,7 +302,7 @@ class CalenderEnvironment(SimpleEnvironment): return f'exec paint OK, saved as a local file, path is: {result.result["file"]}' -class PaintEnvironment(BaseEnvironment): +class PaintEnvironment(SimpleEnvironment): def __init__(self, env_id: str) -> None: super().__init__(env_id) self.is_run = False diff --git a/src/aios/environment/workspace_env.py b/src/aios/environment/workspace_env.py index 93b1005..5d52c29 100644 --- a/src/aios/environment/workspace_env.py +++ b/src/aios/environment/workspace_env.py @@ -6,8 +6,12 @@ import sqlite3 import asyncio from typing import Any,List,Dict import chardet -from ..agent.agent_base import AgentMsg,AgentTodo,AgentPrompt,AgentTodoResult -from ..agent.ai_function import AIFunction,SimpleAIFunction, SimpleAIOperation + +from ..proto.agent_task import * +from ..proto.ai_function import * +from ..proto.compute_task import * +from ..agent.agent_base import * + from ..storage.storage import AIStorage,ResourceLocation from .environment import SimpleEnvironment, CompositeEnvironment @@ -289,16 +293,16 @@ class WorkspaceEnvironment(CompositeEnvironment): def get_prompt(self) -> AgentMsg: return None - def get_role_prompt(self,role_id:str) -> AgentPrompt: + def get_role_prompt(self,role_id:str) -> LLMPrompt: return None - def get_do_prompt(self,todo:AgentTodo=None)->AgentPrompt: + def get_do_prompt(self,todo:AgentTodo=None)->LLMPrompt: return None # result mean: list[op_error_str],have_error async def exec_op_list(self,oplist:List,agent_id:str)->tuple[List[str],bool]: result_str = "op list is none" - if oplist is None: + if oplist is None or len(oplist) == 0: return None,False result_str = [] diff --git a/src/aios/frame/compute_kernel.py b/src/aios/frame/compute_kernel.py index 45670cc..ccdb930 100644 --- a/src/aios/frame/compute_kernel.py +++ b/src/aios/frame/compute_kernel.py @@ -8,7 +8,7 @@ from asyncio import Queue from ..proto.compute_task import * from ..knowledge import ObjectID -from ..agent.agent_base import AgentPrompt + from .compute_node import ComputeNode logger = logging.getLogger(__name__) @@ -106,6 +106,9 @@ class ComputeKernel: @staticmethod def llm_num_tokens_from_text(text:str,model:str) -> int: + if model is None: + model = "gpt4" + try: encoding = tiktoken.encoding_for_model(model) except KeyError: @@ -115,9 +118,12 @@ class ComputeKernel: token_count = len(encoding.encode(text)) return token_count + @staticmethod + def llm_num_tokens(prompt: LLMPrompt, model_name: str = None) -> int: + return ComputeKernel.llm_num_tokens_from_text(prompt.as_str(), model_name) # friendly interface for use: - def llm_completion(self, prompt: AgentPrompt, resp_mode:str="text",mode_name: Optional[str] = None, max_token: int = 0,inner_functions = None): + def llm_completion(self, prompt: LLMPrompt, resp_mode:str="text",mode_name: Optional[str] = None, max_token: int = 0,inner_functions = None): # 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) task_req = ComputeTask() @@ -153,7 +159,7 @@ class ComputeKernel: return time_out_result - async def do_llm_completion(self, prompt: AgentPrompt,resp_mode:str="text", mode_name: Optional[str]=None, max_token:int=0, inner_functions=None, timeout=60) -> str: + async def do_llm_completion(self, prompt: LLMPrompt,resp_mode:str="text", mode_name: Optional[str]=None, max_token:int=0, inner_functions=None, timeout=60) -> str: task_req = self.llm_completion(prompt, resp_mode,mode_name, max_token,inner_functions) return await self._wait_task(task_req, timeout) diff --git a/src/aios/proto/agent_msg.py b/src/aios/proto/agent_msg.py index 21ec3b6..93523d7 100644 --- a/src/aios/proto/agent_msg.py +++ b/src/aios/proto/agent_msg.py @@ -232,9 +232,4 @@ class AgentMsg: def get_quote_msg_id(self) -> str: return self.quote_msg_id - @classmethod - def parse_function_call(cls,func_string:str): - str_list = shlex.split(func_string) - func_name = str_list[0] - params = str_list[1:] - return func_name, params + diff --git a/src/aios/proto/agent_task.py b/src/aios/proto/agent_task.py new file mode 100644 index 0000000..791e0d9 --- /dev/null +++ b/src/aios/proto/agent_task.py @@ -0,0 +1,221 @@ + +import datetime +import time + +from anyio import Path + + +class AgentTodoResult: + TODO_RESULT_CODE_OK = 0, + TODO_RESULT_CODE_LLM_ERROR = 1, + TODO_RESULT_CODE_EXEC_OP_ERROR = 2 + + + def __init__(self) -> None: + self.result_code = AgentTodoResult.TODO_RESULT_CODE_OK + self.result_str = None + self.error_str = None + self.op_list = None + + def to_dict(self) -> dict: + result = {} + result["result_code"] = self.result_code + result["result_str"] = self.result_str + result["error_str"] = self.error_str + result["op_list"] = self.op_list + return result + + + + +class AgentTodo: + TODO_STATE_WAIT_ASSIGN = "wait_assign" + TODO_STATE_INIT = "init" + + TODO_STATE_PENDING = "pending" + TODO_STATE_WAITING_CHECK = "wait_check" + TODO_STATE_EXEC_FAILED = "exec_failed" + TDDO_STATE_CHECKFAILED = "check_failed" + + TODO_STATE_CASNCEL = "cancel" + TODO_STATE_DONE = "done" + TODO_STATE_EXPIRED = "expired" + + def __init__(self): + self.todo_id = "todo#" + uuid.uuid4().hex + self.title = None + self.detail = None + self.todo_path = None # get parent todo,sub todo by path + #self.parent = None + self.create_time = time.time() + + self.state = "wait_assign" + self.worker = None + self.checker = None + self.createor = None + + self.need_check = True + self.due_date = time.time() + 3600 * 24 * 2 + self.last_do_time = None + self.last_check_time = None + self.last_review_time = None + + self.depend_todo_ids = [] + self.sub_todos = {} + + self.result : AgentTodoResult = None + self.last_check_result = None + self.retry_count = 0 + self.raw_obj = None + + @classmethod + def from_dict(cls,json_obj:dict) -> 'AgentTodo': + todo = AgentTodo() + if json_obj.get("id") is not None: + todo.todo_id = json_obj.get("id") + + todo.title = json_obj.get("title") + todo.state = json_obj.get("state") + create_time = json_obj.get("create_time") + if create_time: + todo.create_time = datetime.fromisoformat(create_time).timestamp() + + todo.detail = json_obj.get("detail") + due_date = json_obj.get("due_date") + if due_date: + todo.due_date = datetime.fromisoformat(due_date).timestamp() + + last_do_time = json_obj.get("last_do_time") + if last_do_time: + todo.last_do_time = datetime.fromisoformat(last_do_time).timestamp() + last_check_time = json_obj.get("last_check_time") + if last_check_time: + todo.last_check_time = datetime.fromisoformat(last_check_time).timestamp() + last_review_time = json_obj.get("last_review_time") + if last_review_time: + todo.last_review_time = datetime.fromisoformat(last_review_time).timestamp() + + todo.depend_todo_ids = json_obj.get("depend_todo_ids") + todo.need_check = json_obj.get("need_check") + #todo.result = json_obj.get("result") + #todo.last_check_result = json_obj.get("last_check_result") + todo.worker = json_obj.get("worker") + todo.checker = json_obj.get("checker") + todo.createor = json_obj.get("createor") + if json_obj.get("retry_count"): + todo.retry_count = json_obj.get("retry_count") + + todo.raw_obj = json_obj + + return todo + + def to_dict(self) -> dict: + if self.raw_obj: + result = self.raw_obj + else: + result = {} + + result["id"] = self.todo_id + #result["parent_id"] = self.parent_id + result["title"] = self.title + result["state"] = self.state + result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() + result["detail"] = self.detail + result["due_date"] = datetime.fromtimestamp(self.due_date).isoformat() + result["last_do_time"] = datetime.fromtimestamp(self.last_do_time).isoformat() if self.last_do_time else None + result["last_check_time"] = datetime.fromtimestamp(self.last_check_time).isoformat() if self.last_check_time else None + result["last_review_time"] = datetime.fromtimestamp(self.last_review_time).isoformat() if self.last_review_time else None + result["depend_todo_ids"] = self.depend_todo_ids + result["need_check"] = self.need_check + result["worker"] = self.worker + result["checker"] = self.checker + result["createor"] = self.createor + result["retry_count"] = self.retry_count + + return result + + def can_check(self)->bool: + if self.state != AgentTodo.TODO_STATE_WAITING_CHECK: + return False + + now = datetime.now().timestamp() + if self.last_check_time: + time_diff = now - self.last_check_time + if time_diff < 60*15: + logger.info(f"todo {self.title} is already checked, ignore") + return False + + return True + + def can_do(self) -> bool: + match self.state: + case AgentTodo.TODO_STATE_DONE: + logger.info(f"todo {self.title} is done, ignore") + return False + case AgentTodo.TODO_STATE_CASNCEL: + logger.info(f"todo {self.title} is cancel, ignore") + return False + case AgentTodo.TODO_STATE_EXPIRED: + logger.info(f"todo {self.title} is expired, ignore") + return False + case AgentTodo.TODO_STATE_EXEC_FAILED: + if self.retry_count > 3: + logger.info(f"todo {self.title} retry count ({self.retry_count}) is too many, ignore") + return False + + now = datetime.now().timestamp() + time_diff = self.due_date - now + if time_diff < 0: + logger.info(f"todo {self.title} is expired, ignore") + self.state = AgentTodo.TODO_STATE_EXPIRED + return False + + if time_diff > 7*24*3600: + logger.info(f"todo {self.title} is far before due date, ignore") + return False + + if self.last_do_time: + time_diff = now - self.last_do_time + if time_diff < 60*15: + logger.info(f"todo {self.title} is already do ignore") + return False + + logger.info(f"todo {self.title} can do.") + return True + +class AgentTask: + def __init__(self) -> None: + self.task_id : str = "task#" + uuid.uuid4().hex + self.task_path : Path = None # get parent todo,sub todo by path + self.title = None + self.detail = None + + self.create_time = time.time() + + self.state = "wait_assign" + self.worker = None + self.createor = None + + self.due_date = time.time() + 3600 * 24 * 2 + self.depend_task_ids = [] + self.step_todos = {} + + self.last_plan_time = None + self.last_check_time = None + #self.last_review_time = None + + self.result : LLMResult = None + self.last_check_result = None + self.retry_count = 0 + self.raw_obj = None + + + +class AgentWorkLog: + def __init__(self) -> None: + pass + + +class AgentReport: + def __init__(self) -> None: + pass \ No newline at end of file diff --git a/src/aios/agent/ai_function.py b/src/aios/proto/ai_function.py similarity index 98% rename from src/aios/agent/ai_function.py rename to src/aios/proto/ai_function.py index d13ddca..9786cca 100644 --- a/src/aios/agent/ai_function.py +++ b/src/aios/proto/ai_function.py @@ -73,7 +73,7 @@ class AIFunction: #def load_from_config(self,config:dict) -> bool: # pass -class FunctionItem: +class ActionItem: def __init__(self,name,args) -> None: self.name = name self.args = args diff --git a/src/aios/proto/compute_task.py b/src/aios/proto/compute_task.py index 3390465..f761976 100644 --- a/src/aios/proto/compute_task.py +++ b/src/aios/proto/compute_task.py @@ -1,11 +1,20 @@ +import copy from enum import Enum +import json +import shlex import uuid import time -from typing import Union +from typing import List, Union +from ..proto.ai_function import * from ..knowledge import ObjectID from ..storage.storage import AIStorage + +import logging + +logger = logging.getLogger(__name__) + class ComputeTaskResultCode(Enum): OK = 0 TIMEOUT = 1 @@ -31,6 +40,164 @@ class ComputeTaskType(Enum): TEXT_EMBEDDING ="text_embedding" IMAGE_EMBEDDING ="image_embedding" +class LLMPrompt: + def __init__(self,prompt_str = None) -> None: + self.messages = [] + if prompt_str: + self.messages.append({"role":"user","content":prompt_str}) + self.system_message = None + + def as_str(self)->str: + result_str = "" + if self.system_message: + result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" + if self.messages: + for msg in self.messages: + 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:'LLMPrompt'): + if prompt is None: + return + + if prompt.system_message is not None: + if self.system_message is None: + self.system_message = copy.deepcopy(prompt.system_message) + else: + self.system_message["content"] += prompt.system_message.get("content") + + 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 = [] + for msg in config: + if msg.get("content"): + if msg.get("role") == "system": + self.system_message = msg + else: + self.messages.append(msg) + else: + logger.error("prompt message has no content!") + return True + + +class LLMResultStates(Enum): + IGNORE = "ignore" + OK = "ok" # process done + ERROR = "error" + +class LLMResult: + def __init__(self) -> None: + self.state : str = LLMResultStates.IGNORE + self.compute_error_str = None + self.resp : str = "" # llm say: + self.raw_result = None # raw result from compute kernel + self.inner_functions : List[AIFunction] = [] + self.action_list : List[ActionItem] = [] # op_list is a optimize design for saving token + + #self.post_msgs : List[AgentMsg] = [] # move to op_list + # self.send_msgs : List[AgentMsg] = [] # move to op_list + + + @classmethod + def from_error_str(self,error_str:str) -> 'LLMResult': + r = LLMResult() + r.state = "error" + r.compute_error_str = error_str + return r + + @classmethod + def from_json_str(self,llm_json_str:str) -> 'LLMResult': + r = LLMResult() + if llm_json_str is None: + r.state = LLMResultStates.IGNORE + return r + if llm_json_str == "**IGNORE**": + r.state = LLMResultStates.IGNORE + return r + + llm_json = json.loads(llm_json_str) + r.resp = llm_json.get("resp") + r.raw_result = llm_json + r.action_list = llm_json.get("actions") + + return r + + @classmethod + def parse_action(cls,func_string:str): + str_list = shlex.split(func_string) + func_name = str_list[0] + params = str_list[1:] + return func_name, params + + @classmethod + def from_str(self,llm_result_str:str,valid_func:List[str]=None) -> 'LLMResult': + r = LLMResult() + + if llm_result_str is None: + r.state = LLMResultStates.IGNORE + return r + if llm_result_str == "**IGNORE**": + r.state = LLMResultStates.IGNORE + return r + + if llm_result_str[0] == "{": + return LLMResult.from_json_str(llm_result_str) + + lines = llm_result_str.splitlines() + is_need_wait = False + + def check_args(action_item:ActionItem): + match action_item.name: + case "post_msg":# /post_msg $target_id + if len(action_item.args) != 1: + return False + + new_msg = AgentMsg() + target_id = action_item.args[0] + msg_content = action_item.body + new_msg.set("",target_id,msg_content) + + return True + + + return False + + + current_action : ActionItem = None + for line in lines: + if line.startswith("##/"): + if current_action: + if check_args(current_action) is False: + r.resp += current_action.dumps() + else: + r.action_list.append(current_action) + + action_name,action_args = LLMResult.parse_action(line[3:]) + current_action = ActionItem(action_name,action_args) + else: + if current_action: + current_action.append_body(line + "\n") + else: + r.resp += line + "\n" + + if current_action: + if check_args(current_action) is False: + r.resp += current_action.dumps() + else: + r.action_list.append(current_action) + return r class ComputeTask: def __init__(self) -> None: @@ -140,3 +307,5 @@ class ComputeTaskResult: self.task_id = task.task_id self.callchain_id = task.callchain_id task.result = self + + diff --git a/src/aios_kernel/code_interpreter.py b/src/aios_kernel/code_interpreter.py deleted file mode 100644 index 850356a..0000000 --- a/src/aios_kernel/code_interpreter.py +++ /dev/null @@ -1,426 +0,0 @@ -import logging -import os -import pathlib -import shutil -import subprocess -import sys -import re -import time -import ast -from concurrent.futures import ThreadPoolExecutor -from hashlib import md5 -from typing import Optional, Union, List, Tuple -from generic_escape import GenericEscape - -from aios_kernel import AIStorage - -try: - import docker -except ImportError: - docker = None - -CODE_BLOCK_PATTERN = r"```[ \t]*(\w+)?[ \t]*\r?\n(.*?)\r?\n[ \t]*```" -UNKNOWN = "unknown" -TIMEOUT_MSG = "Timeout" -DEFAULT_TIMEOUT = 600 -WIN32 = sys.platform == "win32" -PATH_SEPARATOR = WIN32 and "\\" or "/" - -logger = logging.getLogger(__name__) - - -BUILT_IN_MODULES = set( - [ - "sys", - "os", - "math", - "random", - "datetime", - "json", - "re", - "subprocess", - "time", - "threading", - "logging", - "collections", - "itertools", - "functools", - "operator", - "pathlib", - "shutil", - "tempfile", - "pickle", - "io", - "argparse", - "typing", - "unittest", - "contextlib", - "abc", - "heapq", - "bisect", - "copy", - "decimal", - "fractions", - "hashlib", - "secrets", - "statistics", - "difflib", - "doctest", - "enum", - "inspect", - "traceback", - "weakref", - "gc", - "mmap", - "msvcrt", - "winreg", - "array", - "audioop", - "binascii", - "cProfile", - "concurrent.futures", - "configparser", - "csv", - "ctypes", - "dateutil", - "dis", - "fnmatch", - "getopt", - "glob", - "gzip", - "pdb", - "pprint", - "profile", - "pstats", - "queue", - "socket", - "sqlite3", - "ssl", - "struct", - "tarfile", - "telnetlib", - "timeit", - "tokenize", - "uuid", - "xml", - "zipfile", - "zlib", - ] -) - - -def get_imports(code: str) -> List[str]: - root = ast.parse(code) - - imports = [] - for node in ast.iter_child_nodes(root): - if isinstance(node, ast.Import): - module_names = [alias.name for alias in node.names] - elif isinstance(node, ast.ImportFrom): - module_names = [node.module] - else: - continue - - for name in module_names: - # Exclude built-in modules - if name not in BUILT_IN_MODULES: - imports.append(name) - - return imports - - -def write_requirements(code: str, requirements_filepath: str): - imports = get_imports(code) - - with open(requirements_filepath, "w") as file: - for module in imports: - file.write(module + "\n") - - -def _cmd(lang): - if lang.startswith("python") or lang in ["bash", "sh", "powershell"]: - return lang - if lang in ["shell"]: - return "sh" - if lang in ["ps1"]: - return "powershell" - raise NotImplementedError(f"{lang} not recognized in code execution") - - -def create_runner(code: str, timeout: int = 30) -> str: - """ - Create a Python script that runs the code and prints the output - """ - code = GenericEscape().escape(code) - # Create a runner script - runner = f""" -import os -import subprocess - -my_env = os.environ.copy() -my_env["PYTHONIOENCODING"] = "utf-8" - -process = subprocess.Popen( - f"python -i -q -u".split(), - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - bufsize=0, - universal_newlines=True, - env=my_env -) - -process.stdin.write("{code}" + "\\n") -process.stdin.write("exit()\\n") -process.stdin.flush() - -try: - process.wait({timeout}) -except Exception as e: - process.terminate() - -for line in iter(process.stdout.readline, ""): - print(line) - -for line in iter(process.stderr.readline, ""): - if line.startswith(">>>"): - continue - print(line) -""" - return runner - - -def _run_cmd(cmd: [str], work_dir: str, timeout: int) -> str: - if WIN32: - logger.warning("SIGALRM is not supported on Windows. No timeout will be enforced.") - result = subprocess.run( - cmd, - cwd=work_dir, - capture_output=True, - text=True, - ) - else: - with ThreadPoolExecutor(max_workers=1) as executor: - future = executor.submit( - subprocess.run, - cmd, - cwd=work_dir, - capture_output=True, - text=True, - ) - result = future.result(timeout=timeout) - return result - - -def execute_code( - code: Optional[str] = None, - timeout: Optional[int] = None, - filename: Optional[str] = None, - work_dir: Optional[str] = None, - use_docker: Optional[Union[List[str], str, bool]] = None, - lang: Optional[str] = "python", -) -> Tuple[int, str]: - """Execute code in a docker container. - This function is not tested on MacOS. - - Args: - code (Optional, str): The code to execute. - If None, the code from the file specified by filename will be executed. - Either code or filename must be provided. - timeout (Optional, int): The maximum execution time in seconds. - If None, a default timeout will be used. The default timeout is 600 seconds. On Windows, the timeout is not enforced when use_docker=False. - filename (Optional, str): The file name to save the code or where the code is stored when `code` is None. - If None, a file with a randomly generated name will be created. - The randomly generated file will be deleted after execution. - The file name must be a relative path. Relative paths are relative to the working directory. - work_dir (Optional, str): The working directory for the code execution. - If None, a default working directory will be used. - The default working directory is the "extensions" directory under - "path_to_autogen". - use_docker (Optional, list, str or bool): The docker image to use for code execution. - If a list or a str of image name(s) is provided, the code will be executed in a docker container - with the first image successfully pulled. - If None, False or empty, the code will be executed in the current environment. - Default is None, which will be converted into an empty list when docker package is available. - Expected behaviour: - - If `use_docker` is explicitly set to True and the docker package is available, the code will run in a Docker container. - - If `use_docker` is explicitly set to True but the Docker package is missing, an error will be raised. - - If `use_docker` is not set (i.e., left default to None) and the Docker package is not available, a warning will be displayed, but the code will run natively. - If the code is executed in the current environment, - the code must be trusted. - lang (Optional, str): The language of the code. Default is "python". - - Returns: - int: 0 if the code executes successfully. - str: The error message if the code fails to execute; the stdout otherwise. - """ - if all((code is None, filename is None)): - error_msg = f"Either {code=} or {filename=} must be provided." - logger.error(error_msg) - raise AssertionError(error_msg) - - # Warn if use_docker was unspecified (or None), and cannot be provided (the default). - # In this case the current behavior is to fall back to run natively, but this behavior - # is subject to change. - if use_docker is None: - if docker is None: - use_docker = False - logger.warning( - "execute_code was called without specifying a value for use_docker. Since the python docker package is not available, code will be run natively. Note: this fallback behavior is subject to change" - ) - else: - # Default to true - use_docker = True - - timeout = timeout or DEFAULT_TIMEOUT - original_filename = filename - if WIN32 and lang in ["sh", "shell"] and (not use_docker): - lang = "ps1" - if filename is None: - code_hash = md5(code.encode()).hexdigest() - # create a file with a automatically generated name - filename = f"tmp_code_{code_hash}.{'py' if lang.startswith('python') else lang}" - if work_dir is None: - WORKING_DIR = os.path.join(AIStorage.get_instance().get_myai_dir(), "tmp_code") - pathlib.Path(WORKING_DIR).mkdir(exist_ok=True) - work_dir = os.path.join(WORKING_DIR, code_hash) - pathlib.Path(work_dir).mkdir(exist_ok=True) - filepath = os.path.join(work_dir, filename) - file_dir = os.path.dirname(filepath) - os.makedirs(file_dir, exist_ok=True) - if code is not None: - write_requirements(code, os.path.join(file_dir, "requirements.txt")) - code = create_runner(code, 30) - with open(filepath, "w", encoding="utf-8") as fout: - fout.write(code) - - - # check if already running in a docker container - in_docker_container = os.path.exists("/.dockerenv") - if not use_docker or in_docker_container: - try: - env_cmd = ["python", "-m", "venv", os.path.join(file_dir, "venv")] - _run_cmd(env_cmd, file_dir, timeout) - if WIN32: - venv_path = os.path.join(file_dir, "venv", "Scripts") - else: - venv_path = os.path.join(file_dir, "venv", "bin") - pip_cmd = [os.path.join(venv_path, "python"), "-m", "pip", "install", "-r", "requirements.txt"] - _run_cmd(pip_cmd, file_dir, timeout) - # already running in a docker container - cmd = [ - os.path.join(venv_path, "python"), - f".\\{filename}" if WIN32 else filename, - ] - result = _run_cmd(cmd, file_dir, timeout) - except TimeoutError: - if original_filename is None: - shutil.rmtree(os.path.join(file_dir, "venv")) - os.remove(filepath) - os.remove(os.path.join(file_dir, "requirements.txt")) - try: - os.removedirs(file_dir) - except Exception: - pass - return 1, TIMEOUT_MSG - if original_filename is None: - shutil.rmtree(os.path.join(file_dir, "venv")) - os.remove(filepath) - os.remove(os.path.join(file_dir, "requirements.txt")) - try: - os.removedirs(file_dir) - except Exception: - pass - if result.returncode: - logs = result.stderr - if original_filename is None: - abs_path = str(pathlib.Path(filepath).absolute()) - logs = logs.replace(str(abs_path), "").replace(filename, "") - else: - abs_path = str(pathlib.Path(work_dir).absolute()) + PATH_SEPARATOR - logs = logs.replace(str(abs_path), "") - else: - logs = result.stdout - return result.returncode, logs - - # create a docker client - client = docker.from_env() - image_list = ( - ["python:3-alpine", "python:3", "python:3-windowsservercore"] - if use_docker is True - else [use_docker] - if isinstance(use_docker, str) - else use_docker - ) - for image in image_list: - # check if the image exists - try: - client.images.get(image) - break - except docker.errors.ImageNotFound: - # pull the image - logger.info("Pulling image", image) - try: - client.images.pull(image, stream=True, decode=True) - break - except docker.errors.DockerException as e: - logger.error("Failed to pull image", image) - logger.exception(e) - # get a randomized str based on current time to wrap the exit code - exit_code_str = f"exitcode{time.time()}" - start_str = f'start{time.time()}' - abs_path = pathlib.Path(work_dir).absolute() - cmd = [ - "sh", - "-c", - f"pip install --quiet -r requirements.txt; echo -n {start_str}; {_cmd(lang)} {filename}; exit_code=$?; echo -n {exit_code_str}; echo -n $exit_code; echo {exit_code_str};", - ] - # create a docker container - container = client.containers.run( - image, - command=cmd, - working_dir="/workspace", - detach=True, - # get absolute path to the working directory - volumes={abs_path: {"bind": "/workspace", "mode": "rw"}}, - ) - start_time = time.time() - while container.status != "exited" and time.time() - start_time < timeout: - # Reload the container object - container.reload() - if container.status != "exited": - container.stop() - container.remove() - if original_filename is None: - os.remove(filepath) - return 1, TIMEOUT_MSG, image - # get the container logs - logs: str = container.logs().decode("utf-8").rstrip() - start_pos = logs.find(start_str) - if start_pos != -1: - logs = logs[start_pos + len(start_str):] - # # commit the image - # tag = filename.replace("/", "") - # container.commit(repository="python", tag=tag) - # remove the container - container.remove() - # check if the code executed successfully - exit_code = container.attrs["State"]["ExitCode"] - if exit_code == 0: - # extract the exit code from the logs - pattern = re.compile(f"{exit_code_str}(\\d+){exit_code_str}") - match = pattern.search(logs) - exit_code = 1 if match is None else int(match.group(1)) - # remove the exit code from the logs - logs = logs if match is None else pattern.sub("", logs) - - if original_filename is None: - os.remove(filepath) - os.remove(os.path.join(file_dir, "requirements.txt")) - os.removedirs(file_dir) - if exit_code: - logs = logs.replace(f"/workspace/{filename if original_filename is None else ''}", "") - # return the exit code, logs and image - return exit_code, logs - diff --git a/src/aios_kernel/code_interpreter_function.py b/src/aios_kernel/code_interpreter_function.py deleted file mode 100644 index 0bf366c..0000000 --- a/src/aios_kernel/code_interpreter_function.py +++ /dev/null @@ -1,41 +0,0 @@ -from typing import Dict - -from aios_kernel.ai_function import AIFunction -from aios_kernel.code_interpreter import execute_code - - -class CodeInterpreterFunction(AIFunction): - def __init__(self): - self.func_id = "code_interpreter" - self.description = "execute python code" - - def get_name(self) -> str: - return self.func_id - - def get_description(self) -> str: - return self.description - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "code": {"type": "string", "description": "python code"} - } - } - - async def execute(self, **kwargs) -> str: - code = kwargs.get("code") - ret_code, result = execute_code(code=code) - if ret_code == 0: - return result.strip() - else: - return result.strip() - - def is_local(self) -> bool: - return True - - def is_in_zone(self) -> bool: - return True - - def is_ready_only(self) -> bool: - return False diff --git a/src/aios_kernel/duckduckgo_text_search_function.py b/src/aios_kernel/duckduckgo_text_search_function.py deleted file mode 100644 index 154f761..0000000 --- a/src/aios_kernel/duckduckgo_text_search_function.py +++ /dev/null @@ -1,52 +0,0 @@ -import json -from typing import Dict - -from aios_kernel.ai_function import AIFunction -from duckduckgo_search import AsyncDDGS - - -class DuckDuckGoTextSearchFunction(AIFunction): - def __init__(self): - self.name = "duckduckgo_text_search" - self.description = "Search text from duckduckgo.com" - self.region = "wt-wt" - self.safesearch = "moderate" - self.time = "y" - self.max_results = 5 - - def get_name(self) -> str: - return self.name - - def get_description(self) -> str: - return self.description - - def get_parameters(self) -> Dict: - return {"type": "object", - "properties": { - "query": {"type": "string", "description": "The query to search for."} - } - } - - async def execute(self, **kwargs) -> str: - query = kwargs.get("query") - - async with AsyncDDGS() as ddgs: - results = [r async for r in ddgs.text( - query, - region=self.region, - safesearch=self.safesearch, - timelimit=self.time, - backend="api", - max_results=self.max_results - )] - - return json.dumps(results) - - def is_local(self) -> bool: - return True - - def is_in_zone(self) -> bool: - return True - - def is_ready_only(self) -> bool: - return False diff --git a/src/aios_kernel/sql_database.py b/src/aios_kernel/sql_database.py deleted file mode 100644 index efee9b4..0000000 --- a/src/aios_kernel/sql_database.py +++ /dev/null @@ -1,493 +0,0 @@ -""" -Taken from: langchain -SQLAlchemy wrapper around a database. -""" -from __future__ import annotations -import os - -import warnings -from typing import Any, Dict, Iterable, List, Literal, Optional, Sequence, Union - -import sqlalchemy -from sqlalchemy import MetaData, Table, create_engine, inspect, select, text -from sqlalchemy.engine import Engine -from sqlalchemy.exc import ProgrammingError, SQLAlchemyError -from sqlalchemy.schema import CreateTable -from sqlalchemy.types import NullType - - -def get_from_env(key: str, env_key: str, default: Optional[str] = None) -> str: - """Get a value from a dictionary or an environment variable.""" - if env_key in os.environ and os.environ[env_key]: - return os.environ[env_key] - elif default is not None: - return default - else: - raise ValueError( - f"Did not find {key}, please add an environment variable" - f" `{env_key}` which contains it, or pass" - f" `{key}` as a named parameter." - ) - - -def _format_index(index: sqlalchemy.engine.interfaces.ReflectedIndex) -> str: - return ( - f'Name: {index["name"]}, Unique: {index["unique"]},' - f' Columns: {str(index["column_names"])}' - ) - - -def truncate_word(content: Any, *, length: int, suffix: str = "...") -> str: - """ - Truncate a string to a certain number of words, based on the max string - length. - """ - - if not isinstance(content, str) or length <= 0: - return content - - if len(content) <= length: - return content - - return content[: length - len(suffix)].rsplit(" ", 1)[0] + suffix - - -class SQLDatabase: - """SQLAlchemy wrapper around a database.""" - - def __init__( - self, - engine: Engine, - schema: Optional[str] = None, - metadata: Optional[MetaData] = None, - ignore_tables: Optional[List[str]] = None, - include_tables: Optional[List[str]] = None, - sample_rows_in_table_info: int = 3, - indexes_in_table_info: bool = False, - custom_table_info: Optional[dict] = None, - view_support: bool = False, - max_string_length: int = 300, - ): - """Create engine from database URI.""" - self._engine = engine - self._schema = schema - if include_tables and ignore_tables: - raise ValueError("Cannot specify both include_tables and ignore_tables") - - self._inspector = inspect(self._engine) - - # including view support by adding the views as well as tables to the all - # tables list if view_support is True - self._all_tables = set( - self._inspector.get_table_names(schema=schema) - + (self._inspector.get_view_names(schema=schema) if view_support else []) - ) - - self._include_tables = set(include_tables) if include_tables else set() - if self._include_tables: - missing_tables = self._include_tables - self._all_tables - if missing_tables: - raise ValueError( - f"include_tables {missing_tables} not found in database" - ) - self._ignore_tables = set(ignore_tables) if ignore_tables else set() - if self._ignore_tables: - missing_tables = self._ignore_tables - self._all_tables - if missing_tables: - raise ValueError( - f"ignore_tables {missing_tables} not found in database" - ) - usable_tables = self.get_usable_table_names() - self._usable_tables = set(usable_tables) if usable_tables else self._all_tables - - if not isinstance(sample_rows_in_table_info, int): - raise TypeError("sample_rows_in_table_info must be an integer") - - self._sample_rows_in_table_info = sample_rows_in_table_info - self._indexes_in_table_info = indexes_in_table_info - - self._custom_table_info = custom_table_info - if self._custom_table_info: - if not isinstance(self._custom_table_info, dict): - raise TypeError( - "table_info must be a dictionary with table names as keys and the " - "desired table info as values" - ) - # only keep the tables that are also present in the database - intersection = set(self._custom_table_info).intersection(self._all_tables) - self._custom_table_info = dict( - (table, self._custom_table_info[table]) - for table in self._custom_table_info - if table in intersection - ) - - self._max_string_length = max_string_length - - self._metadata = metadata or MetaData() - # including view support if view_support = true - self._metadata.reflect( - views=view_support, - bind=self._engine, - only=list(self._usable_tables), - schema=self._schema, - ) - - @classmethod - def from_uri( - cls, database_uri: str, engine_args: Optional[dict] = None, **kwargs: Any - ) -> SQLDatabase: - """Construct a SQLAlchemy engine from URI.""" - _engine_args = engine_args or {} - return cls(create_engine(database_uri, **_engine_args), **kwargs) - - @classmethod - def from_databricks( - cls, - catalog: str, - schema: str, - host: Optional[str] = None, - api_token: Optional[str] = None, - warehouse_id: Optional[str] = None, - cluster_id: Optional[str] = None, - engine_args: Optional[dict] = None, - **kwargs: Any, - ) -> SQLDatabase: - """ - Class method to create an SQLDatabase instance from a Databricks connection. - This method requires the 'databricks-sql-connector' package. If not installed, - it can be added using `pip install databricks-sql-connector`. - - Args: - catalog (str): The catalog name in the Databricks database. - schema (str): The schema name in the catalog. - host (Optional[str]): The Databricks workspace hostname, excluding - 'https://' part. If not provided, it attempts to fetch from the - environment variable 'DATABRICKS_HOST'. If still unavailable and if - running in a Databricks notebook, it defaults to the current workspace - hostname. Defaults to None. - api_token (Optional[str]): The Databricks personal access token for - accessing the Databricks SQL warehouse or the cluster. If not provided, - it attempts to fetch from 'DATABRICKS_TOKEN'. If still unavailable - and running in a Databricks notebook, a temporary token for the current - user is generated. Defaults to None. - warehouse_id (Optional[str]): The warehouse ID in the Databricks SQL. If - provided, the method configures the connection to use this warehouse. - Cannot be used with 'cluster_id'. Defaults to None. - cluster_id (Optional[str]): The cluster ID in the Databricks Runtime. If - provided, the method configures the connection to use this cluster. - Cannot be used with 'warehouse_id'. If running in a Databricks notebook - and both 'warehouse_id' and 'cluster_id' are None, it uses the ID of the - cluster the notebook is attached to. Defaults to None. - engine_args (Optional[dict]): The arguments to be used when connecting - Databricks. Defaults to None. - **kwargs (Any): Additional keyword arguments for the `from_uri` method. - - Returns: - SQLDatabase: An instance of SQLDatabase configured with the provided - Databricks connection details. - - Raises: - ValueError: If 'databricks-sql-connector' is not found, or if both - 'warehouse_id' and 'cluster_id' are provided, or if neither - 'warehouse_id' nor 'cluster_id' are provided and it's not executing - inside a Databricks notebook. - """ - try: - from databricks import sql # noqa: F401 - except ImportError: - raise ValueError( - "databricks-sql-connector package not found, please install with" - " `pip install databricks-sql-connector`" - ) - context = None - try: - from dbruntime.databricks_repl_context import get_context - - context = get_context() - except ImportError: - pass - - default_host = context.browserHostName if context else None - if host is None: - host = get_from_env("host", "DATABRICKS_HOST", default_host) - - default_api_token = context.apiToken if context else None - if api_token is None: - api_token = get_from_env("api_token", "DATABRICKS_TOKEN", default_api_token) - - if warehouse_id is None and cluster_id is None: - if context: - cluster_id = context.clusterId - else: - raise ValueError( - "Need to provide either 'warehouse_id' or 'cluster_id'." - ) - - if warehouse_id and cluster_id: - raise ValueError("Can't have both 'warehouse_id' or 'cluster_id'.") - - if warehouse_id: - http_path = f"/sql/1.0/warehouses/{warehouse_id}" - else: - http_path = f"/sql/protocolv1/o/0/{cluster_id}" - - uri = ( - f"databricks://token:{api_token}@{host}?" - f"http_path={http_path}&catalog={catalog}&schema={schema}" - ) - return cls.from_uri(database_uri=uri, engine_args=engine_args, **kwargs) - - @classmethod - def from_cnosdb( - cls, - url: str = "127.0.0.1:8902", - user: str = "root", - password: str = "", - tenant: str = "cnosdb", - database: str = "public", - ) -> SQLDatabase: - """ - Class method to create an SQLDatabase instance from a CnosDB connection. - This method requires the 'cnos-connector' package. If not installed, it - can be added using `pip install cnos-connector`. - - Args: - url (str): The HTTP connection host name and port number of the CnosDB - service, excluding "http://" or "https://", with a default value - of "127.0.0.1:8902". - user (str): The username used to connect to the CnosDB service, with a - default value of "root". - password (str): The password of the user connecting to the CnosDB service, - with a default value of "". - tenant (str): The name of the tenant used to connect to the CnosDB service, - with a default value of "cnosdb". - database (str): The name of the database in the CnosDB tenant. - - Returns: - SQLDatabase: An instance of SQLDatabase configured with the provided - CnosDB connection details. - """ - try: - from cnosdb_connector import make_cnosdb_langchain_uri - - uri = make_cnosdb_langchain_uri(url, user, password, tenant, database) - return cls.from_uri(database_uri=uri) - except ImportError: - raise ValueError( - "cnos-connector package not found, please install with" - " `pip install cnos-connector`" - ) - - @property - def dialect(self) -> str: - """Return string representation of dialect to use.""" - return self._engine.dialect.name - - def get_usable_table_names(self) -> Iterable[str]: - """Get names of tables available.""" - if self._include_tables: - return sorted(self._include_tables) - return sorted(self._all_tables - self._ignore_tables) - - def get_table_names(self) -> Iterable[str]: - """Get names of tables available.""" - warnings.warn( - "This method is deprecated - please use `get_usable_table_names`." - ) - return self.get_usable_table_names() - - @property - def table_info(self) -> str: - """Information about all tables in the database.""" - return self.get_table_info() - - def get_table_info(self, table_names: Optional[List[str]] = None) -> str: - """Get information about specified tables. - - Follows best practices as specified in: Rajkumar et al, 2022 - (https://arxiv.org/abs/2204.00498) - - If `sample_rows_in_table_info`, the specified number of sample rows will be - appended to each table description. This can increase performance as - demonstrated in the paper. - """ - all_table_names = self.get_usable_table_names() - if table_names is not None: - missing_tables = set(table_names).difference(all_table_names) - if missing_tables: - raise ValueError(f"table_names {missing_tables} not found in database") - all_table_names = table_names - - meta_tables = [ - tbl - for tbl in self._metadata.sorted_tables - if tbl.name in set(all_table_names) - and not (self.dialect == "sqlite" and tbl.name.startswith("sqlite_")) - ] - - tables = [] - for table in meta_tables: - if self._custom_table_info and table.name in self._custom_table_info: - tables.append(self._custom_table_info[table.name]) - continue - - # Ignore JSON datatyped columns - for k, v in table.columns.items(): - if type(v.type) is NullType: - table._columns.remove(v) - - # add create table command - create_table = str(CreateTable(table).compile(self._engine)) - table_info = f"{create_table.rstrip()}" - has_extra_info = ( - self._indexes_in_table_info or self._sample_rows_in_table_info - ) - if has_extra_info: - table_info += "\n\n/*" - if self._indexes_in_table_info: - table_info += f"\n{self._get_table_indexes(table)}\n" - if self._sample_rows_in_table_info: - table_info += f"\n{self._get_sample_rows(table)}\n" - if has_extra_info: - table_info += "*/" - tables.append(table_info) - tables.sort() - final_str = "\n\n".join(tables) - return final_str - - def _get_table_indexes(self, table: Table) -> str: - indexes = self._inspector.get_indexes(table.name) - indexes_formatted = "\n".join(map(_format_index, indexes)) - return f"Table Indexes:\n{indexes_formatted}" - - def _get_sample_rows(self, table: Table) -> str: - # build the select command - command = select(table).limit(self._sample_rows_in_table_info) - - # save the columns in string format - columns_str = "\t".join([col.name for col in table.columns]) - - try: - # get the sample rows - with self._engine.connect() as connection: - sample_rows_result = connection.execute(command) # type: ignore - # shorten values in the sample rows - sample_rows = list( - map(lambda ls: [str(i)[:100] for i in ls], sample_rows_result) - ) - - # save the sample rows in string format - sample_rows_str = "\n".join(["\t".join(row) for row in sample_rows]) - - # in some dialects when there are no rows in the table a - # 'ProgrammingError' is returned - except ProgrammingError: - sample_rows_str = "" - - return ( - f"{self._sample_rows_in_table_info} rows from {table.name} table:\n" - f"{columns_str}\n" - f"{sample_rows_str}" - ) - - def _execute( - self, - command: str, - fetch: Union[Literal["all"], Literal["one"]] = "all", - ) -> Sequence[Dict[str, Any]]: - """ - Executes SQL command through underlying engine. - - If the statement returns no rows, an empty list is returned. - """ - with self._engine.begin() as connection: - if self._schema is not None: - if self.dialect == "snowflake": - connection.exec_driver_sql( - "ALTER SESSION SET search_path = %s", (self._schema,) - ) - elif self.dialect == "bigquery": - connection.exec_driver_sql("SET @@dataset_id=?", (self._schema,)) - elif self.dialect == "mssql": - pass - elif self.dialect == "trino": - connection.exec_driver_sql("USE ?", (self._schema,)) - elif self.dialect == "duckdb": - # Unclear which parameterized argument syntax duckdb supports. - # The docs for the duckdb client say they support multiple, - # but `duckdb_engine` seemed to struggle with all of them: - # https://github.com/Mause/duckdb_engine/issues/796 - connection.exec_driver_sql(f"SET search_path TO {self._schema}") - elif self.dialect == "oracle": - connection.exec_driver_sql( - f"ALTER SESSION SET CURRENT_SCHEMA = {self._schema}" - ) - else: # postgresql and other compatible dialects - connection.exec_driver_sql("SET search_path TO %s", (self._schema,)) - cursor = connection.execute(text(command)) - if cursor.returns_rows: - if fetch == "all": - result = [x._asdict() for x in cursor.fetchall()] - elif fetch == "one": - first_result = cursor.fetchone() - result = [] if first_result is None else [first_result._asdict()] - else: - raise ValueError("Fetch parameter must be either 'one' or 'all'") - return result - return [] - - def run( - self, - command: str, - fetch: Union[Literal["all"], Literal["one"]] = "all", - ) -> str: - """Execute a SQL command and return a string representing the results. - - If the statement returns rows, a string of the results is returned. - If the statement returns no rows, an empty string is returned. - """ - result = self._execute(command, fetch) - # Convert columns values to string to avoid issues with sqlalchemy - # truncating text - res = [ - tuple(truncate_word(c, length=self._max_string_length) for c in r.values()) - for r in result - ] - if not res: - return "" - else: - return str(res) - - def get_table_info_no_throw(self, table_names: Optional[List[str]] = None) -> str: - """Get information about specified tables. - - Follows best practices as specified in: Rajkumar et al, 2022 - (https://arxiv.org/abs/2204.00498) - - If `sample_rows_in_table_info`, the specified number of sample rows will be - appended to each table description. This can increase performance as - demonstrated in the paper. - """ - try: - return self.get_table_info(table_names) - except ValueError as e: - """Format the error message""" - return f"Error: {e}" - - def run_no_throw( - self, - command: str, - fetch: Union[Literal["all"], Literal["one"]] = "all", - ) -> str: - """Execute a SQL command and return a string representing the results. - - If the statement returns rows, a string of the results is returned. - If the statement returns no rows, an empty string is returned. - - If the statement throws an error, the error message is returned. - """ - try: - return self.run(command, fetch) - except SQLAlchemyError as e: - """Format the error message""" - return f"Error: {e}" diff --git a/src/aios_kernel/sql_database_function.py b/src/aios_kernel/sql_database_function.py deleted file mode 100644 index 2e208e6..0000000 --- a/src/aios_kernel/sql_database_function.py +++ /dev/null @@ -1,112 +0,0 @@ -from datetime import timedelta, datetime -from typing import Dict - -from cachetools import TLRUCache, cached - -from aios_kernel.ai_function import AIFunction -from aios_kernel.sql_database import SQLDatabase, get_from_env - - -def _my_ttu(_key, _value, now): - return now + timedelta(seconds=600) - - -database_cache = TLRUCache(ttu=_my_ttu, maxsize=10000, timer=datetime.now) - - -@cached(cache=database_cache) -def get_database(uri: str) -> SQLDatabase: - return SQLDatabase.from_uri(uri) - - -class GetTableInfosFunction(AIFunction): - def __init__(self): - super().__init__() - self.name = "get_table_infos" - self.description = "Get table informations in the database" - - def get_name(self) -> str: - return self.name - - def get_description(self) -> str: - return self.description - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "database_url": {"type": "string", "description": "Database URL,Can be set to None"}, - } - } - - async def execute(self, **kwargs) -> str: - database_url: str = kwargs.get("database_url") - if (database_url is None - or database_url.strip() == "" - or database_url.strip().lower() == "none" - or database_url.strip().lower() == "null"): - database_url = get_from_env(key="database url", env_key="DATABASE_URL") - if database_url is None: - return "error: database_url is None" - database = get_database(database_url) - tables = database.get_usable_table_names() - table_infos = database.get_table_info(tables) - return table_infos - - def is_local(self) -> bool: - return True - - def is_in_zone(self) -> bool: - return True - - def is_ready_only(self) -> bool: - return False - - -class ExecuteSqlFunction(AIFunction): - def __init__(self): - super().__init__() - self.name = "execute_sql" - self.description = """ - Input to this function is a detailed and correct SQL query, output is a result from the database. - If the query is not correct, an error message will be returned. - If an error is returned, rewrite the query, check the query, and try again. - """ - - def get_name(self) -> str: - return self.name - - def get_description(self) -> str: - return self.description - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "database_url": {"type": "string", "description": "Database URL,Can be set to None"}, - "sql": {"type": "string", "description": "SQL to execute"} - } - } - - async def execute(self, **kwargs) -> str: - database_url = kwargs.get("database_url") - if (database_url is None - or database_url.strip() == "" - or database_url.strip().lower() == "none" - or database_url.strip().lower() == "null"): - database_url = get_from_env(key="database url", env_key="DATABASE_URL") - if database_url is None: - return "error: database_url is None" - sql = kwargs.get("sql") - - database = get_database(database_url) - return database.run_no_throw(sql) - - def is_local(self) -> bool: - return True - - def is_in_zone(self) -> bool: - return True - - def is_ready_only(self) -> bool: - return False diff --git a/src/component/common_environment/shell.py b/src/component/common_environment/shell.py index c8209fb..173fc59 100644 --- a/src/component/common_environment/shell.py +++ b/src/component/common_environment/shell.py @@ -1,6 +1,6 @@ import os from typing import Any,List,Dict -from aios import AgentMsg,AgentTodo,AgentPrompt +from aios import AgentMsg,AgentTodo,LLMPrompt from aios import SimpleAIFunction, SimpleAIOperation from aios import SimpleEnvironment diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py index 075d12b..1707d2a 100644 --- a/src/component/mail_environment/issue.py +++ b/src/component/mail_environment/issue.py @@ -218,7 +218,7 @@ class IssueAgent(CustomAIAgent): super().__init__(agent_id, llm_model_name, max_token_size) -class IssueParserEnvironment(Environment): +class IssueParserEnvironment(SimpleEnvironment): def __init__(self, env_id: str, storage: IssueStorage) -> None: super().__init__(env_id) self.storage = storage @@ -305,7 +305,7 @@ class IssueParser: mail_desc = Mail.prompt_desc() issue_desc = Issue.prompt_desc() - prompt = AgentPrompt() + prompt = LLMPrompt() prompt.system_message = {"role": "system", "content": f''' I'm a CEO of a company named 巴克云; You'ar my assistant, and you should help me to manage my issues. Issues is a concept in software development of this company, but I use it to manage my work. I'll give you a mail in json format, {mail_desc}; diff --git a/src/component/openai_node/open_ai_node.py b/src/component/openai_node/open_ai_node.py index a072309..e8dc862 100644 --- a/src/component/openai_node/open_ai_node.py +++ b/src/component/openai_node/open_ai_node.py @@ -1,3 +1,4 @@ +import asyncio import openai from openai import AsyncOpenAI import os @@ -237,7 +238,8 @@ class OpenAI_ComputeNode(ComputeNode): return result logger.info(f"openai response: {resp}") - if mode_name == "gpt-4-vision-preview": + #TODO: gpt-4v api is image_2_text ? + if mode_name == "gpt-4-vision-preview": status_code = resp.choices[0].finish_reason if status_code is None: status_code = resp.choices[0].finish_details['type'] @@ -265,6 +267,7 @@ class OpenAI_ComputeNode(ComputeNode): if token_usage: result.result_refers["token_usage"] = token_usage + logger.info(f"openai success response: {result.result_str}") return result case _: diff --git a/src/component/tg_tunnel.py b/src/component/tg_tunnel.py index 4406591..27b6d16 100644 --- a/src/component/tg_tunnel.py +++ b/src/component/tg_tunnel.py @@ -87,7 +87,7 @@ class TelegramTunnel(AgentTunnel): async def _run_app(): try: update_id = (await self.bot.get_updates())[0].update_id - except IndexError: + except Exception as e: update_id = None except Exception as e: logger.error(f"tg_tunnel error:{e}") @@ -97,7 +97,11 @@ class TelegramTunnel(AgentTunnel): #logger.info("listening for new messages...") while True: try: - update_id = await self._do_process_raw_message(self.bot, update_id) + if update_id: + update_id = await self._do_process_raw_message(self.bot, update_id) + else: + update_id = (await self.bot.get_updates())[0].update_id + except NetworkError: await asyncio.sleep(1) except Forbidden: diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index 3268958..4da3885 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -2,7 +2,7 @@ import logging import toml import os -from aios import AIStorage,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask +from aios import AIStorage,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask,Workflow from agent_manager import AgentManager logger = logging.getLogger(__name__) diff --git a/src/requirements.txt b/src/requirements.txt index 9413232..1e5cbbf 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -47,7 +47,7 @@ mpmath>=1.3.0 multidict>=6.0.4 numpy>=1.25.2 onnxruntime>=1.15.1 -openai>=1.0.0 + overrides>=7.4.0 packaging>=23.1 pandas>=2.1.0 @@ -139,8 +139,9 @@ sentence-transformers==2.2.2 tiktoken markdown PyPDF2 -srt==3.5.3 -webvtt-py==0.4.6 +srt +webvtt-py +openai docker generic_escape duckduckgo-search @@ -152,3 +153,4 @@ oracledb html2text docx2txt opencv-python + diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 4c0c076..23485ec 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -40,7 +40,7 @@ from sd_node import * from st_node import * from agent_manager import AgentManager -# from workflow_manager import WorkflowManager +from workflow_manager import WorkflowManager from knowledge_manager import KnowledgePipelineManager from tg_tunnel import TelegramTunnel from email_tunnel import EmailTunnel @@ -148,9 +148,9 @@ class AIOS_Shell: if await AgentManager.get_instance().initial() is not True: logger.error("agent manager initial failed!") return False - # if await WorkflowManager.get_instance().initial() is not True: - # logger.error("workflow manager initial failed!") - # return False + if await WorkflowManager.get_instance().initial() is not True: + logger.error("workflow manager initial failed!") + return False open_ai_node = OpenAI_ComputeNode.get_instance() if await open_ai_node.initial() is not True: