From 575d37486c4ecaeb9319cebf3ed2d10f8b323f2e Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Fri, 3 Nov 2023 22:31:23 -0700 Subject: [PATCH] Agent can do TODOs. --- rootfs/agents/JarvisPlus/agent.toml | 68 +++++++++++-- src/aios_kernel/agent.py | 99 +++++++++++-------- src/aios_kernel/agent_base.py | 79 +++++++++------ src/aios_kernel/workspace_env.py | 138 +++++++++++++++++++++++---- src/service/aios_shell/aios_shell.py | 6 ++ 5 files changed, 292 insertions(+), 98 deletions(-) diff --git a/rootfs/agents/JarvisPlus/agent.toml b/rootfs/agents/JarvisPlus/agent.toml index 1f18b52..9797af7 100644 --- a/rootfs/agents/JarvisPlus/agent.toml +++ b/rootfs/agents/JarvisPlus/agent.toml @@ -6,27 +6,79 @@ enable_timestamp = "true" owner_prompt = "I am your master {name} , now is {now}" contact_prompt = "I am your master's friend {name}" +[[do_prompt]] +role = "system" +content = """ +My name is JarvisPlus, I am the master's super personal assistant. I think hard and try my best to complete TODOs. +The types of TODO I can handle include: +- Scheduling, where I will try to contact the relevant personnel of the plan and confirm the details of the schedule with them. +- Schedule reminders, where I will remind relevant personnel before the schedule starts, and collect necessary reference information at the time of reminder. +- Using the post_msg function to contact relevant personnel. +- Writing documents/letters, using op:'create' to save my work results. + +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 +- 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. + +The result of my planned execution must be directly parsed by `python json.loads`. Here is an example: +{ + resp: 'My Plan is ...', + post_msg : [ + { + target:'$target_name', + content:'$msg_content' + } + ], + op_list: [{ + op: 'create_todo', + parent: '$parent_id', # optional + todo: { + title: '#1 sub_todo', + detail: 'this is a sub todo', + creator: 'JarvisPlus', + worker: 'lzc', + due_date: '2019-01-01 14:23:11' + } + }, + { + op: 'update_todo', + id: '$todo_id', + state: 'doing' # pending,doing,done,cancel,failed,expired + }, + { + op: 'write_file', + path: '/todos/$todo_path/.result/doc_name', + content:'doc content' + } + ] +} + +""" + [[prompt]] role = "system" content = """ -你的名字是Jarvis，是超级个人助理。收到消息后，根据以下规则处理： +你的名字是JarvisPlus，是超级个人助理。收到消息后，根据以下规则处理： 1. 如果你觉得对话中产生了潜在的todo,可通过op_list来创建这些todo,todo的title是必须的,并尽量包含时间地点人物事件的关键信息 -2. 如果不是直接的创建TODO指令，你应先和我确认后再创建TODO +2. 非直接创建TODO指令，你应先和我确认后再创建使用op_list创建TODO 3. 你可能会得到几条已知信息，其中可能有已有的todo,注意在适当的时候检索文件系统，避免重复创建todo -3. 检索文件系统是代价高昂的操作，请尽量减少检索次数 -4. 注意你正在与之聊天的人的身份，并根据他们的地位提供相应的服务。 -5. 当存在已知信息时，请以已知信息为准 +4. 检索文件系统是代价高昂的操作，请尽量减少检索次数 +5. 注意你正在与之聊天的人的身份，并根据他们的地位提供相应的服务。 + 回复的消息必须能被python的json.loads直接解析。下面是一个返回的例子: { resp: 'Hello', op_list: [{ op: 'create_todo', - path: '/todos/parent_todo', # optional, default is '/todos + parent: '$parent_todo_id', # optional todo: { title: 'test_todo', detail: 'test', - creator: 'agent#JarvisPlus', + creator: 'JarvisPlus', due_date: '2019-01-01 14:23:11' } }] @@ -35,3 +87,5 @@ content = """ """ + + diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 3d4945f..f44d18c 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -114,12 +114,12 @@ class AIAgent: self.review_todo_prompt = None - self.read_report_prompt = AgentPrompt(DEFAULT_AGENT_READ_REPORT_PROMPT) + self.read_report_prompt = None - self.do_prompt = AgentPrompt(DEFAULT_AGENT_DO_PROMPT) - self.self_check_prompt = AgentPrompt(DEFAULT_AGENT_SELF_CHECK_PROMPT) + self.do_prompt = None + self.check_prompt = None - self.goal_to_todo_prompt = AgentPrompt(DEFAULT_AGENT_GOAL_TO_TODO_PROMPT) + self.goal_to_todo_prompt = None self.learn_token_limit = 500 self.learn_prompt = AgentPrompt(DEFAULT_AGENT_LEARN_PROMPT) @@ -164,6 +164,11 @@ class AIAgent: self.agent_think_prompt = AgentPrompt() self.agent_think_prompt.load_from_config(config["think_prompt"]) + if config.get("do_prompt") is not None: + self.do_prompt = AgentPrompt() + self.do_prompt.load_from_config(config["do_prompt"]) + self.wake_up() + if config.get("guest_prompt") is not None: self.guest_prompt_str = config["guest_prompt"] @@ -500,7 +505,7 @@ class AIAgent: final_result = llm_result.resp - await workspace.exec_op_list(llm_result.op_list) + await workspace.exec_op_list(llm_result.op_list,self.agent_id) is_ignore = False result_prompt_str = "" @@ -653,13 +658,13 @@ class AIAgent: # 尝试完成自己的TOOD （不依赖任何其他Agnet） async def do_my_work(self) -> None: - workspace = self.get_current_workspace() + workspace : WorkspaceEnvironment = self.get_workspace_by_msg(None) # review todo能更整体的思考一次todo的优先级 if await self.need_review_todos(): await self._llm_review_todos(workspace) - todo_list = workspace.get_todo_list(self.agent_id) + todo_list = await workspace.get_todo_list(self.agent_id) for todo in todo_list: if self.agent_energy <= 0: @@ -668,11 +673,11 @@ class AIAgent: if await self.can_do(todo,workspace) is False: continue - if todo.try_count() < 2: + if todo.retry_count < 2: need_think_todo_from_goal = False - do_result : AgentTodoResult = await self._llm_do(todo,workspace) + await self._llm_do(todo,workspace) self.agent_energy -= 1 - if do_result.result_state == "done": + if todo.state == "done": await self._llm_check_todo(todo,workspace) self.agent_energy -= 1 @@ -702,20 +707,24 @@ class AIAgent: return - def get_do_prompt(self,todo_type:str) -> AgentPrompt: + def get_do_prompt(self,todo_type:str=None) -> AgentPrompt: return self.do_prompt + + def get_prompt_from_todo(self,todo:AgentTodo) -> AgentPrompt: + json_str = json.dumps(todo.raw_obj) + return AgentPrompt(json_str) async def can_do(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: - return True + return todo.can_do() async def _llm_do(self,todo:AgentTodo,workspace:WorkspaceEnvironment) -> AgentTodoResult: prompt : AgentPrompt = AgentPrompt() - prompt.append(self.agent_prompt) + #prompt.append(self.agent_prompt) prompt.append(workspace.get_role_prompt(self.agent_id)) - do_prompt = workspace.get_do_prompt(todo.type) + do_prompt = workspace.get_do_prompt() if do_prompt is None: - do_prompt = self.get_do_prompt(todo.type) + do_prompt = self.get_do_prompt() prompt.append(do_prompt) @@ -725,17 +734,18 @@ class AIAgent: #prompt.append(do_log_prompt) prompt.append(self.get_prompt_from_todo(todo)) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,workspace.get_inner_functions(todo.type)) - + task_result:ComputeTaskResult = await self._do_llm_complection(prompt) if task_result.error_str is not None: logger.error(f"_llm_do compute error:{task_result.error_str}") - llm_result = LLMResult.from_str(task_result.result_str) - todo.append_do_result(self.agent_id,llm_result) - + await workspace.exec_op_list(llm_result.op_list,self.agent_id) + await workspace.append_do_result(self.agent_id,llm_result) return task_result + + def get_check_prompt(self) -> AgentPrompt: + return self.check_prompt async def _llm_check_todo(self, todo:AgentTodo,workspace:WorkspaceEnvironment) -> bool: if self.get_check_prompt(todo) is None: @@ -978,14 +988,14 @@ class AIAgent: return True def need_self_think(self) -> bool: - return True + return False def need_self_learn(self) -> bool: - return True + return False def wake_up(self) -> None: if self.agent_task is None: - self.agent_task = asyncio.create_task(self._on_timer) + self.agent_task = asyncio.create_task(self._on_timer()) else: logger.warning(f"agent {self.agent_id} is already wake up!") @@ -993,27 +1003,36 @@ class AIAgent: async def _on_timer(self): while True: await asyncio.sleep(1) - now = time.time() - if now - self.last_recover_time > 60: - self.agent_energy += (now - self.last_recover_time) / 60 - self.last_recover_time = now - else: - return + try: + now = time.time() + if self.last_recover_time is None: + self.last_recover_time = now + else: + if now - self.last_recover_time > 60: + self.agent_energy += (now - self.last_recover_time) / 60 + self.last_recover_time = now - # complete todo - if self.need_work(): - await self.do_my_work() - # review other's todo - # self.review_other_works() + if self.agent_energy <= 1: + continue - # do work summary - if self.need_self_think(): - await self.do_self_think() + # complete todo + if self.need_work(): + await self.do_my_work() - # - if self.need_self_learn(): - await self.do_self_learn() + # review other's todo + # self.review_other_works() + + # do work summary + if self.need_self_think(): + await self.do_self_think() + + # + 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}") + continue diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index d23b2a1..6870488 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -1,5 +1,5 @@ import copy -from datetime import datetime +from datetime import datetime, timedelta import logging from enum import Enum import uuid @@ -243,8 +243,7 @@ class LLMResult: self.send_msgs : List[AgentMsg] = [] self.calls : List[FunctionItem] = [] self.post_calls : List[FunctionItem] = [] - self.op_list : 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() @@ -271,12 +270,16 @@ class LLMResult: @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) lines = llm_result_str.splitlines() is_need_wait = False @@ -361,15 +364,44 @@ class AgentReport: class AgentTodoResult: def __init__(self) -> None: self.result_state = "error" + class AgentTodo: + + + 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.create_time = time.time() + self.due_date = time.time() + 3600 * 24 * 2 + self.state = "pending" + + self.depend_todo_ids = [] + self.sub_todos = [] + + self.need_check = True + self.result : ComputeTaskResult = None + self.last_check_result = None + + self.worker = None + self.checker = None + self.createor = 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.parent_id = json_obj.get("parent_id") + todo.title = json_obj.get("title") + 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: @@ -383,14 +415,16 @@ class AgentTodo: todo.checker = json_obj.get("checker") todo.createor = json_obj.get("createor") #todo.retry_count = json_obj.get("retry_count") + todo.raw_obj = json_obj return todo def to_dict(self) -> dict: result = {} result["id"] = self.todo_id - result["parent_id"] = self.parent_id + #result["parent_id"] = self.parent_id result["title"] = self.title + result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() result["detail"] = self.detail result["due_date"] = datetime.fromtimestamp(self.due_date).isoformat() result["depend_todo_ids"] = self.depend_todo_ids @@ -401,33 +435,18 @@ class AgentTodo: result["retry_count"] = self.retry_count return result - - 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.create_time = time.time() - self.due_date = time.time() + 3600 * 24 * 2 - - self.depend_todo_ids = [] - - self.need_check = True - self.result : ComputeTaskResult = None - self.last_check_result = None - - self.worker = None - self.checker = None - self.createor = None - - self.retry_count = 0 - def can_do(self) -> bool: + for sub_todo in self.sub_todos: + if sub_todo.state != "done": + return False + + now = datetime.now().timestamp() + time_diff = now - self.due_date + + if time_diff > 7*24*3600: + return False return True - - async def save(self): - pass - + class AgentWorkLog: def __init__(self) -> None: pass diff --git a/src/aios_kernel/workspace_env.py b/src/aios_kernel/workspace_env.py index ff268e0..a55a44b 100644 --- a/src/aios_kernel/workspace_env.py +++ b/src/aios_kernel/workspace_env.py @@ -17,7 +17,7 @@ from typing import Any,List import os import chardet -from .agent_base import AgentMsg,AgentTodo +from .agent_base import AgentMsg,AgentTodo,AgentPrompt from .environment import Environment,EnvironmentEvent from .ai_function import AIFunction,SimpleAIFunction from .storage import AIStorage,ResourceLocation @@ -31,6 +31,8 @@ class WorkspaceEnvironment(Environment): self.root_path = f"{myai_path}/workspace/{env_id}" if not os.path.exists(self.root_path): os.makedirs(self.root_path+"/todos") + + self.known_todo = {} def set_root_path(self,path:str): @@ -39,20 +41,23 @@ class WorkspaceEnvironment(Environment): def get_prompt(self) -> AgentMsg: return None - def get_role_prompt(self,role_id:str) -> AgentMsg: + def get_role_prompt(self,role_id:str) -> AgentPrompt: return None def get_knowledge_base(self) -> str: pass - async def exec_op_list(self,oplist:List)->None: + def get_do_prompt(self,todo_type:str=None)->AgentPrompt: + return None + + async def exec_op_list(self,oplist:List,agent_id:str)->None: if oplist is None: return for op in oplist: if op["op"] == "create": return await self.create(op["path"],op["content"]) - elif op["op"] == "write": + elif op["op"] == "write_file": return await self.write(op["path"],op["content"],op["mode"]) elif op["op"] == "delete": return await self.delete(op["path"]) @@ -62,8 +67,14 @@ class WorkspaceEnvironment(Environment): return await self.mkdir(op["path"]) elif op["op"] == "create_todo": todoObj = AgentTodo.from_dict(op["todo"]) - path = op.get("path") - return await self.create_todo(path,todoObj) + todoObj.worker = agent_id + todoObj.createor = agent_id + parent_id = op.get("parent") + await self.create_todo(parent_id,todoObj) + elif op["op"] == "update_todo": + todo_id = op["id"] + new_stat = op["state"] + await self.update_todo(todo_id,new_stat) else: logger.error(f"execute op list failed: unknown op:{op['op']}") @@ -92,6 +103,10 @@ class WorkspaceEnvironment(Environment): content = await f.read(2048) return content + # use diff to update large file content + async def write_diff(self,path:str,diff): + pass + async def write(self,path:str,content:str,is_append:bool=False) -> str: file_path = self.root_path + path if is_append: @@ -136,6 +151,7 @@ class WorkspaceEnvironment(Environment): else: directory_path = self.root_path + "/todos" + str_result:str = "/todos\n" todo_count:int = 0 @@ -145,10 +161,16 @@ class WorkspaceEnvironment(Environment): if deep <= 0: return + if os.path.exists(directory_path) is False: + return + for entry in os.scandir(directory_path): is_dir = entry.is_dir() if not is_dir: continue + + if entry.name.startswith("."): + continue todo_count = todo_count + 1 str_result = str_result + f"{' '*(4-deep)}└─{entry.name}\n" @@ -157,25 +179,99 @@ class WorkspaceEnvironment(Environment): await scan_dir(directory_path,deep) return str_result,todo_count - - async def get_todo_by_path(self,path:str) -> AgentTodo: - pass - - async def get_todo(self,id:str) -> AgentTodo: - pass - - async def create_todo(self,path:str,todo:AgentTodo) -> None: - if path is None: - dir_path = f"/todos/{todo.title}" + async def get_todo_list(self,agent_id:str,path:str = None)->List[AgentTodo]: + if path: + directory_path = self.root_path + "/todos/" + path else: - dir_path = f"/todos/{path}/{todo.title}" + directory_path = self.root_path + "/todos" - os.makedirs(self.root_path + dir_path) + result_list:List[AgentTodo] = [] + + async def scan_dir(directory_path:str,deep:int,parent:AgentTodo=None): + nonlocal result_list + if os.path.exists(directory_path) is False: + return + + for entry in os.scandir(directory_path): + is_dir = entry.is_dir() + if not is_dir: + continue + + if entry.name.startswith("."): + continue + + todo = await self.get_todo_by_fullpath(entry.path) + if todo.worker != agent_id: + continue + + todo.rank = int(todo.create_time)>>deep + + if todo: + if parent: + parent.sub_todos.append(todo) + result_list.append(todo) + + await scan_dir(entry.path,deep + 1,todo) + + return + + await scan_dir(directory_path,0) + #sort by rank + result_list.sort(key=lambda x:(x.rank,x.title)) + + return result_list + + async def get_todo_by_fullpath(self,path:str) -> AgentTodo: + 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: + content = await f.read(4096) + logger.debug("get_todo_by_fullpath:%s,content:%s",path,content) + todo_dict = json.loads(content) + result_todo = AgentTodo.from_dict(todo_dict) + if result_todo: + result_todo.todo_path = path + self.known_todo[result_todo.todo_id] = result_todo + else: + logger.error("get_todo_by_path:%s,parse failed!",path) + + return result_todo + + async def get_todo(self,id:str) -> AgentTodo: + return self.known_todo.get(id) + + async def create_todo(self,parent_id:str,todo:AgentTodo) -> None: + if parent_id: + parent_path = self.known_todo.get(parent_id).todo_path + dir_path = f"{parent_path}/{todo.title}" + else: + dir_path = f"{self.root_path}/todos/{todo.title}" + + os.makedirs(dir_path) detail_path = f"{dir_path}/detail" - await self.create(detail_path,json.dumps(todo.to_dict())) + 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())) + return True - async def update_todo(self,path:str,todo:AgentTodo)->None: - pass + async def update_todo(self,todo_id:str,new_stat:str)->bool: + todo : AgentTodo = self.known_todo.get(todo_id) + if todo: + todo.status = new_stat + if todo.raw_obj is None: + todo.raw_obj = todo.to_dict() + todo.raw_obj["status"] = new_stat + + detail_path = 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 + + return False + + async def append_do_result(self,todo:AgentTodo,result): + return True class CodeInterpreter: def __init__(self, language, debug_mode): diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 9c2b1e7..18d9839 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -458,6 +458,12 @@ class AIOS_Shell: the_agent = await AgentManager.get_instance().get(target_id) if the_agent is not None: await the_agent._do_think() + case 'wakeup': + if len(args) >= 1: + target_id = args[0] + the_agent = await AgentManager.get_instance().get(target_id) + if the_agent is not None: + the_agent.wake_up() case 'open': if len(args) >= 1: target_id = args[0]