# coding=utf-8 # Copyright 2023 The AIWaves Inc. team. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """LLM autonoumous agent""" from LLM.base_LLM import * from Component import * from Action import Action from Prompt import * headers = { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no", } class Agent: """ Auto agent, input the JSON of SOP. """ # Agent should have args: agents,states def __init__(self, name, agent_state_roles, **kwargs) -> None: self.state_roles = agent_state_roles self.name = name self.style = kwargs["style"] self.LLMs = kwargs["LLMs"] self.LLM = None self.is_user = kwargs["is_user"] self.begins = kwargs["begins"] if "begins" in kwargs else False self.current_role = "" self.long_term_memory = [] self.short_term_memory = "" self.current_state = None self.first_speak = True self.environment = None @classmethod def from_config(cls, config_path): """ Initialize agents based on json file Return: agents(dict) : key:agent_name;value:class(Agent) names_to_roles(dict) : key:state_name value:(dict; (key:agent_name ; value:agent_role)) roles_to_names(dict) : key:state_name value:(dict; (key:agent_role ; value:agent_name)) """ with open(config_path) as f: config = json.load(f) roles_to_names = {} names_to_roles = {} agents = {} user_names = json.loads(os.environ["User_Names"]) if "User_Names" in os.environ else [] for agent_name, agent_dict in config["agents"].items(): agent_state_roles = {} agent_LLMs = {} agent_begins = {} for state_name, agent_role in agent_dict["roles"].items(): agent_begins[state_name] = {} if state_name not in roles_to_names: roles_to_names[state_name] = {} if state_name not in names_to_roles: names_to_roles[state_name] = {} roles_to_names[state_name][agent_role] = agent_name names_to_roles[state_name][agent_name] = agent_role agent_state_roles[state_name] = agent_role current_state = config["states"][state_name] current_state["roles"] = list(current_state["agent_states"].keys()) if "roles" not in current_state else current_state["roles"] current_state_begin_role = current_state["begin_role"] if "begin_role" in current_state else current_state["roles"][0] agent_begins[state_name]["is_begin"] = current_state_begin_role==agent_role if "begin_role" in current_state else False agent_begins[state_name]["begin_query"] = current_state["begin_query"] if "begin_query" in current_state else " " agent_LLMs[state_name] = init_LLM("logs"+os.sep+f"{agent_name}",**current_state["agent_states"][agent_role]) agents[agent_name] = cls( agent_name, agent_state_roles, LLMs=agent_LLMs, is_user=agent_name in user_names, style = agent_dict["style"], begins = agent_begins ) assert len(config["agents"].keys()) != 2 or (roles_to_names[config["root"]][config["states"][config["root"]]["begin_role"]] not in user_names and "begin_query" in config["states"][config["root"]]),"In a single-agent scenario, there must be an opening statement and it must be the agent" return agents, roles_to_names, names_to_roles def step(self, current_state,input=""): """ return actions by current state and environment Return: action(Action) """ current_state.chat_nums +=1 state_begin = current_state.is_begin agent_begin = self.begins[current_state.name]["is_begin"] self.begins[current_state.name]["is_begin"] = False current_state.is_begin = False environment = self.environment self.current_state = current_state # 先根据当前环境更新信息 # First update the information according to the current environment response = " " res_dict = {} if self.is_user: response = f"{self.name}:{input}" else: if len(environment.shared_memory["long_term_memory"])>0: current_history = self.observe() self.long_term_memory.append(current_history) if agent_begin: response = (char for char in self.begins[current_state.name]["begin_query"]) else: response,res_dict = self.act() action_dict = { "response": response, "res_dict": res_dict, "role": self.state_roles[current_state.name], "name": self.name, "state_begin" : state_begin, "agent_begin" : agent_begin, "is_user" : self.is_user } return Action(**action_dict) def act(self): """ return actions by the current state """ current_state = self.current_state chat_history = self.long_term_memory current_LLM = self.LLMs[current_state.name] system_prompt, last_prompt, res_dict = self.compile() response = current_LLM.get_response( chat_history, system_prompt, last_prompt, stream=True ) return response,res_dict def update_memory(self, memory): self.long_term_memory.append( {"role": "assistant", "content": memory.content} ) MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"]) environment = self.environment current_chat_history_idx = environment.current_chat_history_idx if environment.environment_type == "competive" else 0 current_long_term_memory = environment.shared_memory["long_term_memory"][current_chat_history_idx:] last_conversation_idx = environment._get_agent_last_conversation_idx(self,current_long_term_memory) if len(current_long_term_memory)-last_conversation_idx >= MAX_CHAT_HISTORY: current_state = self.current_state current_role = self.state_roles[current_state.name] current_component_dict = current_state.components[current_role] # get chat history from new conversation conversations = environment._get_agent_new_memory(self,current_long_term_memory) # get summary summary_prompt = ( current_state.summary_prompt[current_role] if current_state.summary_prompt else f"""your name is {self.name},your role is{current_component_dict["style"].role},your task is {current_component_dict["task"].task}.\n""" ) summary_prompt =eval(Agent_summary_system_prompt) summary = self.LLMs[current_state.name].get_response(None, summary_prompt,stream = False) self.short_term_memory = summary def compile(self): """ get prompt from state depend on your role Return: system_prompt:system_prompt for agents's LLM last_prompt:last_prompt for agents's LLM res_dict(dict): Other return from tool component.For example: search engine results """ current_state = self.current_state self.current_roles = self.state_roles[current_state.name] current_state_name = current_state.name self.LLM = self.LLMs[current_state_name] components = current_state.components[self.state_roles[current_state_name]] system_prompt = self.current_state.environment_prompt last_prompt = "" res_dict = {} for component in components.values(): if isinstance(component, (OutputComponent, LastComponent)): last_prompt = last_prompt + "\n" + component.get_prompt(self) elif isinstance(component, PromptComponent): system_prompt = ( system_prompt + "\n" + component.get_prompt(self) ) elif isinstance(component, ToolComponent): response = component.func(self) if "prompt" in response and response["prompt"]: last_prompt = last_prompt + "\n" + response["prompt"] res_dict.update(response) name = self.name query = self.environment.shared_memory["long_term_memory"][-1] if len(self.environment.shared_memory["long_term_memory"]) else "" last_prompt = eval(Agent_last_prompt) system_prompt = eval(Agent_system_prompt) return system_prompt, last_prompt, res_dict def observe(self): """ Update one's own memory according to the current environment, including: updating short-term memory; updating long-term memory """ return self.environment._observe(self) def generate_sop(self): pass def reflection(self): pass