1) Fix bugs.

2) openai_node support async/await.

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.
 

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
 
         

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:
+        for sub_todo in self.sub_todos.values():
-            if sub_todo.state != "done":
+            if sub_todo.state:
-                return False
+                if sub_todo.state != "done":
+                    return False
         
         now = datetime.now().timestamp()
         time_diff = now - self.due_date

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)

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

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

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
+tiktoken