From 5885301c191cee1d020b7970ca27ea07fd35ecea Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Sun, 7 Jan 2024 12:44:47 -0800 Subject: [PATCH] 1) Complete new Agent Behavior: triage_tasks 2) Fix bugs. --- rootfs/agents/Jarvis/agent.toml | 69 +++-- src/aios/__init__.py | 2 + src/aios/agent/agent.py | 79 ++++-- src/aios/agent/llm_do_task.py | 184 +++++++++++++ src/aios/agent/llm_process.py | 389 +++++++-------------------- src/aios/agent/llm_process_loader.py | 42 +++ src/aios/agent/workspace.py | 133 ++++++++- src/aios/proto/agent_task.py | 18 +- 8 files changed, 560 insertions(+), 356 deletions(-) create mode 100644 src/aios/agent/llm_do_task.py create mode 100644 src/aios/agent/llm_process_loader.py diff --git a/rootfs/agents/Jarvis/agent.toml b/rootfs/agents/Jarvis/agent.toml index bfb6cc4..c65e7de 100644 --- a/rootfs/agents/Jarvis/agent.toml +++ b/rootfs/agents/Jarvis/agent.toml @@ -15,22 +15,22 @@ Your name is Jarvis, the super personal assistant to the master, The focus of wo """ [behavior.on_message] -type="LLMAgentMessageProcess" +type="AgentMessageProcess" +# TODO: 是否应该自动记录 inner function和action的执行细节 process_description=""" -1. Based on your role, combined with existing information, make a brief and efficient reply. +1. Based on your role and the existing information, please think and then 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. 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. """ - +# Not work: tags: ['tag1', 'tag2'], #Optional,If the conversation involves important things and people, you can mark by 1-3 tags. 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.' + think:'$think step-by-step to be sure you have the right reply.' resp: '$What you want to reply', - tags: ['tag1', 'tag2'], #Optional,If the conversation involves important things and people, you can mark by 1-3 tags. actions: [{ name: '$action_name', $param_name: '$parm' #Optional, fill in only if the action has parameters. @@ -48,8 +48,35 @@ tools_tips = """ llm_context.actions.enable = ["agent.workspace.create_task"] llm_context.functions.enable = ["agent.workspace.list_task"] -[behavior.review_task] -type="ReviewTaskProcess" + +[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. +3. Organize tasks to remove tasks beyond your ability. And merge the repeated tasks.(update_task + cancel_task) +""" +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 can triage tasks well.' + resp : '$determine, summary what you do', + actions: [{ + name: '$action_name', + $param_name: '$parm' #Optional, fill in only if the action has parameters. + }] +} +""" +context="Your master now in {location}, time: {now}, weather: {weather}." + +llm_context.actions.enable = ["agent.workspace.update_task","agent.workspace.cancel_task","post_message"] + +[behavior.plan_task] +## 首次处理任务 +type="AgentPlanTask" process_description=""" 你得到的输入来自你自己之前记录在TaskList系统里的一个Task。现在你并不需要完成该Task，而是结合已知信息对Task进行一次Review.Review的过程是你独立完成的，你在形成结论的过程中可以使用工具，但不能和其它人交流。 1. 理性的思考如何一步一步的高效的，在潜在的截止时间前完成该Task。明确拒绝超出自己能力范围的Task。 @@ -75,20 +102,18 @@ The Response must be directly parsed by `python json.loads`. Here is an example: } """ # action_list: ['cancle','confirm', 'execute'] -LLMContext.action_list = ['cancle','confirm', 'execute'] +llm_context.actions.enable = ["agent.workspace.cancel_task"] +llm_context.functions.enable = ["agent.workspace.list_task"] context="Your master now in {location}, time: {now}, weather: {weather}." -known_info_tips = """ -""" +[behavior.review_task] +## 处理已经被LLM处理过的任务，包括处理首次出错的任务，处理被的任务 -tools_tips = """ - -""" [behavior.do] # do TODO -type="DoTodoProcess" +type="AgentDo" process_description=""" 1. 你的任务是结合自己的角色定义，手头的工具，已知信息、完成一个确定的TODO。完成该TODO后你会得到$200的小费。 2. 输入的TODO是来自你自己对一个Task的Plan结果。 @@ -111,12 +136,18 @@ The Response must be directly parsed by `python json.loads`. Here is an example: ] } """ +[behavior.check] +type="AgentCheck" -#[behavior.self_thinking] - - - -#[behavior.check] +[behavior.self_thinking] +# self thing的主要目的是对各种chatlog,worklog进行分析，并更新终结。 +type="AgentSelfThinking" +#[behavior.self_learning] +# self_learning的主要目的是根据自己的经验，主动的学习新的知识。这通常是一个专门整理知识库的Agent，一般的Agent谨慎开启 +#type="AgentSelfLearning" +#[behavior.self_improve] +# self_improve 是最后的行为，允许Agent结合自己的工作经验，改进自己的提示词（注意保留历史版本） +#type="AgentSelfImprove" diff --git a/src/aios/__init__.py b/src/aios/__init__.py index cba723e..7927c0b 100644 --- a/src/aios/__init__.py +++ b/src/aios/__init__.py @@ -12,6 +12,8 @@ from .agent.workflow import Workflow from .agent.agent_memory import AgentMemory from .agent.workspace import AgentWorkspace from .agent.llm_context import LLMProcessContext,GlobaToolsLibrary,SimpleLLMContext +from .agent.llm_process import BaseLLMProcess,LLMAgentBaseProcess +from .agent.llm_process_loader import LLMProcessLoader from .frame.compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType from .frame.compute_node import ComputeNode,LocalComputeNode diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index 2b31474..b348301 100644 --- a/src/aios/agent/agent.py +++ b/src/aios/agent/agent.py @@ -12,6 +12,8 @@ import datetime import copy import sys + + from ..proto.agent_msg import AgentMsg from ..proto.ai_function import * from ..proto.agent_task import AgentTaskState,AgentTask,AgentTodo,AgentTodoResult @@ -19,23 +21,23 @@ from ..proto.compute_task import * from .agent_base import * from .llm_process import * +from .llm_process_loader import * +from .llm_do_task import * from .chatsession import * -from ..environment.workspace_env import WorkspaceEnvironment, TodoListType +from ..environment.workspace_env import WorkspaceEnvironment, TodoListType from ..frame.contact_manager import ContactManager,Contact,FamilyMember from ..frame.compute_kernel import ComputeKernel from ..frame.bus import AIBus from ..environment.environment import * from ..environment.workspace_env import WorkspaceEnvironment from ..storage.storage import AIStorage - from ..knowledge import * from ..utils import video_utils, image_utils from ..proto.compute_task import ComputeTaskResult,ComputeTaskResultCode,LLMPrompt,LLMResult logger = logging.getLogger(__name__) - class AIAgentTemplete: def __init__(self) -> None: self.llm_model_name:str = "gpt-4-0613" @@ -200,7 +202,15 @@ class AIAgent(BaseAIAgent): return self.agent_prompt + async def _get_context_info(self) -> Dict: + context_info = {} + context_info["location"] = "SanJose" + context_info["now"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + context_info["weather"] = "Partly Cloudy, 60°F" + + return context_info + async def llm_process_msg(self,msg:AgentMsg) -> AgentMsg: need_process:bool = True if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: @@ -218,8 +228,10 @@ class AIAgent(BaseAIAgent): resp_msg = msg.create_group_resp_msg(self.agent_id,"") return resp_msg + context_info = await self._get_context_info() input_parms = { - "msg":msg + "msg":msg, + "context_info":context_info } msg_process = self.behaviors.get("on_message") llm_result : LLMResult = await msg_process.process(input_parms) @@ -233,38 +245,47 @@ class AIAgent(BaseAIAgent): return resp_msg async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: - msg.context_info = {} - msg.context_info["location"] = "SanJose" - msg.context_info["now"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - msg.context_info["weather"] = "Partly Cloudy, 60°F" return await self.llm_process_msg(msg) - async def llm_review_tasklist(self): - llm_process : BaseLLMProcess = self.behaviors.get("review_task") + async def llm_triage_tasklist(self): + llm_process : BaseLLMProcess = self.behaviors.get("triage_tasks") if llm_process: if self.prviate_workspace: tasklist = await self.prviate_workspace.task_mgr.list_task() if tasklist: - for agent_task in tasklist: - if self.agent_energy <= 0: - break + input_parms = { + "tasklist":tasklist, + "context_info": await self._get_context_info() + } + llm_result : LLMResult = await llm_process.process(input_parms) + if llm_result.state == LLMResultStates.ERROR: + logger.error(f"llm process triage_tasks error:{llm_result.compute_error_str}") + elif llm_result.state == LLMResultStates.IGNORE: + logger.info(f"llm process triage_tasks ignore!") + else: + logger.info(f"llm process triage_tasks ok!,think is:{llm_result.resp}") + self.agent_energy -= 5 - if agent_task.state == AgentTaskState.TASK_STATE_WAIT: - input_parms = { - "task":agent_task - } - llm_result : LLMResult = await llm_process.process(input_parms) - if llm_result.state == LLMResultStates.ERROR: - logger.error(f"llm process review_task error:{llm_result.error_str}") - continue - elif llm_result.state == LLMResultStates.IGNORE: - logger.info(f"llm process review_task ignore!") - continue - else: - determine = llm_result.raw_result.get("determine") - logger.info(f"llm process review_task ok!,think is:{determine}") - self.agent_energy -= 1 + # for agent_task in tasklist: + # if self.agent_energy <= 0: + # break + + # if agent_task.state == AgentTaskState.TASK_STATE_WAIT: + # input_parms = { + # "task":agent_task + # } + # llm_result : LLMResult = await llm_process.process(input_parms) + # if llm_result.state == LLMResultStates.ERROR: + # logger.error(f"llm process review_task error:{llm_result.error_str}") + # continue + # elif llm_result.state == LLMResultStates.IGNORE: + # logger.info(f"llm process review_task ignore!") + # continue + # else: + # determine = llm_result.raw_result.get("determine") + # logger.info(f"llm process review_task ok!,think is:{determine}") + # self.agent_energy -= 1 @@ -543,7 +564,7 @@ class AIAgent(BaseAIAgent): if self.agent_energy <= 1: continue - await self.llm_review_tasklist() + await self.llm_triage_tasklist() # complete & check todo #await self._llm_run_todo_list(TodoListType.TO_WORK) diff --git a/src/aios/agent/llm_do_task.py b/src/aios/agent/llm_do_task.py new file mode 100644 index 0000000..7f7bd59 --- /dev/null +++ b/src/aios/agent/llm_do_task.py @@ -0,0 +1,184 @@ +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 ..frame.compute_kernel import ComputeKernel + +from .agent_memory import AgentMemory +from .workspace import AgentWorkspace +from .llm_context import LLMProcessContext,GlobaToolsLibrary, SimpleLLMContext +from .llm_process import BaseLLMProcess,LLMAgentBaseProcess + +from abc import ABC,abstractmethod +import copy +import json +import datetime +from datetime import datetime +from typing import Any, Callable, Coroutine, Optional,Dict,Awaitable,List +from enum import Enum +import logging + +logger = logging.getLogger(__name__) + +class AgentTriageTaskList(LLMAgentBaseProcess): + def __init__(self) -> None: + super().__init__() + + + async def load_from_config(self,config:dict) -> bool: + if await super().load_from_config(config) is False: + return False + + async def prepare_prompt(self,input:Dict) -> LLMPrompt: + prompt = LLMPrompt() + + task_list:List[AgentTask] = input.get("tasklist") + context_info = input.get("context_info") + if task_list is None: + logger.error(f"tasklist not found in input") + return None + + task_dict_list = [] + for task in task_list: + task_dict_list.append(task.to_dict()) + + prompt.append_user_message(json.dumps(task_dict_list,ensure_ascii=False)) + + system_prompt_dict = self.prepare_role_system_prompt(context_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 + 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() + await self._execute_actions(actions,action_params) + + + +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]: + + + 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)) + prompt.append_user_message(json.dumps(agent_task.to_dict(),ensure_ascii=False)) + return prompt + + + async def get_review_task_actions(self) -> Dict[str,Dict]: + pass + + async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: + pass + + async def post_llm_process(self,actions:List[ActionNode]) -> bool: + pass + +class AgentReviewTask(BaseLLMProcess): + def __init__(self) -> None: + super().__init__() + + async def load_from_config(self, config: dict): + 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_for_exec(self,func_name:str) -> AIFunction: + pass + + async def post_llm_process(self,actions:List[ActionNode]) -> bool: + pass + + +class AgentCheck(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_for_exec(self,func_name:str) -> AIFunction: + pass + + async def post_llm_process(self,actions:List[ActionNode]) -> bool: + pass + +class AgentDo(BaseLLMProcess): + def __init__(self) -> None: + super().__init__() + + async def load_from_config(self, config: dict): + 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_for_exec(self,func_name:str) -> AIFunction: + pass + + async def post_llm_process(self,actions:List[ActionNode]) -> bool: + pass diff --git a/src/aios/agent/llm_process.py b/src/aios/agent/llm_process.py index a32d124..d372613 100644 --- a/src/aios/agent/llm_process.py +++ b/src/aios/agent/llm_process.py @@ -25,8 +25,6 @@ logger = logging.getLogger(__name__) MIN_PREDICT_TOKEN_LEN = 32 - - class BaseLLMProcess(ABC): def __init__(self) -> None: self.behavior:str = None #行为名字 @@ -42,6 +40,8 @@ class BaseLLMProcess(ABC): self.max_prompt_token = 1000 # not include input prompt self.timeout = 1800 # 30 min + self.llm_context:LLMProcessContext = None + @abstractmethod async def prepare_prompt(self,input:Dict) -> LLMPrompt: pass @@ -50,6 +50,10 @@ class BaseLLMProcess(ABC): async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: pass + @abstractmethod + def prepare_inner_function_context_for_exec(self,inner_func_name:str,parameters:Dict): + return + @abstractmethod async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: pass @@ -80,8 +84,6 @@ class BaseLLMProcess(ABC): def _format_content_by_env_value(self,content:str,env)->str: return content.format_map(env) - def prepare_inner_function_context_for_exec(self,inner_func_name:str,parameters:Dict): - return async def _execute_inner_func(self,inner_func_call_node:Dict,prompt: LLMPrompt,stack_limit = 1) -> ComputeTaskResult: arguments = None @@ -205,68 +207,12 @@ class LLMAgentBaseProcess(BaseLLMProcess): self.process_description:str = None self.reply_format:str = None self.context : str = None - - self.known_info_tips :str = None - self.tools_tips:str = None - + self.workspace : AgentWorkspace = None # If Workspace is not none , enable Agent Tasklist self.memory : AgentMemory = None + self.enable_kb : bool = False self.kb = None - async def load_default_config(self) -> bool: - return True - - - async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: - if is_load_default: - await self.load_default_config() - - 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") - - if config.get("known_info_tips"): - self.known_info_tips = config.get("known_info_tips") - - if config.get("tools_tips"): - self.tools_tips = config.get("tools_tips") - - if config.get("knowledge_base"): - self.kb = config.get("knowledge_base") - - -class LLMAgentMessageProcess(BaseLLMProcess): - def __init__(self) -> None: - super().__init__() - - self.role_description:str = None - self.process_description:str = None - self.reply_format:str = None - self.context : str = None - - self.known_info_tips :str = None - self.tools_tips:str = None - - self.workspace : AgentWorkspace = None # If Workspace is not none , enable Agent Tasklist - self.memory : AgentMemory = None - self.enable_kb = False - self.kb = None - - self.llm_context : LLMProcessContext = None - async def initial(self,params:Dict = None) -> bool: self.memory = params.get("memory") if self.memory is None: @@ -274,9 +220,7 @@ class LLMAgentMessageProcess(BaseLLMProcess): return False self.workspace = params.get("workspace") - return True - async def load_default_config(self) -> bool: return True @@ -302,27 +246,86 @@ class LLMAgentMessageProcess(BaseLLMProcess): if config.get("context"): self.context = config.get("context") - if config.get("known_info_tips"): - self.known_info_tips = config.get("known_info_tips") - - if config.get("tools_tips"): - self.tools_tips = config.get("tools_tips") - - if config.get("enable_kb"): - self.enable_kb = config.get("enable_kb") == "true" - self.llm_context = SimpleLLMContext() if config.get("llm_context"): self.llm_context.load_from_config(config.get("llm_context")) - + if config.get("enable_kb"): + self.enable_kb = config.get("enable_kb") == "true" + + def prepare_role_system_prompt(self,context_info:Dict) -> Dict: + system_prompt_dict = {} + # System Prompt + ## LLM的身份说明 + system_prompt_dict["role_description"] = self.role_description + #prompt.append_system_message(self.role_description) + + ## 处理信息的流程说明 + system_prompt_dict["process_rule"] = self.process_description + #prompt.append_system_message(self.process_description) + ### 回复的格式 + system_prompt_dict["reply_format"] = self.reply_format + #prompt.append_system_message(self.reply_format) + + ## Context + context = self._format_content_by_env_value(self.context,context_info) + system_prompt_dict["context"] = context + #prompt.append_system_message(context) + + system_prompt_dict["support_actions"] = self.get_action_desc() + + return system_prompt_dict + + def prepare_inner_function_context_for_exec(self,inner_func_name:str,parameters:Dict): + parameters["_workspace"] = self.workspace + + def get_action_desc(self) -> Dict: + result = {} + actions_list = self.llm_context.get_all_ai_action() + for action in actions_list: + result[action.get_name()] = action.get_description() + return result + + async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: + return self.llm_context.get_ai_function(func_name) + + async def _execute_actions(self,actions:List[ActionNode],action_params:Dict): + for action_item in actions: + op : AIAction = self.llm_context.get_ai_action(action_item.name) + if op: + if action_item.parms is None: + action_item.parms = {} + + real_parms = {**action_params,**action_item.parms} + + action_item.parms["_result"] = await op.execute(real_parms) + action_item.parms["_end_at"] = datetime.now() + else: + logger.warn(f"action {action_item.name} not found") + return False + + +class AgentMessageProcess(LLMAgentBaseProcess): + def __init__(self) -> None: + super().__init__() + + async def load_default_config(self) -> bool: + return True + + async def load_from_config(self, config: dict,is_load_default=True) -> Coroutine[Any, Any, bool]: + if is_load_default: + await self.load_default_config() + + if await super().load_from_config(config) is False: + return False + + def check_and_to_base64(self, image_path: str) -> str: if image_utils.is_file(image_path): return image_utils.to_base64(image_path, (1024, 1024)) else: return image_path - async def get_prompt_from_msg(self,msg:AgentMsg) -> LLMPrompt: msg_prompt = LLMPrompt() if msg.is_image_msg(): @@ -356,13 +359,6 @@ class LLMAgentMessageProcess(BaseLLMProcess): return msg_prompt - async def get_action_desc(self) -> Dict: - result = {} - actions_list = self.llm_context.get_all_ai_action() - for action in actions_list: - result[action.get_name()] = action.get_description() - return result - async def sender_info(self,msg:AgentMsg)->str: sender_id = msg.sender #TODO Is sender an agent? @@ -386,6 +382,7 @@ class LLMAgentMessageProcess(BaseLLMProcess): # User Prompt ## Input Msg msg : AgentMsg = input.get("msg") + context_info = input.get("context_info") if msg is None: logger.error(f"LLMAgeMessageProcess prepare_prompt failed! input msg not found") return None @@ -395,31 +392,8 @@ class LLMAgentMessageProcess(BaseLLMProcess): return None prompt.append(msg_prompt) - system_prompt_dict = {} - - # System Prompt - ## LLM的身份说明 - system_prompt_dict["role_description"] = self.role_description - #prompt.append_system_message(self.role_description) - - ## 处理信息的流程说明 - system_prompt_dict["process_rule"] = self.process_description - #prompt.append_system_message(self.process_description) - ### 回复的格式 - system_prompt_dict["reply_format"] = self.reply_format - #prompt.append_system_message(self.reply_format) - ### 修改chatlog的action - ### 修改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 （文本替换）,是否应该覆盖全部消息 - context = self._format_content_by_env_value(self.context,msg.context_info) - system_prompt_dict["context"] = context - #prompt.append_system_message(context) + ## 通用的角色相关的系统提示词 + system_prompt_dict = self.prepare_role_system_prompt(context_info) ## 已知信息 known_info = {} @@ -441,10 +415,6 @@ class LLMAgentMessageProcess(BaseLLMProcess): #prompt.append_system_message(await self.get_log_summary(self,msg)) system_prompt_dict["known_info"] = known_info - ## 可以使用的tools(inner function)的解释，注意不定义该tips,则不会导入任何workspace中的tools - if self.tools_tips: - system_prompt_dict["tools_tips"] = self.tools_tips - prompt.inner_functions =LLMProcessContext.aifunctions_to_inner_functions(self.llm_context.get_all_ai_functions()) if self.workspace: #TODO eanble workspace functions? @@ -461,11 +431,6 @@ class LLMAgentMessageProcess(BaseLLMProcess): return prompt - def prepare_inner_function_context_for_exec(self,inner_func_name:str,parameters:Dict): - parameters["_workspace"] = self.workspace - - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - return self.llm_context.get_ai_function(func_name) async def post_llm_process(self,actions:List[ActionNode],input:Dict,llm_result:LLMResult) -> bool: msg:AgentMsg = input.get("msg") @@ -476,137 +441,24 @@ class LLMAgentMessageProcess(BaseLLMProcess): llm_result.raw_result["_resp_msg"] = resp_msg - for action_item in actions: - op : AIAction = self.llm_context.get_ai_action(action_item.name) - if op: - if action_item.parms is None: - action_item.parms = {} + action_params = {} + action_params["_input"] = input + action_params["_memory"] = self.memory + action_params["_workspace"] = self.workspace + action_params["_resp_msg"] = resp_msg + action_params["_llm_result"] = llm_result + action_params["_agentid"] = self.memory.agent_id + action_params["_start_at"] = datetime.now() - action_item.parms["_input"] = input - action_item.parms["_memory"] = self.memory - action_item.parms["_workspace"] = self.workspace - action_item.parms["_resp_msg"] = resp_msg - action_item.parms["_llm_result"] = llm_result - action_item.parms["_start_at"] = datetime.now() - action_item.parms["_agentid"] = self.memory.agent_id - - action_item.parms["_result"] = await op.execute(action_item.parms) - action_item.parms["_end_at"] = datetime.now() - else: - logger.warn(f"action {action_item.name} not found") - return False + await self._execute_actions(actions,action_params) chatsession = self.memory.get_session_from_msg(msg) chatsession.append(msg) chatsession.append(resp_msg) - return True - - - -class ReviewTaskProcess(BaseLLMProcess): - 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]: - - 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)) - prompt.append_user_message(json.dumps(agent_task.to_dict(),ensure_ascii=False)) - return prompt - - - async def get_review_task_actions(self) -> Dict[str,Dict]: - pass - - async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: - pass - - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass - -class QuickReviewTaskProcess(BaseLLMProcess): - def __init__(self) -> None: - super().__init__() - - async def load_from_config(self, config: dict): - 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_for_exec(self,func_name:str) -> AIFunction: - pass - - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass - -class DoTodoProcess(BaseLLMProcess): - def __init__(self) -> None: - super().__init__() - - async def load_from_config(self, config: dict): - 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_for_exec(self,func_name:str) -> AIFunction: - pass - - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass - - -class CheckTodoProcess(BaseLLMProcess): +class AgentSelfLearning(BaseLLMProcess): def __init__(self) -> None: super().__init__() @@ -624,25 +476,7 @@ class CheckTodoProcess(BaseLLMProcess): async def post_llm_process(self,actions:List[ActionNode]) -> bool: pass -class SelfLearningProcess(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_for_exec(self,func_name:str) -> AIFunction: - pass - - async def post_llm_process(self,actions:List[ActionNode]) -> bool: - pass - -class SelfThinkingProcess(BaseLLMProcess): +class AgentSelfThinking(BaseLLMProcess): def __init__(self) -> None: super().__init__() @@ -727,43 +561,10 @@ class SelfThinkingProcess(BaseLLMProcess): async def post_llm_process(self,actions:List[ActionNode]) -> bool: pass - -class LLMProcessLoader: + +class AgentSelfImprove(BaseLLMProcess): def __init__(self) -> None: - self.loaders : Dict[str,Callable[[dict],Awaitable[BaseLLMProcess]]] = {} - return - - @classmethod - def get_instance(cls)->"LLMProcessLoader": - if not hasattr(cls,"_instance"): - cls._instance = LLMProcessLoader() - return cls._instance - - def register_loader(self, typename:str,loader:Callable[[dict],Awaitable[BaseLLMProcess]]): - self.loaders[typename] = loader - - async def load_from_config(self,config:dict) -> BaseLLMProcess: - llm_type_name = config.get("type") - if llm_type_name: - loader = self.loaders.get(llm_type_name) - if loader: - return await loader(config) - - selected_type = globals().get(llm_type_name) - if selected_type: - result : BaseLLMProcess = selected_type() - load_result = await result.load_from_config(config) - if load_result is False: - logger.warn(f"load LLMProcess {llm_type_name} from config failed! load_from_config return False") - return None - else: - return result - - - logger.warn(f"load LLMProcess {llm_type_name} from config failed! type not found") - return None - - + super().__init__() diff --git a/src/aios/agent/llm_process_loader.py b/src/aios/agent/llm_process_loader.py new file mode 100644 index 0000000..912b3a0 --- /dev/null +++ b/src/aios/agent/llm_process_loader.py @@ -0,0 +1,42 @@ +from .llm_process import BaseLLMProcess, AgentMessageProcess,AgentSelfThinking,AgentSelfLearning,AgentSelfImprove +from .llm_do_task import AgentTriageTaskList,AgentPlanTask,AgentReviewTask,AgentDo,AgentCheck + +from typing import Awaitable, Callable, Coroutine, Dict, List, Any +import logging + +logger = logging.getLogger(__name__) + +class LLMProcessLoader: + def __init__(self) -> None: + self.loaders : Dict[str,Callable[[dict],Awaitable[BaseLLMProcess]]] = {} + return + + @classmethod + def get_instance(cls)->"LLMProcessLoader": + if not hasattr(cls,"_instance"): + cls._instance = LLMProcessLoader() + return cls._instance + + def register_loader(self, typename:str,loader:Callable[[dict],Awaitable[BaseLLMProcess]]): + self.loaders[typename] = loader + + async def load_from_config(self,config:dict) -> BaseLLMProcess: + llm_type_name = config.get("type") + if llm_type_name: + loader = self.loaders.get(llm_type_name) + if loader: + return await loader(config) + + selected_type = globals().get(llm_type_name) + if selected_type: + result : BaseLLMProcess = selected_type() + load_result = await result.load_from_config(config) + if load_result is False: + logger.warn(f"load LLMProcess {llm_type_name} from config failed! load_from_config return False") + return None + else: + return result + + + logger.warn(f"load LLMProcess {llm_type_name} from config failed! type not found") + return None \ No newline at end of file diff --git a/src/aios/agent/workspace.py b/src/aios/agent/workspace.py index 8f09a3e..c28ede8 100644 --- a/src/aios/agent/workspace.py +++ b/src/aios/agent/workspace.py @@ -3,16 +3,18 @@ from ast import Dict import json import sqlite3 import os -import logging +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,AgentTodoTask,AgentWorkLog,AgentTaskManager +from ..proto.agent_task import AgentTask, AgentTaskState,AgentTodoTask,AgentWorkLog,AgentTaskManager from ..storage.storage import AIStorage +from ..frame.bus import AIBus from .llm_context import GlobaToolsLibrary +import logging logger = logging.getLogger(__name__) class LocalAgentTaskManger(AgentTaskManager): @@ -262,8 +264,9 @@ class LocalAgentTaskManger(AgentTaskManager): async def update_task(self,task:AgentTask): detail_path = f"{self.root_path}/{task.task_path}/detail" 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(json.dumps(task.to_dict(),ensure_ascii=False)) + await f.write(new_task_content)) except Exception as e: logger.error("update_task failed:%s",e) return str(e) @@ -276,8 +279,9 @@ class LocalAgentTaskManger(AgentTaskManager): return f"todo {todo.todo_id} not found" try: + new_todo_content = json.dumps(todo.to_dict(),ensure_ascii=False) async with aiofiles.open(todo_path, mode='w', encoding="utf-8") as f: - await f.write(json.dumps(todo.to_dict(),ensure_ascii=False)) + await f.write(new_todo_content) except Exception as e: logger.error("update_todo failed:%s",e) return str(e) @@ -315,9 +319,86 @@ class AgentWorkspace: self.owner_id : str = owner_id self.task_mgr : AgentTaskManager = LocalAgentTaskManger(owner_id) - @staticmethod def register_ai_functions(): + async def post_message(parameters): + _agent_id = parameters.get("_agentid") + if _agent_id is None: + return "_agentid not found" + + target = parameters.get("target") + if target is None: + return "target not found" + message = parameters.get("message") + if message is None: + return "message not found" + topic = parameters.get("topic") + + msg = AgentMsg() + msg.sender = _agent_id + msg.body = message + msg.topic = topic + msg.target = target + msg.create_time = time.time() + + is_post_ok = await AIBus.get_default_bus().post_message(msg) + if is_post_ok: + return "post message ok!" + else: + return f"post message to {target} failed!" + + parameters = ParameterDefine.create_parameters({ + "target": {"type": "string", "description": "target agent/contact id"}, + "topic": {"type": "string", "description": "optional, message topic"}, + "message": {"type": "string", "description": "message content"}, + }) + post_message_action = SimpleAIFunction( + "post_message", + "Post a message to target agent/contact", + post_message, + parameters, + ) + GlobaToolsLibrary.get_instance().register_tool_function(post_message_action) + + async def send_message(parameters): + _agent_id = parameters.get("_agentid") + if _agent_id is None: + return "_agentid not found" + + target = parameters.get("target") + if target is None: + return "target not found" + message = parameters.get("message") + if message is None: + return "message not found" + topic = parameters.get("topic") + + msg = AgentMsg() + msg.sender = _agent_id + msg.body = message + msg.topic = topic + msg.target = target + msg.create_time = time.time() + + resp = await AIBus.get_default_bus().send_message(msg) + if resp: + return f"resp is : {resp.body}" + else: + return f"send message to {target} failed!" + + parameters = ParameterDefine.create_parameters({ + "target": {"type": "string", "description": "target agent/contact id"}, + "topic": {"type": "string", "description": "optional, message topic"}, + "message": {"type": "string", "description": "message content"}, + }) + send_message_action = SimpleAIFunction( + "send_message", + "send a message to target agent/contact, and wait for reply", + send_message, + parameters, + ) + GlobaToolsLibrary.get_instance().register_tool_function(send_message_action) + async def create_task(params): _workspace = params.get("_workspace") _agent_id = params.get("_agentid") @@ -348,7 +429,7 @@ class AgentWorkspace: if _workspace is None: return "_workspace not found" task_id = parameters.get("task_id") - task = await _workspace.task_mgr.get_task(task_id) + task : AgentTask = await _workspace.task_mgr.get_task(task_id) if task is None: return f"task {task_id} not found" task.state = "cancel" @@ -380,4 +461,40 @@ class AgentWorkspace: "list all tasks in json format", list_task,{}) GlobaToolsLibrary.get_instance().register_tool_function(list_task_ai_function) - \ No newline at end of file + + async def update_task(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" + if parameters.get("title"): + task.title = parameters.get("title") + if parameters.get("detail"): + task.detail = parameters.get("detail") + if parameters.get("priority"): + task.priority = parameters.get("priority") + if parameters.get("new_state"): + task.state = AgentTaskState.from_str(parameters.get("new_state")) + if parameters.get("next_do_date"): + task.next_do_date = parameters.get("next_do_date") + if parameters.get("due_date"): + task.due_date = parameters.get("due_date") + await _workspace.task_mgr.update_task(task) + return "update task ok" + parameters = ParameterDefine.create_parameters({ + "task_id": {"type": "string", "description": "task id which want to update"}, + "new_state": {"type": "string", "description": "optional,new task state: cancel or done"}, + "next_do_date": {"type": "string", "description": "optional,confirm task next do date"}, + "priority": {"type": "int", "description": "optional,task priority from 1-10"}, + "title": {"type": "string", "description": "optional, new task title"}, + "detail": {"type": "string", "description": "optional, new task detail(simple task can not be filled)"}, + "due_date": {"type": "string", "description": "optional,new task due date"}, + }) + update_task_ai_function = SimpleAIFunction("agent.workspace.update_task", + "update task to new state", + update_task,parameters) + GlobaToolsLibrary.get_instance().register_tool_function(update_task_ai_function) + diff --git a/src/aios/proto/agent_task.py b/src/aios/proto/agent_task.py index 223e087..703bdde 100644 --- a/src/aios/proto/agent_task.py +++ b/src/aios/proto/agent_task.py @@ -254,7 +254,7 @@ class AgentTask: # 确定的执行时间（执行条件） self.next_do_time = None # 如果next check time设置，说明任务适合在该时间点可能具备执行调教，尝试检查并执行 - self.next_check_time = None + #self.next_check_time = None self.depend_task_ids = [] #self.step_todo_ids = [] @@ -279,6 +279,12 @@ class AgentTask: if self.state == AgentTaskState.TASK_STATE_FAILED: return True + + if self.due_date: + if self.due_date < time.time(): + self.state = AgentTaskState.TASK_STATE_EXPIRED + return True + return False def to_dict(self) -> dict: @@ -295,8 +301,8 @@ class AgentTask: 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() + #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() @@ -327,9 +333,9 @@ class AgentTask: 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() + #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")