🤖AI Autonomous Agents

Brief Introduction

🤖 AI Autonomous Agents are intelligent entities capable of perceiving their environment, making decisions, and taking actions without direct human intervention. As human society rapidly develops, AI Autonomous Agents find applications in diverse circumstances.

Demonstrations & Attributes

Basic Codes

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

# Remark:
# state_roles(dict): The agent’s role in different state.
# is_user: True if the agent is a user operation, False otherwise.
# long_term_memory: The Agent's historical conversation records, in which the conversations of other agents are informed in the form of historical records.
# short_term_memory: The Agent’s short-term memory is a summary of its past historical memory.

Methods

From_config

@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_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 " "

            LLM_type = (
                current_state["agent_states"][agent_role]["LLM_type"]
                if "LLM_type" in current_state["agent_states"][agent_role]
                else "OpenAI"
            )
            if LLM_type == "OpenAI":
                if "LLM" in current_state["agent_states"][agent_role]:
                    agent_LLMs[state_name] = OpenAILLM(
                        **current_state["agent_states"][agent_role]["LLM"]
                    )
                else:
                    agent_LLMs[state_name] = OpenAILLM(model="gpt-3.5-turbo-16k-0613", temperature=0.3, log_path=f"logs/{agent_name}")
        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

# Remark:
# The from_config method starts the agent according to the given attributes and data.

Act

def act(self):
    """
    return actions by the current state
    """
    current_state = self.current_state
    system_prompt, last_prompt, res_dict = self.compile()
    chat_history = self.long_term_memory

    current_LLM = self.LLMs[current_state.name]

    response = current_LLM.get_response(
        chat_history, system_prompt, last_prompt, stream=True
    )
    return response, res_dict

# Remark:
# The act method generates and outputs the response of the Agent. Detailed explanations on particular attributes will be shown afterwards.

Step

def step(self, current_state, environment, input):
    """
    return actions by current state and environment
    """
    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

    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(environment)
            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)

# Remark:
# Closely related to the act method, the step method updates the current circumstance and then returns the response of an Agent. Detailed explanations on particular attributes will be shown afterwards.

Compile

def compile(self):
    """
    get prompt from state depend on your role
    Return:
    system_prompt: system_prompt for agent's LLM
    last_prompt: last_prompt for agent'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
    last_prompt = eval(Agent_last_prompt)
    return system_prompt, last_prompt, res_dict

# Remark:
# The Compile method reaches for the current role and returns the action of a certain agent state.

Observe

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)

# Remark:
# The Observe method is the core method of an agent. It updates and reads the current environment, including the chatting history and the basic information, and returns particular actions for the agent.

Update_memory

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

# Remark:
# The update_memory method is responsible for updating its long-term memory and short-term memory according to the environment. Every time it is more than a certain number of rounds since the last speech of the agent, it will be summarized to obtain short-term memory.

Examples

🌐 We provide various types of Agents in our QuickStart part. You can also prepare your OWN Agent in a customized style! 🚀