✨States & Environment & Action
States
Definition:
As previously mentioned, we utilize SOP to operate the autonomous agent. Our SOP reasoning graph is composed of various States. These states play distinct roles, contributing to the entire system. We’ve developed a straightforward state, the SIMPLIST state, primarily based on LLM.
Attributes & Examples:
Basic codes for State class are as follows:
class State: """ Sub-scenes of role activities, responsible for storing the tasks that each role needs to do """ def __init__(self, **kwargs): self.next_states = {} self.name = kwargs["name"] self.environment_prompt = ( kwargs["environment_prompt"] if "environment_prompt" in kwargs else "" ) self.roles = kwargs["roles"] if "roles" in kwargs else [0] self.begin_role = ( kwargs["begin_role"] if "begin_role" in kwargs else self.roles[0] ) self.begin_query = kwargs["begin_query"] if "begin_query" in kwargs else None self.is_begin = True self.summary_prompt = ( kwargs["summary_prompt"] if "summary_prompt" in kwargs else None ) self.current_role = self.begin_role self.components = ( self.init_components(kwargs["agent_states"]) if "agent_states" in kwargs else {} ) self.index = ( self.roles.index(self.begin_role) if self.begin_role in self.roles else 0 ) self.chat_nums = 0
Part of the attributes are shown below:
name: The only tag of the state.
components: Contains components with different roles in this state. When different agents perform tasks in this state, they will obtain components from them according to their own roles.
environment_prompt: For the description of the current state, it will be added in front of the system prompt of each agent in the state.
summary_prompt: Prompt initiated from the system. For detailed information, please turn to Component page.
begin_role & begin_query: Role and query of the agents which is set at the beginning of the conversation.
next_states: Relations between states. Extraordinarily useful when the state graph is relatively sophisticated.
Methods:
Our states provide one single method, namely init_components, which is shown as follows:
init_components
The init_components method receives various types of components, and then classify and place them. Basic codes are omitted.
Environment
Definition:
Apparently, every autonomous agent should adjust to different circumstances, thus changing their chatting style and information immediately. To help manage their self-evolution, we established the memory mode to guide its behaviors. As its name shows, the memory module stores the whole chatting history of the particular agent. To edit and compile its contents and update the memory in time, we use the Environment module to guide its behavior.
Attributes & Examples:
Basic codes of an Environment module are as follows:
class Environment:
def __init__(self, config) -> None:
self.shared_memory = {"long_term_memory": [], "short_term_memory": None}
self.agents = None
self.summary_system_prompt = {}
self.summary_last_prompt = {}
self.environment_prompt = {}
self.environment_type = config["environment_type"] if "environment_type" in config else "cooperate"
self.current_chat_history_idx = 0
self.LLMs = {}
# Initialize the summary method for each state
for state_name, state_dict in config["states"].items():
if state_name != "end_state":
self.summary_system_prompt[state_name] = (
state_dict["summary_system_prompt"]
if "summary_system_prompt" in state_dict
else eval(Default_environment_summary_system_prompt)
)
self.summary_last_prompt[state_name] = (
state_dict["summary_last_prompt"]
if "summary_last_prompt" in state_dict
else eval(Default_environment_summary_last_prompt)
)
self.environment_prompt[state_name] = (
state_dict["environment_prompt"]
if "environment_prompt" in state_dict
else " "
)
LLM_type = (
state_dict["LLM_type"] if "LLM_type" in state_dict else "OpenAI"
)
if LLM_type == "OpenAI":
if "LLM" in state_dict:
self.LLMs[state_name] = OpenAILLM(**state_dict["LLM"])
else:
self.LLMs[state_name] = OpenAILLM(model="gpt-3.5-turbo-16k-0613", temperature=0.3,
log_path=f"logs/{state_name}")
self.roles_to_names = None
self.names_to_roles = None
Part of the attributes are shown below:
- LLM: As is aforementioned, our autonomous agents are based on LLM. This attribute receives the tag of a certain type of LLM and invokes it.
Methods:
summary:
The summary method receives the current chatting history, and then summarizes the situation in the current environment every once in a while.
def summary(self, current_state):
"""
Summarize the situation in the current environment every once in a while
"""
MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])
current_state_name = current_state.name
query = self.shared_memory["long_term_memory"][-1].content
relevant_history = get_relevant_history(
query,
self.shared_memory["long_term_memory"][:-1],
self.shared_memory["chat_embeddings"][:-1],
)
relevant_history = Memory.get_chat_history(relevant_history)
chat_history = Memory.get_chat_history(
self.shared_memory["long_term_memory"][-MAX_CHAT_HISTORY + 1 :]
)
summary = self.shared_memory["short_term_memory"]
# system prompt = environment prompt + current memory + system prompt
# current_memory = summary + chat history + relevant history
current_memory = eval(Environment_summary_memory)
environment_prompt = self.environment_prompt[current_state_name]
summary_system_prompt = self.summary_system_prompt[current_state_name]
environment_summary_system_prompt = eval(Environment_summary_system_prompt)
response = self.LLMs[current_state_name].get_response(None, environment_summary_system_prompt, stream=False)
return response
update_memory:
The update_memory method updates memory immediately, enabling the agent to adjust to current circumstance.
def update_memory(self, memory, current_state):
"""
update chat embbedings and long term memory,short term memory,agents long term memory
"""
MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])
self.shared_memory["long_term_memory"].append(memory)
current_embedding = get_embedding(memory.content)
if "chat_embeddings" not in self.shared_memory:
self.shared_memory["chat_embeddings"] = current_embedding
else:
self.shared_memory["chat_embeddings"] = torch.cat(
[self.shared_memory["chat_embeddings"], current_embedding], dim=0
)
if len(self.shared_memory["long_term_memory"]) % MAX_CHAT_HISTORY == 0:
summary = self.summary(current_state)
self.shared_memory["short_term_memory"] = summary
self.agents[memory.send_name].update_memory(memory)
_observe:
The _observe method helps the agent obtain the memories it needs to reply from the environment, including related memories and new memories.
def _observe(self, agent):
MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])
current_state = agent.current_state
current_role = agent.state_roles[current_state.name]
current_component_dict = current_state.components[current_role]
# cooperative: Sharing information between different states ; competitive: No information is shared between different states
current_chat_history_idx = self.current_chat_history_idx if self.environment_type == "competitive" else 0
current_long_term_memory = self.shared_memory["long_term_memory"][current_chat_history_idx:]
current_chat_embeddings = self.shared_memory["chat_embeddings"][current_chat_history_idx:]
# relevant_memory
query = current_long_term_memory[-1].content
relevant_memory = get_relevant_history(
query,
current_long_term_memory[:-1],
current_chat_embeddings[:-1],
)
relevant_memory = Memory.get_chat_history(relevant_memory, agent.name)
relevant_memory = eval(Agent_observe_relevant_memory)
agent.relevant_memory = relevant_memory
# get chat history from new conversation
conversations = self._get_agent_new_memory(agent, current_long_term_memory)
# memory = relevant_memory + summary + history + query
query = current_long_term_memory[-1]
current_memory = eval(Agent_observe_memory)
return {"role": "user", "content": current_memory}
Action
Definition:
The basic unit for each Agent to interact
Attributes & Examples:
Basic codes of an Action module are as follows:
class Action:
"""
The basic action unit of agent
"""
def __init__(self, **kwargs):
self.response = None
self.is_user = False
self.res_dict = {}
self.name = ""
self.role = ""
for key, value in kwargs.items():
setattr(self, key, value)
Methods:
process:
The current action will be processed, and the response required by the user will be obtained.
def process(self):
"""
processing action
Return: memory(Memory)
"""
response = self.response
send_name = self.name
send_role = self.role
all = ""
for res in response:
all += res
parse = f"{send_name}:"
# The third person in the dialogue was deleted.
while parse in all:
index = all.index(parse) + len(parse)
all = all[index:]
if not self.is_user:
print(f"{send_name}({send_role}):{all}")
memory = Memory(send_role, send_name, all)
return memory