From 7fd3749a4069271f500f9f7b651feb726d382cfd Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sun, 5 Nov 2023 15:21:27 -0800 Subject: [PATCH] 1) Fix bugs. 2) openai_node support async/await. --- rootfs/agents/JarvisPlus/agent.toml | 2 +- src/aios_kernel/agent.py | 14 +++++++----- src/aios_kernel/agent_base.py | 19 ++++++++++------ src/aios_kernel/compute_kernel.py | 4 +++- src/aios_kernel/open_ai_node.py | 12 ++++++---- src/aios_kernel/workspace_env.py | 34 +++++++++++++++++++++-------- src/requirements.txt | 6 ++--- 7 files changed, 61 insertions(+), 30 deletions(-) diff --git a/rootfs/agents/JarvisPlus/agent.toml b/rootfs/agents/JarvisPlus/agent.toml index 9797af7..19c621c 100644 --- a/rootfs/agents/JarvisPlus/agent.toml +++ b/rootfs/agents/JarvisPlus/agent.toml @@ -19,7 +19,7 @@ The types of TODO I can handle include: I receive a TODO described in json format, I will handle it according to the following rules: - Determine whether I have the ability to handle the TODO independently. If not, I will try to break the TODO down into smaller sub-TODOs, or hand it over to someone more suitable that I know. - I will plan the steps to solve the TODO in combination with known information, and break down the generalized TODO into more specific sub-todos. The title of the sub-todo should contain step number like #1, #2 -- sub-todo must set parent +- Sub-todo must set parent, The maximum depth of sub-todo is 4. - A specific sub-todo refers to a task that can be completed in one execution within my ability range. - After each execution, I will decide whether to update the status of the TODO. And use op:'update_todo' to update when necessary. diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index f44d18c..cf3ae75 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -1,3 +1,4 @@ +import traceback from typing import Optional from asyncio import Queue @@ -9,6 +10,7 @@ import json import shlex import datetime import copy +import sys from .agent_base import AgentMsg, AgentMsgStatus, AgentMsgType,FunctionItem,LLMResult,AgentPrompt,AgentReport,AgentTodo,AgentGoal,AgentTodoResult,AgentWorkLog from .chatsession import AIChatSession @@ -659,6 +661,7 @@ class AIAgent: # 尝试完成自己的TOOD （不依赖任何其他Agnet） async def do_my_work(self) -> None: workspace : WorkspaceEnvironment = self.get_workspace_by_msg(None) + logger.info(f"agent {self.agent_id} do my work start!") # review todo能更整体的思考一次todo的优先级 if await self.need_review_todos(): @@ -680,6 +683,8 @@ class AIAgent: if todo.state == "done": await self._llm_check_todo(todo,workspace) self.agent_energy -= 1 + + logger.info(f"agent {self.agent_id} do my work done!") def get_review_todo_prompt(self) -> AgentPrompt: return self.review_todo_prompt @@ -981,9 +986,6 @@ class AIAgent: return task_result - async def execute_op_list(self,oplist:list,workspace:WorkspaceEnvironment): - pass - def need_work(self) -> bool: return True @@ -1002,7 +1004,7 @@ class AIAgent: # agent loop async def _on_timer(self): while True: - await asyncio.sleep(1) + await asyncio.sleep(15) try: now = time.time() if self.last_recover_time is None: @@ -1030,8 +1032,10 @@ class AIAgent: # if self.need_self_learn(): await self.do_self_learn() + except Exception as e: - logger.error(f"agent {self.agent_id} on timer error:{e}") + tb_str = traceback.format_exc() + logger.error(f"agent {self.agent_id} on timer error:{tb_str},{e}") continue diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index 6870488..e1e2c6e 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -374,12 +374,13 @@ class AgentTodo: 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.due_date = time.time() + 3600 * 24 * 2 self.state = "pending" self.depend_todo_ids = [] - self.sub_todos = [] + self.sub_todos = {} self.need_check = True self.result : ComputeTaskResult = None @@ -406,7 +407,7 @@ class AgentTodo: due_date = json_obj.get("due_date") if due_date: todo.due_date = datetime.fromisoformat(due_date).timestamp() - #todo.todo_path = json_obj.get("todo_path") + todo.todo_path = json_obj.get("todo_path") todo.depend_todo_ids = json_obj.get("depend_todo_ids") todo.need_check = json_obj.get("need_check") #todo.result = json_obj.get("result") @@ -414,7 +415,9 @@ class AgentTodo: todo.worker = json_obj.get("worker") todo.checker = json_obj.get("checker") todo.createor = json_obj.get("createor") - #todo.retry_count = json_obj.get("retry_count") + if json_obj.get("retry_count"): + todo.retry_count = json_obj.get("retry_count") + todo.raw_obj = json_obj return todo @@ -422,6 +425,7 @@ class AgentTodo: def to_dict(self) -> dict: result = {} result["id"] = self.todo_id + result["todo_path"] = self.todo_path #result["parent_id"] = self.parent_id result["title"] = self.title result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() @@ -434,11 +438,12 @@ class AgentTodo: result["createor"] = self.createor result["retry_count"] = self.retry_count - return result + return result def can_do(self) -> bool: - for sub_todo in self.sub_todos: - if sub_todo.state != "done": - return False + for sub_todo in self.sub_todos.values(): + if sub_todo.state: + if sub_todo.state != "done": + return False now = datetime.now().timestamp() time_diff = now - self.due_date diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index f88a835..eb09574 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -151,7 +151,9 @@ class ComputeKernel: time_out_result = ComputeTaskResult() time_out_result.result_code = ComputeTaskResultCode.TIMEOUT time_out_result.set_from_task(task_req) - ## craete timeout task_result + task_req.result = time_out_result + return time_out_result + async def do_llm_completion(self, prompt: AgentPrompt, mode_name: Optional[str] = None, max_token: int = 0, inner_functions = None) -> str: task_req = self.llm_completion(prompt, mode_name, max_token,inner_functions) diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index e529b92..206d223 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -4,11 +4,13 @@ import asyncio from asyncio import Queue import logging import json +import aiohttp from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType,ComputeTaskResultCode from .compute_node import ComputeNode from .storage import AIStorage,UserConfig + logger = logging.getLogger(__name__) @@ -57,7 +59,8 @@ class OpenAI_ComputeNode(ComputeNode): async def remove_task(self, task_id: str): pass - def _run_task(self, task: ComputeTask): + + async def _run_task(self, task: ComputeTask): task.state = ComputeTaskState.RUNNING result = ComputeTaskResult() @@ -110,16 +113,17 @@ class OpenAI_ComputeNode(ComputeNode): max_token_size = 4000 result_token = max_token_size + try: if llm_inner_functions is None: logger.info(f"call openai {mode_name} prompts: {prompts}") - resp = openai.ChatCompletion.create(model=mode_name, + resp = await openai.ChatCompletion.acreate(model=mode_name, messages=prompts, #max_tokens=result_token, temperature=0.7) else: logger.info(f"call openai {mode_name} prompts: \n\t {prompts} \nfunctions: \n\t{json.dumps(llm_inner_functions)}") - resp = openai.ChatCompletion.create(model=mode_name, + resp = await openai.ChatCompletion.acreate(model=mode_name, messages=prompts, functions=llm_inner_functions, #max_tokens=result_token, @@ -171,7 +175,7 @@ class OpenAI_ComputeNode(ComputeNode): while True: task = await self.task_queue.get() logger.info(f"openai_node get task: {task.display()}") - result = self._run_task(task) + result = await self._run_task(task) if result is not None: task.state = ComputeTaskState.DONE task.result = result diff --git a/src/aios_kernel/workspace_env.py b/src/aios_kernel/workspace_env.py index a55a44b..8c67446 100644 --- a/src/aios_kernel/workspace_env.py +++ b/src/aios_kernel/workspace_env.py @@ -58,7 +58,10 @@ class WorkspaceEnvironment(Environment): if op["op"] == "create": return await self.create(op["path"],op["content"]) elif op["op"] == "write_file": - return await self.write(op["path"],op["content"],op["mode"]) + is_append = op.get("is_append") + if is_append is None: + is_append = False + return await self.write(op["path"],op["content"],is_append) elif op["op"] == "delete": return await self.delete(op["path"]) elif op["op"] == "rename": @@ -173,13 +176,14 @@ class WorkspaceEnvironment(Environment): continue todo_count = todo_count + 1 - str_result = str_result + f"{' '*(4-deep)}└─{entry.name}\n" + str_result = str_result + f"{' '*(4-deep)}{entry.name}\n" await scan_dir(entry.path,deep-1) await scan_dir(directory_path,deep) return str_result,todo_count async def get_todo_list(self,agent_id:str,path:str = None)->List[AgentTodo]: + logger.info("get_todo_list:%s,%s",agent_id,path) if path: directory_path = self.root_path + "/todos/" + path else: @@ -208,7 +212,7 @@ class WorkspaceEnvironment(Environment): if todo: if parent: - parent.sub_todos.append(todo) + parent.sub_todos[todo.todo_id] = todo result_list.append(todo) await scan_dir(entry.path,deep + 1,todo) @@ -218,11 +222,11 @@ class WorkspaceEnvironment(Environment): await scan_dir(directory_path,0) #sort by rank result_list.sort(key=lambda x:(x.rank,x.title)) - + logger.info("get_todo_list return,todolist.length() is %d",len(result_list)) return result_list async def get_todo_by_fullpath(self,path:str) -> AgentTodo: - logger.info("get_todo_by_fullpath:%s",path) + #logger.info("get_todo_by_fullpath:%s",path) detail_path = path + "/detail" async with aiofiles.open(detail_path, mode='r', encoding="utf-8") as f: @@ -231,7 +235,10 @@ class WorkspaceEnvironment(Environment): todo_dict = json.loads(content) result_todo = AgentTodo.from_dict(todo_dict) if result_todo: - result_todo.todo_path = path + relative_path = os.path.relpath(path, self.root_path + "/todos/") + if not relative_path.startswith('/'): + relative_path = '/' + relative_path + result_todo.todo_path = relative_path self.known_todo[result_todo.todo_id] = result_todo else: logger.error("get_todo_by_path:%s,parse failed!",path) @@ -243,16 +250,25 @@ class WorkspaceEnvironment(Environment): async def create_todo(self,parent_id:str,todo:AgentTodo) -> None: if parent_id: + if parent_id not in self.known_todo: + logger.error("create_todo failed: parent_id not found!") + return False + parent_path = self.known_todo.get(parent_id).todo_path - dir_path = f"{parent_path}/{todo.title}" + todo_path = f"{parent_path}/{todo.title}" else: - dir_path = f"{self.root_path}/todos/{todo.title}" + todo_path = todo.title + dir_path = f"{self.root_path}/todos/{todo_path}" + os.makedirs(dir_path) detail_path = f"{dir_path}/detail" + if todo.todo_path is None: + todo.todo_path = todo_path logger.info("create_todo %s",detail_path) async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: await f.write(json.dumps(todo.to_dict())) + self.known_todo[todo.todo_id] = todo return True async def update_todo(self,todo_id:str,new_stat:str)->bool: @@ -263,7 +279,7 @@ class WorkspaceEnvironment(Environment): todo.raw_obj = todo.to_dict() todo.raw_obj["status"] = new_stat - detail_path = todo.todo_path + "/detail" + detail_path = f"{self.root_path}/todos/{todo.todo_path}/detail" async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: await f.write(json.dumps(todo.raw_obj)) return True diff --git a/src/requirements.txt b/src/requirements.txt index b83e564..0c452e5 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -19,7 +19,7 @@ click>=8.1.7 colorama>=0.4.6 coloredlogs>=15.0.1 decorator>=4.4.2 -fastapi>=0.99.1 +fastapi filelock>=3.12.3 flatbuffers>=23.5.26 frozenlist>=1.4.0 @@ -59,7 +59,7 @@ proto-plus>=1.22.3 pulsar-client>=3.3.0 pyasn1>=0.5.0 pyasn1-modules>=0.3.0 -pydantic>=1.10.12 +pydantic PyPika>=0.48.9 pyreadline3>=3.4.1 python-dateutil>=2.8.2 @@ -137,4 +137,4 @@ python-telegram-bot pydub stability_sdk sentence-transformers==2.2.2 -tiktoken \ No newline at end of file +tiktoken