From 3d000956500ea3a2464ff7fd3261f35b3819a7f5 Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sun, 10 Dec 2023 21:42:23 -0800 Subject: [PATCH] Add implement of Agent Workspace (include a taskmanager system) --- doc/promps/Do TODO.md | 4 +- doc/promps/Review Task.md | 77 +++++- rootfs/agents/Jarvis/agent.toml | 14 +- rootfs/agents/JarvisPlus/agent.toml | 4 +- src/aios/agent/agent.py | 26 +- src/aios/agent/llm_process.py | 93 +++++-- src/aios/agent/workspace.py | 357 ++++++++++++++++++++++++++ src/aios/environment/workspace_env.py | 2 + src/aios/proto/agent_task.py | 291 +++++++++++++++++++-- src/component/tg_tunnel.py | 13 +- 10 files changed, 813 insertions(+), 68 deletions(-) create mode 100644 src/aios/agent/workspace.py diff --git a/doc/promps/Do TODO.md b/doc/promps/Do TODO.md index 7e13f04..111b826 100644 --- a/doc/promps/Do TODO.md +++ b/doc/promps/Do TODO.md @@ -1,3 +1,5 @@ # Do (TODO) 目标是结合 角色定义，手头的工具，已知知识 完成一个确定的任务。 - 完成任务时应使用ReAct的方法：应在给出执行动作前，先自言自语的输出一个计划，然后在动作（这个自言自语会变成TODO Logs） \ No newline at end of file + 完成任务时应使用ReAct的方法：应在给出执行动作前，先自言自语的输出一个计划，然后在动作（这个自言自语会变成TODO Logs） + + \ No newline at end of file diff --git a/doc/promps/Review Task.md b/doc/promps/Review Task.md index 18f1162..48c8526 100644 --- a/doc/promps/Review Task.md +++ b/doc/promps/Review Task.md @@ -1,8 +1,79 @@ -# Review (Task/Todo) +# Review Task/Todo 目的是结合已知信息（重点是已经进行操作的记录），对失败的，完成的不好的任务进行思考，尝试给出更好的解决方案 1. 管理学方法：更换负责人 2. 管理学方法：拆分 3. 给出建议（该建议可以在下次一次DO-Check）循环中被使用 -## Quick Review -有一些简单的Task是永远不会结束的（比如定时提醒）。此时通过Quick Review来调整这些Task的状态，让其在正确的时间进入Review和DO +## ReviewTasklist +目的是选择一个优先级最高的任务开始工作 +（这个流程现在是通过计算的方法，基于优先级排序后，FIFO的处理） + +## ReviewTask 对未开始的Task进行首次处理 +LLM 结果动作： +- 确认执行人，在非Workflow环境中，执行人就是Agent自己，所以不存在这个选项 +- 确认执行时间和过期时间，任务只有在执行时间以后和过期时间以前才有机会执行，无法确认执行时间的可以设置下一次检查时间 +- 对任务进行拆分（如何防止无限拆分是个大问题），或则有一些简单任务不允许拆分。 +- 判断可以立刻执行任务（将任务当成TODO工作），通过Action进入下一个LLMProcess +- 判断任务超出Agent能力范围，宣告失败 + +Example: +```markdown +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. +- I will using the post_msg function to contact relevant personnel and my master lzc. +- 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, 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. + +The result of my planned execution must be directly parsed by `python json.loads`. Here is an example: +{ + resp: '$what_did_I_do', + 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: 'cancel' # pending,cancel + }, + { + op: 'write_file', + path: '/todos/$todo_path/.result/$doc_name', + content:'$doc_content' + } + ] +} + +``` + +## PlanTask 对已经确认的Task进行执行 +根据任务的分类，进入不同的LLM Plan逻辑 + 简单任务：当作TODO立刻执行 + 普通任务：拆分TODO + + +## QuickCheckTask 对处于半确认状态Task进行Quick Review +有一些Task是永远不会结束的（比如定时提醒）。此时通过Quick Review来调整这些Task的状态，让其在正确的时间进入Review和DO + + +## RetryTask 对未成功的任务进行再次处理 diff --git a/rootfs/agents/Jarvis/agent.toml b/rootfs/agents/Jarvis/agent.toml index 377d58c..f9a70a2 100644 --- a/rootfs/agents/Jarvis/agent.toml +++ b/rootfs/agents/Jarvis/agent.toml @@ -14,13 +14,14 @@ role_desc = """ Your name is Jarvis, the super personal assistant to the master. """ -[LLMProcess.message] +[behavior.on_message] type="LLMAgentMessageProcess" process_description=""" 1. Based on your role, combined with existing information, make a brief and efficient reply. 2. Be mindful of the identity of the person you are chatting with and provide services accordingly based on their status. 3. Understand the intention of the dialogue, while using the necessary reply, use the appropriate, supported ACTION. -4. You are proficient in the languages of various countries and try to communicate with each other's mother tongue. +4. If you feel that there is a potential Task in the dialogue, you can create these tasks through appropriate ACTION. Be careful to query whether there are the same task before creating.Using the query interface is a high -cost behavior. +5. You are proficient in the languages of various countries and try to communicate with each other's mother tongue. """ reply_format = """ @@ -44,4 +45,13 @@ known_info_tips = """ tools_tips = """ """ +#[behavior.self_thinking] + +#[behavior.review_task] + +#[behavior.do] + +#[behavior.check] + + diff --git a/rootfs/agents/JarvisPlus/agent.toml b/rootfs/agents/JarvisPlus/agent.toml index d264f81..563bdcd 100644 --- a/rootfs/agents/JarvisPlus/agent.toml +++ b/rootfs/agents/JarvisPlus/agent.toml @@ -9,8 +9,8 @@ owner_env = ["knowledge"] [[work.do]] 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: +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. - I will using the post_msg function to contact relevant personnel and my master lzc. diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index 1a1a463..c19d7ea 100644 --- a/src/aios/agent/agent.py +++ b/src/aios/agent/agent.py @@ -135,16 +135,20 @@ class AIAgent(BaseAIAgent): self.owenr_bus = None self.enable_function_list = None - self.llm_process:Dict[str,BaseLLMProcess] = {} + self.memory : AgentMemory = None + self.prviate_workspace : AgentWorkspace = None + self.behaviors:Dict[str,BaseLLMProcess] = {} + async def initial(self,params:Dict = None): self.memory = AgentMemory(self.agent_id,self.chat_db) - + self.prviate_workspace = AgentWorkspace(self.agent_id) init_params = {} init_params["memory"] = self.memory - for process_name in self.llm_process.keys(): - init_result = await self.llm_process[process_name].initial(init_params) + init_params["workspace"] = self.prviate_workspace + for process_name in self.behaviors.keys(): + init_result = await self.behaviors[process_name].initial(init_params) if init_result is False: logger.error(f"llm process {process_name} initial failed! initial return False") return False @@ -222,16 +226,16 @@ class AIAgent(BaseAIAgent): self.history_len = int(config.get("history_len")) #load all LLMProcess - self.llm_process = {} - LLMProcess = config.get("LLMProcess") - for process_config_name in LLMProcess.keys(): - process_config = LLMProcess[process_config_name] + self.behaviors = {} + behaviors = config.get("behavior") + for process_config_name in behaviors.keys(): + process_config = behaviors[process_config_name] real_config = {} real_config.update(config) real_config.update(process_config) load_result = await LLMProcessLoader.get_instance().load_from_config(real_config) if load_result: - self.llm_process[process_config_name] = load_result + self.behaviors[process_config_name] = load_result else: logger.error(f"load LLMProcess {process_config_name} failed!") return False @@ -337,7 +341,7 @@ class AIAgent(BaseAIAgent): input_parms = { "msg":msg } - msg_process = self.llm_process.get("message") + msg_process = self.behaviors.get("on_message") llm_result : LLMResult = await msg_process.process(input_parms) if llm_result.state == LLMResultStates.ERROR: error_resp = msg.create_error_resp(llm_result.error_str) @@ -602,7 +606,7 @@ class AIAgent(BaseAIAgent): async def _llm_review_unassigned_todos(self,workspace:WorkspaceEnvironment): pass - async def _llm_read_report(self,report:AgentReport,worksapce:WorkspaceEnvironment): + async def _llm_read_report(self,report,worksapce:WorkspaceEnvironment): work_summary = worksapce.get_work_summary(self.agent_id) prompt : LLMPrompt = LLMPrompt() prompt.append(self.agent_prompt) diff --git a/src/aios/agent/llm_process.py b/src/aios/agent/llm_process.py index cc39a09..a87f6ce 100644 --- a/src/aios/agent/llm_process.py +++ b/src/aios/agent/llm_process.py @@ -15,6 +15,7 @@ from ..proto.ai_function import * from .agent_base import * from .agent_memory import * +from .workspace import * from ..frame.compute_kernel import * from ..environment.environment import * @@ -45,6 +46,19 @@ class BaseLLMProcess(ABC): self.envs : Dict[str,BaseEnvironment] = [] self.env : CompositeEnvironment = None + def aifunction_to_inner_function(self,all_inner_function:List[AIFunction]) -> List[Dict]: + result_func = [] + result_len = 0 + for inner_func in all_inner_function: + func_name = inner_func.get_name() + this_func = {} + this_func["name"] = func_name + this_func["description"] = inner_func.get_description() + this_func["parameters"] = inner_func.get_parameters() + result_len += len(json.dumps(this_func)) / 4 + result_func.append(this_func) + return result_func + @abstractmethod async def prepare_prompt(self,input:Dict) -> LLMPrompt: pass @@ -54,7 +68,7 @@ class BaseLLMProcess(ABC): pass @abstractmethod - async def exec_actions(self,actions:List[ActionItem],input:Dict,llm_result:LLMResult) -> bool: + async def post_llm_process(self,actions:List[ActionItem],input:Dict,llm_result:LLMResult) -> bool: pass @abstractmethod @@ -87,8 +101,9 @@ class BaseLLMProcess(ABC): def _format_content_by_env_value(self,content:str,env)->str: return content.format_map(env) - async def _execute_inner_func(self,inner_func_call_node,prompt: LLMPrompt,stack_limit = 5) -> ComputeTaskResult: + async def _execute_inner_func(self,inner_func_call_node,prompt: LLMPrompt,stack_limit = 1) -> ComputeTaskResult: arguments = None + stack_limit = stack_limit - 1 try: func_name = inner_func_call_node.get("name") arguments = json.loads(inner_func_call_node.get("arguments")) @@ -117,13 +132,18 @@ class BaseLLMProcess(ABC): task_result.result_code = ComputeTaskResultCode.ERROR task_result.error_str = f"prompt too long,can not predict" return task_result + + if stack_limit > 0: + inner_functions=prompt.inner_functions + else: + inner_functions = None 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, #NOTICE: inner_function in prompt can be a subset of get_inner_function + inner_functions=inner_functions, #NOTICE: inner_function in prompt can be a subset of get_inner_function timeout=self.timeout)) if task_result.result_code != ComputeTaskResultCode.OK: @@ -131,19 +151,15 @@ class BaseLLMProcess(ABC): 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 + + 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) + if inner_func_call_node: return await self._execute_inner_func(inner_func_call_node,prompt,stack_limit-1) @@ -194,7 +210,7 @@ class BaseLLMProcess(ABC): # use action to save history? if llm_result.action_list or len(llm_result.action_list) > 0: - await self.exec_actions(llm_result.action_list,input,llm_result) + await self.post_llm_process(llm_result.action_list,input,llm_result) return llm_result @@ -213,7 +229,7 @@ class LLMAgentMessageProcess(BaseLLMProcess): self.enable_inner_functions : Dict[str,bool] = None self.enable_actions : Dict[str,AIOperation] = None self.actions_desc : Dict[str,Dict] = None - self.workspace : WorkspaceEnvironment = None + self.workspace : AgentWorkspace = None self.memory : AgentMemory = None self.enable_kb = False @@ -236,7 +252,8 @@ class LLMAgentMessageProcess(BaseLLMProcess): if self.memory is None: logger.error(f"LLMAgeMessageProcess initial failed! memory not found") return False - + self.workspace = params.get("workspace") + self.init_actions() return True @@ -370,6 +387,8 @@ class LLMAgentMessageProcess(BaseLLMProcess): ### 修改todo/task的action ### workspace提供的额外的action system_prompt_dict["support_actions"] = await self.get_action_desc() + + #prompt.append_system_message(await self.get_action_desc()) ## Context （文本替换）,是否应该覆盖全部消息 @@ -403,6 +422,9 @@ class LLMAgentMessageProcess(BaseLLMProcess): #prompt.append_system_message(self.tools_tips) prompt.inner_functions.extend(self.get_inner_function_desc_from_env()) + if self.workspace: + prompt.inner_functions.extend(self.aifunction_to_inner_function(self.workspace.get_inner_function_desc())) + ## 给予查询KB的权限 if self.enable_kb: prompt.inner_functions.extend(self.get_inner_function_desc_from_kb()) @@ -415,9 +437,9 @@ class LLMAgentMessageProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: - return None + return self.workspace.inner_functions.get(func_name) - async def exec_actions(self,actions:List[ActionItem],input:Dict,llm_result:LLMResult) -> bool: + async def post_llm_process(self,actions:List[ActionItem],input:Dict,llm_result:LLMResult) -> bool: msg = input.get("msg") if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: resp_msg = msg.create_group_resp_msg(self.memory.agent_id,llm_result.resp) @@ -436,6 +458,7 @@ class LLMAgentMessageProcess(BaseLLMProcess): action_item.parms["resp_msg"] = resp_msg action_item.parms["llm_result"] = llm_result action_item.parms["start_at"] = datetime.now() + action_item.parms["creator"] = self.memory.agent_id action_item.parms["result"] = await op.execute(action_item.parms) action_item.parms["end_at"] = datetime.now() else: @@ -461,7 +484,25 @@ class ReviewTaskProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: pass - async def exec_actions(self,actions:List[ActionItem]) -> bool: + async def post_llm_process(self,actions:List[ActionItem]) -> bool: + pass + +class QuickReviewTaskProcess(BaseLLMProcess): + def __init__(self) -> None: + super().__init__() + + async def load_from_config(self, config: dict) -> Coroutine[Any, Any, bool]: + if await super().load_from_config(config) is False: + return False + + async def prepare_prompt(self) -> LLMPrompt: + prompt = LLMPrompt() + pass + + async def get_inner_function(self,func_name:str) -> AIFunction: + pass + + async def post_llm_process(self,actions:List[ActionItem]) -> bool: pass class DoTodoProcess(BaseLLMProcess): @@ -479,7 +520,7 @@ class DoTodoProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: pass - async def exec_actions(self,actions:List[ActionItem]) -> bool: + async def post_llm_process(self,actions:List[ActionItem]) -> bool: pass @@ -498,7 +539,7 @@ class CheckTodoProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: pass - async def exec_actions(self,actions:List[ActionItem]) -> bool: + async def post_llm_process(self,actions:List[ActionItem]) -> bool: pass class SelfLearningProcess(BaseLLMProcess): @@ -516,7 +557,7 @@ class SelfLearningProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: pass - async def exec_actions(self,actions:List[ActionItem]) -> bool: + async def post_llm_process(self,actions:List[ActionItem]) -> bool: pass class SelfThinkingProcess(BaseLLMProcess): @@ -534,7 +575,7 @@ class SelfThinkingProcess(BaseLLMProcess): async def get_inner_function(self,func_name:str) -> AIFunction: pass - async def exec_actions(self,actions:List[ActionItem]) -> bool: + async def post_llm_process(self,actions:List[ActionItem]) -> bool: pass class LLMProcessLoader: diff --git a/src/aios/agent/workspace.py b/src/aios/agent/workspace.py new file mode 100644 index 0000000..e97453d --- /dev/null +++ b/src/aios/agent/workspace.py @@ -0,0 +1,357 @@ +from ast import Dict +import json +import sqlite3 +import os +from typing import List + +import aiofiles + +from ..proto.ai_function import * +from ..proto.agent_task import * +from ..storage.storage import * + +logger = logging.getLogger(__name__) + +class LocalAgentTaskManger(AgentTaskManager): + def __init__(self, owner_id): + super().__init__() + self.root_path = f"{AIStorage.get_instance().get_myai_dir()}/tasklist/{owner_id}" + #self.root_path = os.path.join(workspace, list_type) + if not os.path.exists(self.root_path): + os.makedirs(self.root_path) + + self.db_path = os.path.join(self.root_path, "tasklist.db") + self.conn = None + try: + self.conn = sqlite3.connect(self.db_path) + except Exception as e: + logger.error("Error occurred while connecting to database: %s", e) + return None + + cursor = self.conn.cursor() + cursor.execute(''' + CREATE TABLE IF NOT EXISTS obj_list ( + id TEXT, + path TEXT + ) + ''') + self.conn.commit() + + def _get_obj_path(self,objid:str) -> str: + cursor = self.conn.cursor() + cursor.execute(''' + SELECT path FROM obj_list WHERE id = ? + ''',(objid,)) + row = cursor.fetchone() + if row: + return row[0] + else: + return None + + def _save_obj_path(self,objid:str,path:str): + cursor = self.conn.cursor() + cursor.execute(''' + INSERT INTO obj_list (id,path) VALUES (?,?) + ''',(objid,path)) + self.conn.commit() + + async def create_task(self,task:AgentTask,parent_id:str = None) -> str: + try: + #perfix = task.task_id[-5] + if parent_id: + parent_path = self._get_obj_path(parent_id) + task_path = f"{parent_path}/{task.title}" + else: + task_path = f"{task.title}" + + dir_path = f"{self.root_path}/{task_path}" + + os.makedirs(dir_path) + detail_path = f"{dir_path}/detail" + if task.task_path is None: + task.task_path = task_path + self._save_obj_path(task.task_id,task_path) + logger.info("create_task at %s",detail_path) + async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(task.to_dict())) + except Exception as e: + logger.error("create_task failed:%s",e) + return str(e) + + return None + + async def create_todos(self,owner_task_id:str,todos:List[AgentTodoTask]): + owner_task_path = self._get_obj_path(owner_task_id) + if owner_task_path is None: + return f"owner task {owner_task_id} not found" + + try: + step_order = 0 + for todo in todos: + todo.step_order = step_order + todo.owner_taskid = owner_task_id + todo_path = f"{self.root_path}/{owner_task_path}/#{step_order} {todo.title}.todo" + self._save_obj_path(todo.todo_id,todo_path) + async with aiofiles.open(todo_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(todo.to_dict())) + logger.info("create_todos at %s OK!",todo_path) + step_order += 1 + except Exception as e: + logger.error("create_todos failed:%s",e) + return str(e) + + return None + + + async def append_worklog(self,task:AgentTask,log:AgentWorkLog): + worklog = f"{self.root_path}/{task.task_path}/.worklog" + + async with aiofiles.open(worklog, mode='w+', encoding="utf-8") as f: + content = await f.read() + if len(content) > 0: + json_obj = json.loads(content) + else: + json_obj = {} + logs = json_obj.get("logs") + if logs is None: + logs = [] + logs.append(log.to_dict()) + json_obj["logs"] = logs + await f.write(json.dumps(json_obj)) + + + async def get_worklog(self,obj_id:str)->List[AgentWorkLog]: + obj_path = self._get_obj_path(obj_id) + if obj_path is None: + return [] + + if obj_path.endswith(".todo"): + dir_path = os.path.dirname(obj_path) + worklog_path = f"{self.root_path}/{dir_path}/.worklog" + else: + worklog_path = f"{self.root_path}/{obj_path}/.worklog" + + async with aiofiles.open(worklog_path, mode='r', encoding="utf-8") as f: + content = await f.read() + if len(content) > 0: + json_obj = json.loads(content) + else: + json_obj = {} + logs = json_obj.get("logs") + return logs + + + async def get_task(self,task_id:str) -> AgentTask: + task_path = self._get_obj_path(task_id) + if task_path is None: + logger.error("get_task:%s,not found!",task_id) + return None + + return await self.get_task_by_path(task_path) + + async def _get_task_by_fullpath(self,task_fullpath) -> AgentTask: + detail_path = f"{task_fullpath}/detail" + try: + with open(detail_path, mode='r', encoding="utf-8") as f: + task_dict = json.load(f) + result_task:AgentTask = AgentTask.from_dict(task_dict) + if result_task: + relative_path = os.path.relpath(task_fullpath, self.root_path) + result_task.task_path = relative_path + else: + logger.error("_get_task_by_fullpath:%s,parse failed!",detail_path) + + return result_task + except Exception as e: + logger.error("_get_task_by_fullpath:%s,failed:%s",task_fullpath,e) + return None + + async def get_task_by_path(self,task_path:str) -> AgentTask: + full_path = f"{self.root_path}/{task_path}" + return await self._get_task_by_fullpath(full_path) + + async def get_todo(self,todo_id:str) -> AgentTodoTask: + todo_path = self._get_obj_path(todo_id) + if todo_path is None: + logger.error("get_todo:%s,not found!",todo_id) + return None + + try: + with open(todo_path, mode='r', encoding="utf-8") as f: + todo_dict = json.load(f) + result_todo:AgentTodoTask = AgentTodoTask.from_dict(todo_dict) + if result_todo: + result_todo.todo_path = todo_path + else: + logger.error("get_todo:%s,parse failed!",todo_path) + + return result_todo + except Exception as e: + logger.error("get_todo:%s,failed:%s",todo_path,e) + + return None + + async def get_sub_tasks(self,task_id:str) -> List[AgentTask]: + task_path = self._get_obj_path(task_id) + if task_path is None: + return [] + + sub_tasks = [] + for sub_item in os.listdir(task_path): + if sub_item.startswith("."): + continue + if sub_item == "workspace": + continue + + full_path = os.path.join(task_path, sub_item) + if os.path.isdir(full_path): + sub_task = await self.get_task_by_path(f"{task_path}/{sub_item}") + if sub_task: + sub_tasks.append(sub_task) + pass + + + async def get_sub_todos(self,task_id:str) -> List[AgentTodoTask]: + task_path = self._get_obj_path(task_id) + if task_path is None: + return [] + + sub_todos = [] + for sub_item in os.listdir(task_path): + if sub_item.startswith("."): + continue + if sub_item == "workspace": + continue + + full_path = os.path.join(task_path, sub_item) + if os.path.isfile(full_path) and sub_item.endswith(".todo"): + sub_todo = await self.get_todo_by_path(f"{task_path}/{sub_item}") + if sub_todo: + sub_todos.append(sub_todo) + + return sub_todos + + + #async def get_task_depends(self,task_id:str) -> List[AgentTask]: + # pass + + + async def list_task(self,filter:dict) -> List[AgentTask]: + + directory_path = self.root_path + result_list:List[AgentTask] = [] + + for entry in os.scandir(directory_path): + if not entry.is_dir(): + continue + if entry.name.startswith("."): + continue + if entry.name == "workspace": + continue + task_item = await self.get_task_by_path(entry.path) + if task_item: + if not task_item.is_finish(): + result_list.append(task_item) + + return result_list + + + + + async def update_task(self,task:AgentTask): + detail_path = f"{self.root_path}/{task.task_path}/detail" + try: + async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(task.to_dict())) + except Exception as e: + logger.error("update_task failed:%s",e) + return str(e) + + return None + + async def update_todo(self,todo:AgentTodoTask): + todo_path = self._get_obj_path(todo.todo_id) + if todo_path is None: + return f"todo {todo.todo_id} not found" + + try: + async with aiofiles.open(todo_path, mode='w', encoding="utf-8") as f: + await f.write(json.dumps(todo.to_dict())) + except Exception as e: + logger.error("update_todo failed:%s",e) + return str(e) + + return None + + #async def update_task_state(self,task_id,state:str): + # pass + + #async def update_todo_state(self,task_id,state:str): + # pass + + #todo共享其所在task的文件夹 + + async def get_task_file(self,task_id:str,path:str)->str: + #return fileid + pass + + + async def set_task_file(self,task_id:str,path:str,fileid:str): + pass + + + async def list_task_file(self,task_id:str,path:str): + pass + + + async def remove_task_file(self,task_id:str,path:str): + pass + + + +class AgentWorkspace: + def __init__(self,owner_agent_id:str) -> None: + self.agent_id : str = owner_agent_id + self.task_mgr : AgentTaskManager = LocalAgentTaskManger(owner_agent_id) + self.actions : Dict[str,ActionItem] = {} + self.inner_functions : Dict[str,AIFunction] = {} + + self.init_actions() + self.init_inner_functions() + + + def init_actions(self): + async def create_task(params): + taskObj = AgentTask.create_by_dict(params) + parent_id = params.get("parent") + return await self.task_mgr.create_task(taskObj,parent_id) + + create_task_action = SimpleAIOperation( + "create_task", + "Create a task in the task system, the supported parameters are: title, detail (simple task can not be filled), tags,due_date", + create_task, + ) + + self.actions[create_task_action.get_name()] = create_task_action + + def get_actions(self) -> Dict: + return self.actions + + def init_inner_functions(self): + async def list_tasks(): + result = {} + fitler = {} + task_list = await self.task_mgr.list_task(fitler) + for task_item in task_list: + result[task_item.task_id] = task_item.title + + return json.dumps(result) + + self.inner_functions["list_tasks"] = SimpleAIFunction("list_tasks", + "list all tasks in json format like {{$task_id:$task_title}...}", + list_tasks) + + def get_inner_function_desc(self) -> List[AIFunction]: + func_list = [] + func_list.extend(self.inner_functions.values()) + return func_list \ No newline at end of file diff --git a/src/aios/environment/workspace_env.py b/src/aios/environment/workspace_env.py index 5d52c29..84f20a4 100644 --- a/src/aios/environment/workspace_env.py +++ b/src/aios/environment/workspace_env.py @@ -50,6 +50,7 @@ class TodoListEnvironment(SimpleEnvironment): todoObj = AgentTodo.from_dict(params["todo"]) parent_id = params.get("parent") return await self.create_todo(parent_id,todoObj) + self.add_ai_operation(SimpleAIOperation( op="create_todo", description="create todo", @@ -61,6 +62,7 @@ class TodoListEnvironment(SimpleEnvironment): todo_id = params["id"] new_stat = params["state"] return await self.update_todo(todo_id,new_stat) + self.add_ai_operation(SimpleAIOperation( op="update_todo", description="update todo", diff --git a/src/aios/proto/agent_task.py b/src/aios/proto/agent_task.py index 791e0d9..3a64b2b 100644 --- a/src/aios/proto/agent_task.py +++ b/src/aios/proto/agent_task.py @@ -1,9 +1,15 @@ +from abc import ABC, abstractmethod +from typing import List, Optional import datetime import time - +import uuid from anyio import Path +import logging +from enum import Enum +from datetime import datetime +logger = logging.getLogger(__name__) class AgentTodoResult: TODO_RESULT_CODE_OK = 0, @@ -68,6 +74,7 @@ class AgentTodo: self.retry_count = 0 self.raw_obj = None + @classmethod def from_dict(cls,json_obj:dict) -> 'AgentTodo': todo = AgentTodo() @@ -182,40 +189,294 @@ class AgentTodo: logger.info(f"todo {self.title} can do.") return True + +############################################################################################ +class AgentTaskState(Enum): + TASK_STATE_WAIT= "wait_assign" + TASK_STATE_ASSIGNED = "assigned" + TASK_STATE_CONFIRMED = "confirmed" + + TASK_STATE_CANCEL = "cancel" + TASK_STATE_EXPIRED = "expired" + + TASK_STATE_DOING = "doing" + TASK_STATE_WAITING_CHECK = "wait_check" + TASK_STATE_CHECKFAILED = "check_failed" + TASK_STATE_DONE = "done" + TASK_STATE_FAILED = "failed" + @staticmethod + def from_str(value): + return next((s for s in AgentTaskState.__members__.values() if s.value == value), None) + +class AgentTodoState(Enum): + TODO_STATE_WAITING = "waiting" + TODO_STATE_WORKING = "working" + TODO_STATE_WAIT_CHECK = "wait_check" + TODO_STATE_CHECK_FAILED = "check_failed" + TODO_STATE_DONE = "done" + TASK_STATE_FAILED = "failed" + + @staticmethod + def from_str(value): + return next((s for s in AgentTodoState.__members__.values() if s.value == value), None) + +class AgentTodoTask: + def __init__(self) -> None: + self.todo_id = "todo#" + uuid.uuid4().hex + self.todo_path : str = None + self.owner_taskid = None + self.name:str = None + self.detail:str = None + self.state = AgentTodoState.TODO_STATE_WAITING + self.category = None + self.step_order:int = 0 + + def to_dict(self) -> dict: + pass + + def from_dict(self,json_obj:dict) -> 'AgentTask': + pass 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.task_path : str = None # get parent todo,sub todo by path self.title = None self.detail = None - - self.create_time = time.time() - - self.state = "wait_assign" + self.state = AgentTaskState.TASK_STATE_WAIT + self.priority:int = 5 # 1-10 + self.tags:List[str] = [] self.worker = None self.createor = None + # if due_date is none ,means no due date self.due_date = time.time() + 3600 * 24 * 2 - self.depend_task_ids = [] - self.step_todos = {} + # 确定的执行时间（执行条件） + self.next_do_time = None + # 如果next check time设置，说明任务适合在该时间点可能具备执行调教，尝试检查并执行 + self.next_check_time = None + self.depend_task_ids = [] + #self.step_todo_ids = [] + + self.create_time = time.time() + self.done_time = None + + self.last_do_time = None 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 + def is_finish(self) -> bool: + if self.state == AgentTaskState.TASK_STATE_DONE: + return True + + if self.state == AgentTaskState.TASK_STATE_CANCEL: + return True + + if self.state == AgentTaskState.TASK_STATE_EXPIRED: + return True + + if self.state == AgentTaskState.TASK_STATE_FAILED: + return True + return False + def to_dict(self) -> dict: + result = {} + result["task_id"] = self.task_id + result["title"] = self.title + result["detail"] = self.detail + result["state"] = self.state.value + result["priority"] = self.priority + result["tags"] = self.tags + result["worker"] = self.worker + result["createor"] = self.createor + if self.due_date: + result["due_date"] = datetime.fromtimestamp(self.due_date).isoformat() + if self.next_do_time: + result["next_do_time"] = datetime.fromtimestamp(self.next_do_time).isoformat() + if self.next_check_time: + result["next_check_time"] = datetime.fromtimestamp(self.next_check_time).isoformat() + result["depend_task_ids"] = self.depend_task_ids + #result["step_todo_ids"] = self.step_todo_ids + result["create_time"] = datetime.fromtimestamp(self.create_time).isoformat() + if self.done_time: + result["done_time"] = datetime.fromtimestamp(self.done_time).isoformat() + if self.last_do_time: + result["last_do_time"] = datetime.fromtimestamp(self.last_do_time).isoformat() + if self.last_plan_time: + result["last_plan_time"] = datetime.fromtimestamp(self.last_plan_time).isoformat() + if self.last_check_time: + result["last_check_time"] = datetime.fromtimestamp(self.last_check_time).isoformat() + return result + @classmethod + def from_dict(cls,json_obj:dict) -> 'AgentTask': + result = AgentTask() + result.task_id = json_obj.get("task_id") + result.title = json_obj.get("title") + result.detail = json_obj.get("detail") + result.state = AgentTaskState.from_str(json_obj.get("state")) + result.priority = json_obj.get("priority") + result.tags = json_obj.get("tags") + result.worker = json_obj.get("worker") + result.createor = json_obj.get("createor") + due_date = json_obj.get("due_date") + if due_date: + result.due_date = datetime.fromisoformat(due_date).timestamp() + next_do_time = json_obj.get("next_do_time") + if next_do_time: + result.next_do_time = datetime.fromisoformat(next_do_time).timestamp() + next_check_time = json_obj.get("next_check_time") + if next_check_time: + result.next_check_time = datetime.fromisoformat(next_check_time).timestamp() + result.depend_task_ids = json_obj.get("depend_task_ids") + #result.step_todo_ids = json_obj.get("step_todo_ids") + create_time = json_obj.get("create_time") + if create_time: + result.create_time = datetime.fromisoformat(create_time).timestamp() + done_time = json_obj.get("done_time") + if done_time: + result.done_time = datetime.fromisoformat(done_time).timestamp() + last_do_time = json_obj.get("last_do_time") + if last_do_time: + result.last_do_time = datetime.fromisoformat(last_do_time).timestamp() + last_plan_time = json_obj.get("last_plan_time") + if last_plan_time: + result.last_plan_time = datetime.fromisoformat(last_plan_time).timestamp() + last_check_time = json_obj.get("last_check_time") + if last_check_time: + result.last_check_time = datetime.fromisoformat(last_check_time).timestamp() + + if result.task_id is None or result.title is None or result.create_time is None or result.create_time is None: + logger.error(f"invalid task {json_obj}") + return None + + return result + @classmethod + def create_by_dict(cls,json_obj:dict) -> 'AgentTask': + creator = json_obj.get("creator") + if creator is None: + logger.error(f"invalid create task, creator is None") + return None + + result = AgentTask() + + result.title = json_obj.get("title") + result.detail = json_obj.get("detail") + if result.detail is None: + result.detail = result.title + result.priority = json_obj.get("priority") + if result.priority is None: + result.priority = 5 + + result.tags = json_obj.get("tags") + result.worker = json_obj.get("worker") + result.createor = creator + due_date = json_obj.get("due_date") + if due_date: + result.due_date = datetime.fromisoformat(due_date).timestamp() + + return result class AgentWorkLog: def __init__(self) -> None: + self.logid = "worklog#" + uuid.uuid4().hex + self.owner_taskid:str = None + self.owner_todoid:str = None + self.type:str = "" # 默认为普通类型的log,特殊类型的Log一般伴随着重要的状态改变 + self.timestamp = time.time() + self.content:str = None + self.result:str = None + self.meta : dict = None + self.operator = None + + def to_dict(self) -> dict: + pass + +class AgentTaskManager(ABC): + def __init__(self) -> None: + pass + + @abstractmethod + async def create_task(self,task:AgentTask,parent_id:str = None) -> str: + pass + + @abstractmethod + async def create_todos(self,owner_task_id:str,todos:List[AgentTodoTask]): + # return todo_id + pass + + @abstractmethod + async def append_worklog(self,log:AgentWorkLog): + pass + + @abstractmethod + async def get_worklog(self,obj_id:str)->List[AgentWorkLog]: + pass + + @abstractmethod + async def get_task(self,task_id:str) -> AgentTask: + pass + + #@abstractmethod + #async def get_task_by_fullpath(self,task_path:str) -> AgentTask: + # pass + + @abstractmethod + async def get_todo(self,todo_id:str) -> AgentTodoTask: + pass + + @abstractmethod + async def get_sub_tasks(self,task_id:str) -> List[AgentTask]: + pass + + @abstractmethod + async def get_sub_todos(self,task_id:str) -> List[AgentTodoTask]: + pass + + #@abstractmethod + #async def get_task_depends(self,task_id:str) -> List[AgentTask]: + # pass + + @abstractmethod + async def list_task(self,filter:Optional[dict]) -> List[AgentTask]: + pass + + @abstractmethod + async def update_task(self,task:AgentTask): + pass + + @abstractmethod + async def update_todo(self,todo:AgentTodoTask): + pass + + #@abstractmethod + #async def update_task_state(self,task_id,state:str): + # pass + + #@abstractmethod + #async def update_todo_state(self,task_id,state:str): + # pass + + #subtask,todo共享其所在task的文件夹 + @abstractmethod + async def get_task_file(self,task_id:str,path:str)->str: + #return fileid + pass + + @abstractmethod + async def set_task_file(self,task_id:str,path:str,fileid:str): + pass + + @abstractmethod + async def list_task_file(self,task_id:str,path:str): + pass + + @abstractmethod + async def remove_task_file(self,task_id:str,path:str): pass -class AgentReport: - def __init__(self) -> None: - pass \ No newline at end of file + + diff --git a/src/component/tg_tunnel.py b/src/component/tg_tunnel.py index ec8cea4..fd29d8c 100644 --- a/src/component/tg_tunnel.py +++ b/src/component/tg_tunnel.py @@ -85,10 +85,11 @@ class TelegramTunnel(AgentTunnel): TelegramTunnel.all_bots[self.target_id] = self.bot async def _run_app(): + update_id = 0 try: - update_id = (await self.bot.get_updates())[0].update_id - except Exception as e: - update_id = None + update = await self.bot.get_updates() + if len(update) > 0: + update_id = update[0].update_id except Exception as e: logger.error(f"tg_tunnel error:{e}") logger.exception(e) @@ -97,11 +98,7 @@ class TelegramTunnel(AgentTunnel): #logger.info("listening for new messages...") while True: try: - 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 - + update_id = await self._do_process_raw_message(self.bot, update_id) except NetworkError: await asyncio.sleep(1) except Forbidden: