diff --git a/doc/agent & workflow.drawio b/doc/agent & workflow.drawio index 2d4e10f..227ba15 100644 --- a/doc/agent & workflow.drawio +++ b/doc/agent & workflow.drawio @@ -1,6 +1,6 @@ - + - + @@ -514,7 +514,7 @@ - + @@ -637,21 +637,243 @@ - + - + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/doc/function_list.md b/doc/function_list.md index 994d3a6..15f7466 100644 --- a/doc/function_list.md +++ b/doc/function_list.md @@ -94,6 +94,14 @@ _self = parameters.get("_workspace") ##### Send Msg (系统原生能力) +##### Workspace File System + +agent.workspace.write_file +agent.workspace.read_file +agent.workspace.delte_file +agent.workspace.append_file +agent.workspace.list_file + ## Knowledge Base Knowledge Base对大部分Agent来说，是一个获得私有信息,并让LLM处理结果更好的基础设施(RAG支持）。少部分Agent会使用相关API,结合Knowledge Base所服务的目标来整理Knowledge Base. diff --git a/doc/storage.md b/doc/storage.md new file mode 100644 index 0000000..1f7b43f --- /dev/null +++ b/doc/storage.md @@ -0,0 +1,16 @@ +# NDN Storage + +```python + +class NDNStorage: + async def get(self,data_name:str)->Dict: + #return object desc, include local cache path + + async def set_file(self,local_path)->str: + # return data_name + + async def read_content(self,data_name:str)->bytes: + # return bytes + # 这里不但会读取本地缓存，还会对内容进行验证 + +``` \ No newline at end of file diff --git a/rootfs/agents/Jarvis/agent.toml b/rootfs/agents/Jarvis/agent.toml index c65e7de..9b0706d 100644 --- a/rootfs/agents/Jarvis/agent.toml +++ b/rootfs/agents/Jarvis/agent.toml @@ -51,12 +51,12 @@ llm_context.functions.enable = ["agent.workspace.list_task"] [behavior.triage_tasks] ## 处理任务列表，任务列表里会包含所有未执行过，且未过期的任务 -## 对于简单的任务会一次性完成处理 +## 对于简单的任务会一次性完成处理(一系列简单的提醒任务） type="AgentTriageTaskList" process_description=""" You are expected to effectively TRIAGE the task list described in JSON format, in accordance with your role. Your GOAL is : 1. Adjust the priority of the task and set up a reasonable processing time.(update_task) -2. Immediately perform a simple (similar to reminding one category) task. Send a message using send_message, set the task to complete the use of update_task. +2. Immediately perform a simple (similar to reminding one category) task. Send a message using post_message, then set the task to complete.(use update_task). 3. Organize tasks to remove tasks beyond your ability. And merge the repeated tasks.(update_task + cancel_task) """ reply_format = """ @@ -77,58 +77,71 @@ llm_context.actions.enable = ["agent.workspace.update_task","agent.workspace.can [behavior.plan_task] ## 首次处理任务 type="AgentPlanTask" +# 是否要加入对任务到期时间的关注？ process_description=""" -你得到的输入来自你自己之前记录在TaskList系统里的一个Task。现在你并不需要完成该Task，而是结合已知信息对Task进行一次Review.Review的过程是你独立完成的，你在形成结论的过程中可以使用工具，但不能和其它人交流。 -1. 理性的思考如何一步一步的高效的，在潜在的截止时间前完成该Task。明确拒绝超出自己能力范围的Task。 -2. 尝试对Task进行确认操作。确认操作的关键在于任务有了明确的执行时间。 -3. 对于需要多个步骤才能完成的Task,对Task进行TODO Plan。尤其注意与相关人员确认的步骤 -4. 对于不需要拆分TODO，且可立刻执行的任务。直接执行该任务。 +The input is a task comes from a Tasklist. You need to perform a PLAN. PLAN process on TASK in combination with the known information. +1. Carefully think about whether the task is within your ability, and the task other than the scope of ability is directly rejected. (cancel_task). +2. Immediately perform a simple (similar to reminding one category) task. Send a message using post_message, then set the task to complete.(use update_task). +3. Plan for non-simple tasks, and generate a TODO list. Every TODO MUST be an independent work with a clear goal. (set_todos) +4. If the task has been dealt with, it means that the task is ultimately completed.You need to analyze the processing report of the entire task and make new plans. """ reply_format = """ The Response must be directly parsed by `python json.loads`. Here is an example: { - think:'$think step-by-step to be sure you have the right answer.', - determine : '$determine, summary what you will do', - plans:[ #Optional - {"todo":"$todo_name","detail":"$todo_detail,"category":"$todo_category"} - ... - ], - tags: ['tag1', 'tag2'], #Optional,If the task involves important things and people, you can mark by 1-3 tags. + think:'$think step-by-step to be sure you have the right plan.', + resp:'$determine, summary what you do' actions: [{ name: '$action_name', $param_name: '$parm' #Optional, fill in only if the action has parameters. }] } """ -# action_list: ['cancle','confirm', 'execute'] -llm_context.actions.enable = ["agent.workspace.cancel_task"] -llm_context.functions.enable = ["agent.workspace.list_task"] +llm_context.actions.enable = ["agent.workspace.create_task","agent.workspace.update_task","agent.workspace.set_todos","agent.workspace.cancel_task","post_message"] +#llm_context.functions.enable = ["agent.workspace.list_task"] context="Your master now in {location}, time: {now}, weather: {weather}." [behavior.review_task] -## 处理已经被LLM处理过的任务，包括处理首次出错的任务，处理被的任务 - - - -[behavior.do] # do TODO -type="AgentDo" +## 当task的所有todo/subtask都完成后（不敢成功或是失败），进行一次review +type="AgentReviewTask" process_description=""" -1. 你的任务是结合自己的角色定义，手头的工具，已知信息、完成一个确定的TODO。完成该TODO后你会得到$200的小费。 -2. 输入的TODO是来自你自己对一个Task的Plan结果。 -3. 完成TODO的过程中你应该先思考再执行。执行的过程中可以使用工具，访问前置步骤的结果。执行的结果通常是按顺序执行的ActionList。 -4. 你必须独立的，一次性完成该TODO，你无法得到来自任何他人的协助。 -5. 对确认超出任务范围的TODO，你可以取消该TODO。对执行任务条件不满足的TODO，你可以标记为失败，但要说明失败原因 -7. TODO的完成结果如有需要应保存成数字文档 +The input you get is a task has been executed. You need to combine the execution results of the Task's TOOLIST or SUBTASK to review the TASK. +1. Determine whether TASK has reached the goal.If the goal has been reached, the task is marked with DONE .If the goal is not achieved, simply illustrate the reason and mark TASK as a failure.(update_task) +2. If task that have not been completed for a long time, the task is marked as the final failure after analyzing the reasons.(cancel_task) """ - reply_format = """ The Response must be directly parsed by `python json.loads`. Here is an example: { - think:'我的思考.' - tags: ['tag1', 'tag2'], #Optional,If the TODO involves important things and people, you can mark by 1-3 tags. + think:'$think step-by-step to be sure you have the right result.', + resp : '$determine, summary what you will do', + actions: [{ + name: '$action_name', + $param_name: '$parm' #Optional, fill in only if the action has parameters. + }] +} +""" +llm_context.actions.enable = ["agent.workspace.cancel_task","agent.workspace.update_task"] + +context="Your master now in {location}, time: {now}, weather: {weather}." +[behavior.do] +# do TODO +type="AgentDo" +process_description=""" +The input is a TODO comes from a Task. +1. Your task is to combine your role definition, tools on hand, known information, and complete a certain Todo.After completing the Todo, you will get a tip of $ 200. +2. 8000 word limit for short term memory. Your short term memory is short, so immediately save important information to files. +3. In the process of completing Todo, you should think first and then execute. During the execution, you can use functions to access the results of the front steps. +4. You must be independent and complete the TODO at once, and you cannot get the assistance from any others. +5. The execution result of TODO should be saved into digital documents if necessary +""" + +reply_format = """ +The Response must be directly parsed by `python json.loads`. Here is an example: +{ + think:'$think step by step, how to complete the todo', + resp: '$simport report about what you do', actions: [{ name: '$action1_name', $param_name: '$parm' #Optional, fill in only if the action has parameters. @@ -136,18 +149,46 @@ The Response must be directly parsed by `python json.loads`. Here is an example: ] } """ + +# 对于DO操作来说，让Agent查询自己的能力集合是否更合适？ +llm_context.actions.enable = ["agent.workspace.update_todo","post_message","agent.workspace.write_file","agent.workspace.append_file"] +llm_context.functions.enable = ["agent.workspace.read_file","agent.workspace.list_dir","system.shell.exec","system.shell.run_code","aigc.text_2_image","aigc.text_2_voice","web.search.duckduckgo"] + [behavior.check] +# check TODO result type="AgentCheck" +process_description=""" +1. The input is a TODO that has been successfully executed, which is created by you to complete a task.Your goal is to check whether the Todo has been completed in combination with relevant information. +2. In the process of checking the Todo, the focus is on the analysis of whether the Todo has gained the established goal in combination with TASK. +3. 使用update_todo来更新todo的最终 +""" + +reply_format = """ +The Response must be directly parsed by `python json.loads`. Here is an example: +{ + resp:'$think step by step, how to check the todo', + name: '$action1_name', + $param_name: '$parm' #Optional, fill in only if the action has parameters. + }, ... + ] +} +""" + +llm_context.actions.enable = ["agent.workspace.update_todo"] +llm_context.functions.enable = ["agent.workspace.read_file","agent.workspace.list_dir","system.shell.exec","system.shell.run_code","aigc.image_2_text","aigc.voice_2_text","web.search.duckduckgo"] + [behavior.self_thinking] -# self thing的主要目的是对各种chatlog,worklog进行分析，并更新终结。 +# self thing的主要目的是对各种chatlog,worklog进行分析，并更新面向人和事的summary。 type="AgentSelfThinking" -#[behavior.self_learning] -# self_learning的主要目的是根据自己的经验，主动的学习新的知识。这通常是一个专门整理知识库的Agent，一般的Agent谨慎开启 -#type="AgentSelfLearning" - #[behavior.self_improve] # self_improve 是最后的行为，允许Agent结合自己的工作经验，改进自己的提示词（注意保留历史版本） #type="AgentSelfImprove" +#[behavior.self_learning] +# self_learning的主要目的是根据自己的经验，主动的学习新的知识。这通常是一个专门整理知识库的Agent。由于Self Learn可能会消耗大量的Token，我们建议Agent通过共享的知识库更新来获得效果，谨慎开启 +#type="AgentSelfLearning" + + + diff --git a/src/aios/agent/agent_memory.py b/src/aios/agent/agent_memory.py index c852057..0194b23 100644 --- a/src/aios/agent/agent_memory.py +++ b/src/aios/agent/agent_memory.py @@ -1,10 +1,14 @@ # pylint:disable=E0402 from datetime import datetime,timedelta -from typing import Dict +import json +import threading +from typing import Dict, List +import sqlite3 from ..frame.compute_kernel import ComputeKernel from ..proto.ai_function import SimpleAIAction from ..proto.agent_msg import AgentMsg, AgentMsgType +from ..proto.agent_task import AgentWorkLog from .llm_context import GlobaToolsLibrary from .chatsession import AIChatSession @@ -16,28 +20,39 @@ logger = logging.getLogger(__name__) class AgentMemory: def __init__(self,agent_id:str,db_path:str) -> None: self.agent_id:str= agent_id - self.chat_db:str = db_path + self.memory_db:str = db_path self.model_name:str = "gp4-1106-preview" self.threshold_hours = 72 - @classmethod - def register_actions(cls): - async def action_chatlog_append(parms:Dict): - memory = parms.get("_memory") - if memory: - return await memory.action_chatlog_append(parms) - - chatlog_append_action = SimpleAIAction("chatlog_append","Append request & reply message to chatlog. No params",action_chatlog_append) - GlobaToolsLibrary.get_instance().register_tool_function(chatlog_append_action,"agent.memory.chatlog.append") - + def _get_conn(self): + """ get db connection """ + local = threading.local() + if not hasattr(local, 'conn'): + local.conn = self._create_connection(self.memory_db) + return local.conn + + def _create_connection(self, db_file): + """ create a database connection to a SQLite database """ + conn = None + try: + conn = sqlite3.connect(db_file) + except Error as e: + logging.error("Error occurred while connecting to database: %s", e) + return None + + if conn: + self._create_table(conn) + + return conn + def get_session_from_msg(self,msg:AgentMsg) -> AIChatSession: if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: session_topic = msg.target + "#" + msg.topic - chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) + chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.memory_db) else: session_topic = msg.get_sender() + "#" + msg.topic - chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) + chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.memory_db) return chatsession async def load_chatlogs(self,msg:AgentMsg,n:int=6,m:int=64,token_limit=800)->str: @@ -84,19 +99,83 @@ class AgentMemory: return histroy_str - async def action_chatlog_append(self,params:Dict) -> str: - # 使用params可以得到: LLM Process的输入，LLM Result,基于LLM Result构造的参数，当前actionItem - input_msg:AgentMsg = params.get("input").get("msg") - llm_result = params.get("llm_result") - chatsession = self.get_session_from_msg(input_msg) - resp_msg = params.get("resp_msg") - if resp_msg: - tags = llm_result.raw_result.get("tags") - chatsession.append(input_msg,tags) - chatsession.append(resp_msg,tags) + # async def action_chatlog_append(self,params:Dict) -> str: + # + # input_msg:AgentMsg = params.get("input").get("msg") + # llm_result = params.get("llm_result") + # chatsession = self.get_session_from_msg(input_msg) + # resp_msg = params.get("resp_msg") + # if resp_msg: + # tags = llm_result.raw_result.get("tags") + # chatsession.append(input_msg,tags) + # chatsession.append(resp_msg,tags) - return "OK" + # return "OK" + + async def load_worklogs(self,operator_id:str,owner_id:str=None, work_types:List[str]=None,token_limit=800): + conn = self._get_conn() + c = conn.cursor() + + query = 'SELECT * FROM worklog WHERE 1=1' + params = [] + + if operator_id is not None: + query += ' AND operator=?' + params.append(operator_id) + + if owner_id is not None: + query += ' AND owner_id=?' + params.append(owner_id) + + if work_types: + query += ' AND work_type IN ({})'.format(', '.join('?'*len(work_types))) + params.extend(work_types) + + query += ' ORDER BY timestamp DESC LIMIT 8' + + c.execute(query, tuple(params)) + rows = c.fetchall() + conn.close() + + return [self.from_db_row(row) for row in rows] + + def _create_table(self,conn): + c = conn.cursor() + c.execute(''' + CREATE TABLE IF NOT EXISTS worklog ( + logid TEXT PRIMARY KEY, + owner_id TEXT, + work_type TEXT, + timestamp REAL, + content TEXT, + result TEXT, + meta TEXT, + operator TEXT + ) + ''') + conn.commit() + conn.close() + + @classmethod + def from_db_row(self,row): + log = AgentWorkLog() + # 这里高度依赖表结构的顺序 + log.logid, log.owner_id, log.work_type, log.timestamp, log.content, log.result, meta_str, log.operator = row + log.meta = json.loads(meta_str) if meta_str else None + return log + async def append_worklog(self,log:AgentWorkLog)->str: + conn = self._get_conn() + c = conn.cursor() + # 将meta字典转换为JSON字符串 + meta_str = json.dumps(log.meta,ensure_ascii=False) if log.meta else None + c.execute(''' + INSERT INTO worklog (logid, owner_id, work_type, timestamp, content, result, meta, operator) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ''', (log.logid, log.owner_id, log.work_type, log.timestamp, log.content, log.result, meta_str, log.operator)) + conn.commit() + conn.close() + async def get_contact_summary(self,contact_id:str) -> str: if contact_id is None: return None @@ -114,8 +193,6 @@ class AgentMemory: async def update_sth_summary(self,sth_id:str,summary:str) -> str: return None - async def get_log_summary(self,msg:AgentMsg) -> str: - return None diff --git a/src/aios/agent/llm_do_task.py b/src/aios/agent/llm_do_task.py index 7f7bd59..17630b1 100644 --- a/src/aios/agent/llm_do_task.py +++ b/src/aios/agent/llm_do_task.py @@ -1,7 +1,7 @@ from ..proto.compute_task import LLMPrompt,LLMResult,ComputeTaskResult,ComputeTaskResultCode from ..proto.ai_function import AIFunction,AIAction,ActionNode from ..proto.agent_msg import AgentMsg,AgentMsgType -from ..proto.agent_task import AgentTask +from ..proto.agent_task import AgentTask, AgentTodo, AgentWorkLog from ..frame.compute_kernel import ComputeKernel from .agent_memory import AgentMemory @@ -20,6 +20,7 @@ import logging logger = logging.getLogger(__name__) +#LLM Process All the unfinished tasks,will sort the priority of the task after LLM, determine the next execution time, and complete the simple task class AgentTriageTaskList(LLMAgentBaseProcess): def __init__(self) -> None: super().__init__() @@ -45,6 +46,27 @@ class AgentTriageTaskList(LLMAgentBaseProcess): prompt.append_user_message(json.dumps(task_dict_list,ensure_ascii=False)) system_prompt_dict = self.prepare_role_system_prompt(context_info) + # May all logs is good for Agent Triage Task List? + have_known_info = False + known_info = {} + working_logs = await self.memory.load_worklogs(self.memory.agent_id) + if len(working_logs) > 0: + have_known_info = True + all_worklog_node = [] + for worklog in working_logs: + workNode = {} + dt = datetime.fromtimestamp(float(worklog.timestamp)) + workNode["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S") + workNode["type"] = worklog.work_type + workNode["content"] = worklog.content + workNode["result"] = worklog.result + all_worklog_node.append(workNode) + + known_info["worklogs"] = all_worklog_node + + if have_known_info: + system_prompt_dict["known_info"] = known_info + prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) return prompt @@ -58,127 +80,298 @@ class AgentTriageTaskList(LLMAgentBaseProcess): action_params["_llm_result"] = llm_result action_params["_agentid"] = self.memory.agent_id action_params["_start_at"] = datetime.now() - await self._execute_actions(actions,action_params) - + result_str = "OK" + try: + if await self._execute_actions(actions,action_params) is False: + result_str = "execute action failed!" + except Exception as e: + logger.error(f"execute action failed! {e}") + result_str = "execute action failed!,error:" + str(e) + + worklog = AgentWorkLog.create_by_content(self.memory.agent_id,"triage",llm_result.resp,self.memory.agent_id) + worklog.result = result_str + await self.memory.append_worklog(worklog) +# LLM a Task that never been LLMed, the result of LLM Process may be adjusted, splitting subtask or do simple task as a todo directly. class AgentPlanTask(LLMAgentBaseProcess): def __init__(self) -> None: super().__init__() - self.role_description:str = None - self.process_description:str = None - self.reply_format = None - - # 虽然在架构上LLM Process可以很容易的去Call另一个Process，但实际应用中还是应该慎重的保持LLM Process的简单性 - #self.do_task_llm_process : BaseLLMProcess = None - - async def initial(self,params:Dict = None) -> bool: - self.memory = params.get("memory") - if self.memory is None: - logger.error(f"LLMAgeMessageProcess initial failed! memory not found") - return False - self.workspace = params.get("workspace") - - - return True - - async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: - - + async def load_from_config(self, config: dict,is_load_default=True) -> bool: if await super().load_from_config(config) is False: return False - - self.role_description = config.get("role_desc") - if self.role_description is None: - logger.error(f"role_description not found in config") - return False - - if config.get("process_description"): - self.process_description = config.get("process_description") - - if config.get("reply_format"): - self.reply_format = config.get("reply_format") - - if config.get("context"): - self.context = config.get("context") - - self.llm_context = SimpleLLMContext() - if config.get("llm_context"): - self.llm_context.load_from_config(config.get("llm_context")) async def prepare_prompt(self,input:Dict) -> LLMPrompt: - agent_task = input.get("task") prompt = LLMPrompt() - system_prompt_dict = {} - system_prompt_dict["role_description"] = self.role_description - system_prompt_dict["process_rule"] = self.process_description - system_prompt_dict["reply_format"] = self.reply_format - prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) + + agent_task : AgentTask= input.get("task") + context_info = input.get("context_info") + if agent_task is None: + logger.error(f"task not found in input") + return None + prompt.append_user_message(json.dumps(agent_task.to_dict(),ensure_ascii=False)) + + system_prompt_dict = self.prepare_role_system_prompt(context_info) + + have_known_info = False + known_info = {} + working_logs = await self.memory.load_worklogs(None,agent_task.task_id) + if len(working_logs) > 0: + have_known_info = True + all_worklog_node = [] + for worklog in working_logs: + workNode = {} + dt = datetime.fromtimestamp(float(worklog.timestamp)) + workNode["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S") + workNode["type"] = worklog.work_type + workNode["operator"] = worklog.operator + workNode["content"] = worklog.content + workNode["result"] = worklog.result + all_worklog_node.append(workNode) + + known_info["worklogs"] = all_worklog_node + + if have_known_info: + system_prompt_dict["known_info"] = known_info + + prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) + prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) return prompt - async def get_review_task_actions(self) -> Dict[str,Dict]: - pass + async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: + action_params = {} + action_params["_input"] = input + agent_task : AgentTask= input.get("task") + action_params["_memory"] = self.memory + action_params["_workspace"] = self.workspace + action_params["_llm_result"] = llm_result + action_params["_agentid"] = self.memory.agent_id + action_params["_start_at"] = datetime.now() - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - pass + result_str = "OK" + try: + if await self._execute_actions(actions,action_params) is False: + result_str = "execute action failed!" + except Exception as e: + logger.error(f"execute action failed! {e}") + result_str = "execute action failed!,error:" + str(e) + + worklog = AgentWorkLog.create_by_content(agent_task.task_id,"plan",llm_result.resp,self.memory.agent_id) + worklog.result = result_str + await self.memory.append_worklog(worklog) - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass -class AgentReviewTask(BaseLLMProcess): +# Agent DO Todo +# The purpose is to complete Todo.It is the core LLM process. Can use sufficient external tools to do your best according to the identity and ability of AGENT.It is also the LLM Process of the main extension of Agent extension +class AgentDo(LLMAgentBaseProcess): def __init__(self) -> None: super().__init__() - async def load_from_config(self, config: dict): + async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: if await super().load_from_config(config) is False: return False - async def prepare_prompt(self) -> LLMPrompt: + async def prepare_prompt(self,input:Dict) -> LLMPrompt: prompt = LLMPrompt() - pass - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - pass + agent_todo : AgentTodo= input.get("todo") + context_info = input.get("context_info") + if agent_todo is None: + logger.error(f"task not found in input") + return None - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass + prompt.append_user_message(json.dumps(agent_todo.to_dict(),ensure_ascii=False)) + system_prompt_dict = self.prepare_role_system_prompt(context_info) + # May all logs is good for Agent Triage Task List? + have_known_info = False + known_info = {} + working_logs = await self.memory.load_worklogs(None,agent_todo.todo_id) + if len(working_logs) > 0: + have_known_info = True + all_worklog_node = [] + for worklog in working_logs: + workNode = {} + dt = datetime.fromtimestamp(float(worklog.timestamp)) + workNode["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S") + workNode["type"] = worklog.work_type + workNode["content"] = worklog.content + workNode["result"] = worklog.result + all_worklog_node.append(workNode) + + known_info["worklogs"] = all_worklog_node -class AgentCheck(BaseLLMProcess): + if have_known_info: + system_prompt_dict["known_info"] = known_info + + prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) + prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) + return prompt + + + async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: + action_params = {} + action_params["_input"] = input + agent_todo : AgentTodo= input.get("todo") + action_params["_memory"] = self.memory + action_params["_workspace"] = self.workspace + action_params["_llm_result"] = llm_result + action_params["_agentid"] = self.memory.agent_id + action_params["_start_at"] = datetime.now() + + result_str = "OK" + try: + if await self._execute_actions(actions,action_params) is False: + result_str = "execute action failed!" + except Exception as e: + logger.error(f"execute action failed! {e}") + result_str = "execute action failed!,error:" + str(e) + + worklog = AgentWorkLog.create_by_content(agent_todo.todo_id,"do",llm_result.resp,self.memory.agent_id) + worklog.result = result_str + await self.memory.append_worklog(worklog) + +#Agent check todo +# LLM a already-DO TODO, the purpose is to check whether it is completed to face the illusion of LLM.Check can use some tools, which is also the core of the agent extension。 +class AgentCheck(LLMAgentBaseProcess): def __init__(self) -> None: super().__init__() - async def load_from_config(self, config: dict) -> Coroutine[Any, Any, bool]: + async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: if await super().load_from_config(config) is False: return False - async def prepare_prompt(self) -> LLMPrompt: + async def prepare_prompt(self,input:Dict) -> LLMPrompt: prompt = LLMPrompt() - pass - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - pass + agent_todo : AgentTodo= input.get("todo") + context_info = input.get("context_info") + if agent_todo is None: + logger.error(f"task not found in input") + return None - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass + prompt.append_user_message(json.dumps(agent_todo.to_dict(),ensure_ascii=False)) -class AgentDo(BaseLLMProcess): + system_prompt_dict = self.prepare_role_system_prompt(context_info) + # May all logs is good for Agent Triage Task List? + have_known_info = False + known_info = {} + working_logs = await self.memory.load_worklogs(None,agent_todo.todo_id) + if len(working_logs) > 0: + have_known_info = True + all_worklog_node = [] + for worklog in working_logs: + workNode = {} + dt = datetime.fromtimestamp(float(worklog.timestamp)) + workNode["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S") + workNode["type"] = worklog.work_type + workNode["content"] = worklog.content + workNode["result"] = worklog.result + all_worklog_node.append(workNode) + + known_info["worklogs"] = all_worklog_node + + if have_known_info: + system_prompt_dict["known_info"] = known_info + + prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) + prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) + return prompt + + + async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: + action_params = {} + action_params["_input"] = input + agent_todo : AgentTodo= input.get("todo") + action_params["_memory"] = self.memory + action_params["_workspace"] = self.workspace + action_params["_llm_result"] = llm_result + action_params["_agentid"] = self.memory.agent_id + action_params["_start_at"] = datetime.now() + + result_str = "OK" + try: + if await self._execute_actions(actions,action_params) is False: + result_str = "execute action failed!" + except Exception as e: + logger.error(f"execute action failed! {e}") + result_str = "execute action failed!,error:" + str(e) + + worklog = AgentWorkLog.create_by_content(agent_todo.todo_id,"check",llm_result.resp,self.memory.agent_id) + worklog.result = result_str + await self.memory.append_worklog(worklog) + +#Agent review task +#When Task's Todolist is completed, or Task's subtask is completed, LLM review a TASK to determine that the Task has been completed.This Review also failed to execute. +class AgentReviewTask(LLMAgentBaseProcess): def __init__(self) -> None: super().__init__() - async def load_from_config(self, config: dict): + + async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: if await super().load_from_config(config) is False: return False - async def prepare_prompt(self) -> LLMPrompt: + async def prepare_prompt(self,input:Dict) -> LLMPrompt: prompt = LLMPrompt() - pass - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - pass + agent_task : AgentTask= input.get("task") + context_info = input.get("context_info") + if agent_task is None: + logger.error(f"task not found in input") + return None - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass + prompt.append_user_message(json.dumps(agent_task.to_dict(),ensure_ascii=False)) + + system_prompt_dict = self.prepare_role_system_prompt(context_info) + # May all logs is good for Agent Triage Task List? + have_known_info = False + known_info = {} + working_logs = await self.memory.load_worklogs(None,agent_task.task_id) + if len(working_logs) > 0: + have_known_info = True + all_worklog_node = [] + for worklog in working_logs: + workNode = {} + dt = datetime.fromtimestamp(float(worklog.timestamp)) + workNode["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S") + workNode["type"] = worklog.work_type + workNode["operator"] = worklog.operator + workNode["content"] = worklog.content + workNode["result"] = worklog.result + all_worklog_node.append(workNode) + + known_info["worklogs"] = all_worklog_node + + if have_known_info: + system_prompt_dict["known_info"] = known_info + + prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) + prompt.append_system_message(json.dumps(system_prompt_dict,ensure_ascii=False)) + return prompt + + + async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: + action_params = {} + action_params["_input"] = input + agent_task : AgentTask= input.get("task") + action_params["_memory"] = self.memory + action_params["_workspace"] = self.workspace + action_params["_llm_result"] = llm_result + action_params["_agentid"] = self.memory.agent_id + action_params["_start_at"] = datetime.now() + + result_str = "OK" + try: + if await self._execute_actions(actions,action_params) is False: + result_str = "execute action failed!" + except Exception as e: + logger.error(f"execute action failed! {e}") + result_str = "execute action failed!,error:" + str(e) + + worklog = AgentWorkLog.create_by_content(agent_task.task_id,"review",llm_result.resp,self.memory.agent_id) + worklog.result = result_str + await self.memory.append_worklog(worklog) \ No newline at end of file diff --git a/src/aios/agent/llm_process.py b/src/aios/agent/llm_process.py index d372613..9e75461 100644 --- a/src/aios/agent/llm_process.py +++ b/src/aios/agent/llm_process.py @@ -160,8 +160,8 @@ class BaseLLMProcess(ABC): # Action define in prompt, will be execute after llm compute prompt = await self.prepare_prompt(input) max_result_token = self.max_token - ComputeKernel.llm_num_tokens(prompt,self.model_name) - if max_result_token < MIN_PREDICT_TOKEN_LEN: - return LLMResult.from_error_str(f"prompt too long,can not predict") + #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, diff --git a/src/aios/agent/workspace.py b/src/aios/agent/workspace.py index c28ede8..a746f83 100644 --- a/src/aios/agent/workspace.py +++ b/src/aios/agent/workspace.py @@ -3,13 +3,14 @@ from ast import Dict import json import sqlite3 import os +import glob import time from typing import List, Optional import aiofiles from ..proto.agent_msg import AgentMsg from ..proto.ai_function import AIFunction, ParameterDefine,SimpleAIFunction,ActionNode,SimpleAIAction -from ..proto.agent_task import AgentTask, AgentTaskState,AgentTodoTask,AgentWorkLog,AgentTaskManager +from ..proto.agent_task import AgentTask, AgentTaskState,AgentTodo,AgentWorkLog,AgentTaskManager from ..storage.storage import AIStorage from ..frame.bus import AIBus from .llm_context import GlobaToolsLibrary @@ -86,11 +87,23 @@ class LocalAgentTaskManger(AgentTaskManager): - async def create_todos(self,owner_task_id:str,todos:List[AgentTodoTask]): + async def set_todos(self,owner_task_id:str,todos:List[AgentTodo]): 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: + directory = f"{self.root_path}/{owner_task_path}" + file_extension = "*.todo" + pattern = os.path.join(directory, file_extension) + files = glob.glob(pattern) + + for file in files: + os.remove(file) + logger.info(f"Deleted {file}") + except Exception as e: + logger.error("set_todos deleted todos failed:%s",e) + try: step_order = 0 for todo in todos: @@ -176,7 +189,7 @@ class LocalAgentTaskManger(AgentTaskManager): 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: + async def get_todo(self,todo_id:str) -> AgentTodo: todo_path = self._get_obj_path(todo_id) if todo_path is None: logger.error("get_todo:%s,not found!",todo_id) @@ -185,7 +198,7 @@ class LocalAgentTaskManger(AgentTaskManager): 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) + result_todo:AgentTodo = AgentTodo.from_dict(todo_dict) if result_todo: result_todo.todo_path = todo_path else: @@ -214,10 +227,11 @@ class LocalAgentTaskManger(AgentTaskManager): sub_task = await self.get_task_by_path(f"{task_path}/{sub_item}") if sub_task: sub_tasks.append(sub_task) - pass + + return sub_tasks - async def get_sub_todos(self,task_id:str) -> List[AgentTodoTask]: + async def get_sub_todos(self,task_id:str) -> List[AgentTodo]: task_path = self._get_obj_path(task_id) if task_path is None: return [] @@ -266,14 +280,14 @@ class LocalAgentTaskManger(AgentTaskManager): try: new_task_content = json.dumps(task.to_dict(),ensure_ascii=False) async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: - await f.write(new_task_content)) + await f.write(new_task_content) except Exception as e: logger.error("update_task failed:%s",e) return str(e) return None - async def update_todo(self,todo:AgentTodoTask): + async def update_todo(self,todo:AgentTodo): todo_path = self._get_obj_path(todo.todo_id) if todo_path is None: return f"todo {todo.todo_id} not found" @@ -424,6 +438,8 @@ class AgentWorkspace: ) GlobaToolsLibrary.get_instance().register_tool_function(create_task_action) + + async def cancel_task(parameters): _workspace = parameters.get("_workspace") if _workspace is None: @@ -498,3 +514,54 @@ class AgentWorkspace: update_task,parameters) GlobaToolsLibrary.get_instance().register_tool_function(update_task_ai_function) + async def set_todos(parameters): + _workspace : AgentWorkspace= parameters.get("_workspace") + if _workspace is None: + return "_workspace not found" + task_id = parameters.get("task_id") + task:AgentTask = await _workspace.task_mgr.get_task(task_id) + if task is None: + return f"task {task_id} not found" + todos = parameters.get("todos") + if todos is None: + return "todos not found" + await _workspace.task_mgr.set_todos(task_id,todos) + return "set todos ok" + + todo_demo = """ + [ + { + "title": "todo1", + "detail": "todo1 detail", + "tags": "tag1,tag2", + "due_date": "2021-01-01", + "priority": 1 + }, + ] + """ + parameters = ParameterDefine.create_parameters({ + "task_id": {"type": "string", "description": "task id which want to set todos"}, + "todos": {"type": "list", "description": f"List of todo, todo is a dict like {todo_demo}"}, + }) + set_todos_ai_function = SimpleAIFunction("agent.workspace.set_todos", + "set todos for task", + set_todos,parameters) + GlobaToolsLibrary.get_instance().register_tool_function(set_todos_ai_function) + + async def update_todo(parameters): + _workspace : AgentWorkspace= parameters.get("_workspace") + if _workspace is None: + return "_workspace not found" + todo_id = parameters.get("todo_id") + todo : AgentTodo = await _workspace.task_mgr.get_todo(todo_id) + if todo is None: + return f"todo {todo_id} not found" + + parameters = ParameterDefine.create_parameters({ + "todo_id": {"type": "string", "description": "todo id which want to update"}, + "new_state": {"type": "string", "description": "optional,new todo state: execute_ok , execute_failed, done or check_failed"}, + }) + update_todo_ai_function = SimpleAIFunction("agent.workspace.update_todo", + "update todo to new state", + update_todo,parameters) + GlobaToolsLibrary.get_instance().register_tool_function(update_todo_ai_function) diff --git a/src/aios/proto/agent_task.py b/src/aios/proto/agent_task.py index 703bdde..3e3c56f 100644 --- a/src/aios/proto/agent_task.py +++ b/src/aios/proto/agent_task.py @@ -1,5 +1,6 @@ # pylint:disable=E0402 from abc import ABC, abstractmethod +import json from typing import List, Optional import datetime import time @@ -31,9 +32,6 @@ class AgentTodoResult: result["op_list"] = self.op_list return result - - - class AgentTodo: TODO_STATE_WAIT_ASSIGN = "wait_assign" TODO_STATE_INIT = "init" @@ -220,7 +218,7 @@ class AgentTodoState(Enum): def from_str(value): return next((s for s in AgentTodoState.__members__.values() if s.value == value), None) -class AgentTodoTask: +class AgentTodo: def __init__(self) -> None: self.todo_id = "todo#" + uuid.uuid4().hex self.todo_path : str = None @@ -385,20 +383,30 @@ class AgentTask: return result +# 谁在什么时间做了什么 class AgentWorkLog: + # work type : [triage,plan,do,check] 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.owner_id:str = None # taskid or todoid + self.work_type:str = "" # 默认为普通类型的log,特殊类型的Log一般伴随着重要的状态改变 self.timestamp = time.time() self.content:str = None self.result:str = None self.meta : dict = None self.operator = None + + @classmethod + def create_by_content(cls,owner_id:str,work_type:str,content:str,operator:str) -> 'AgentWorkLog': + log = AgentWorkLog() + log.owner_id = owner_id + log.work_type = work_type + log.content = content + log.operator = operator + log.result = "OK" + return log - def to_dict(self) -> dict: - pass + class AgentTaskManager(ABC): def __init__(self) -> None: @@ -409,7 +417,7 @@ class AgentTaskManager(ABC): pass @abstractmethod - async def create_todos(self,owner_task_id:str,todos:List[AgentTodoTask]): + async def set_todos(self,owner_task_id:str,todos:List[AgentTodo]): # return todo_id pass @@ -430,7 +438,7 @@ class AgentTaskManager(ABC): # pass @abstractmethod - async def get_todo(self,todo_id:str) -> AgentTodoTask: + async def get_todo(self,todo_id:str) -> AgentTodo: pass @abstractmethod @@ -438,7 +446,7 @@ class AgentTaskManager(ABC): pass @abstractmethod - async def get_sub_todos(self,task_id:str) -> List[AgentTodoTask]: + async def get_sub_todos(self,task_id:str) -> List[AgentTodo]: pass #@abstractmethod @@ -454,7 +462,7 @@ class AgentTaskManager(ABC): pass @abstractmethod - async def update_todo(self,todo:AgentTodoTask): + async def update_todo(self,todo:AgentTodo): pass #@abstractmethod