agentsociety.agent.agent

Module Contents

Classes

CitizenAgentBase	Represents a citizen agent within the simulation environment.
InstitutionAgentBase	Represents an institution agent within the simulation environment.
FirmAgentBase	Represents a firm agent within the simulation environment.
BankAgentBase	Represents a bank agent within the simulation environment.
NBSAgentBase	Represents a National Bureau of Statistics agent within the simulation environment.
GovernmentAgentBase	Represents a government agent within the simulation environment.
SupervisorBase
IndividualAgentBase

Data

__all__

API

agentsociety.agent.agent.__all__

[‘CitizenAgentBase’, ‘FirmAgentBase’, ‘BankAgentBase’, ‘NBSAgentBase’, ‘GovernmentAgentBase’, ‘Super…

class agentsociety.agent.agent.CitizenAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent_base.Agent

Represents a citizen agent within the simulation environment.

• Description:

  • This class extends the base Agent class and is designed to simulate the behavior of a city resident.

  • It includes initialization of various clients (like LLM, economy) and services required for the agent’s operation.

  • Provides methods for binding the agent to the simulator and economy system, as well as handling specific types of messages.

Initialization

Initialize a new instance of the CitizenAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the CitizenAgent with the provided parameters and sets up necessary internal states.

async init()

Initialize the agent.

• Description:

  • Calls the _bind_to_simulator method to establish the agent within the simulation environment.

  • Calls the _bind_to_economy method to integrate the agent into the economy simulator.

async _bind_to_simulator()

Bind the agent to the Traffic Simulator.

• Description:

  • If the simulator is set, this method binds the agent by creating a person entity in the simulator based on the agent’s attributes.

  • Updates the agent’s status with the newly created person ID from the simulator.

  • Logs the successful binding to the person entity added to the simulator.

async _bind_to_economy()

Bind the agent to the Economy Simulator.

async update_motion()

Update the motion of the agent. Usually used in the starting of the forward method.

async do_survey(survey: agentsociety.survey.models.Survey) → str

Generate a response to a user survey based on the agent’s memory and current state.

• Args:

  • survey (Survey): The survey that needs to be answered.

• Returns:

  • str: The generated response from the agent.

• Description:

  • Prepares a prompt for the Language Model (LLM) based on the provided survey.

  • Constructs a dialog including system prompts, relevant memory context, and the survey question itself.

  • Uses the LLM client to generate a response asynchronously.

  • If the LLM client is not available, it returns a default message indicating unavailability.

  • This method can be overridden by subclasses to customize survey response generation.

async _handle_survey_with_storage(survey: agentsociety.survey.models.Survey, survey_day: Optional[int] = None, survey_t: Optional[float] = None, is_pending_survey: bool = False, pending_survey_id: Optional[int] = None) → str

Process a survey by generating a response and recording it in Database.

• Args:

  • survey (Survey): The survey data that includes an ID and other relevant information.

  • survey_day (Optional[int]): The day of the survey.

  • survey_t (Optional[float]): The time of the survey.

  • is_pending_survey (bool): Whether the survey is a pending survey.

  • pending_survey_id (Optional[int]): The ID of the pending survey.

• Description:

  • Generates a survey response using generate_user_survey_response.

  • Records the response with metadata (such as timestamp, survey ID, etc.) in Database.

  • Sends a message through the Messager indicating user feedback has been processed.

  • Handles asynchronous tasks and ensures thread-safe operations when writing to PostgreSQL.

async do_interview(question: str) → str

Generate a response to a user’s chat question based on the agent’s memory and current state.

• Args:

  • question (str): The question that needs to be answered.

• Returns:

  • str: The generated response from the agent.

• Description:

  • Prepares a prompt for the Language Model (LLM) with a system prompt to guide the response style.

  • Constructs a dialog including relevant memory context and the user’s question.

  • Uses the LLM client to generate a concise and clear response asynchronously.

  • If the LLM client is not available, it returns a default message indicating unavailability.

  • This method can be overridden by subclasses to customize chat response generation.

async _handle_interview_with_storage(message: agentsociety.message.Message) → str

Process an interview interaction by generating a response and recording it in Database.

• Args:

  • question (str): The interview data containing the content of the user’s message.

async save_agent_thought(thought: str)

Save the agent’s thought to the memory.

• Args:

  • thought (str): The thought data to be saved.

• Description:

  • Saves the thought data to the memory.

async do_chat(message: agentsociety.message.Message) → str

Process a chat message received from another agent and record it.

• Args:

  • message (Message): The chat message data received from another agent.

async _handle_agent_chat_with_storage(message: agentsociety.message.Message)

Process a chat message received from another agent and record it.

• Args:

  • payload (dict): The chat message data received from another agent.

• Description:

  • Logs the incoming chat message from another agent.

  • Prepares the chat message for storage in Database.

  • Writes the chat message and metadata into Database.

async get_aoi_info()

Get the surrounding environment information - aoi information

async get_nowtime()

Get the current time

async before_forward()

Before forward.

class agentsociety.agent.agent.InstitutionAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent_base.Agent

Represents an institution agent within the simulation environment.

• Description:

  • This class extends the base Agent class and is designed to simulate the behavior of an institution, such as a bank, government body, or corporation.

  • It includes initialization of various clients (like LLM, economy) and services required for the agent’s operation.

  • Provides methods for binding the agent to the economy system and handling specific types of messages, like gathering information from other agents.

Initialization

Initialize a new instance of the InstitutionAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the InstitutionAgent with the provided parameters and sets up necessary internal states.

async init()

Initialize the agent.

• Description:

  • Calls the _bind_to_economy method to integrate the agent into the economy simulator.

async _bind_to_economy()

Bind the agent to the Economy Simulator.

• Description:

  • Calls the _bind_to_economy method to integrate the agent into the economy system.

  • Note that this method does not bind the agent to the simulator itself; it only handles the economy integration.

async react_to_intervention(intervention_message: str)

React to an intervention.

• Args:

  • intervention_message (str): The message of the intervention.

• Description:

  • React to an intervention.

class agentsociety.agent.agent.FirmAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent.InstitutionAgentBase

Represents a firm agent within the simulation environment.

Initialization

Initialize a new instance of the InstitutionAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the InstitutionAgent with the provided parameters and sets up necessary internal states.

class agentsociety.agent.agent.BankAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent.InstitutionAgentBase

Represents a bank agent within the simulation environment.

Initialization

Initialize a new instance of the InstitutionAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the InstitutionAgent with the provided parameters and sets up necessary internal states.

class agentsociety.agent.agent.NBSAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent.InstitutionAgentBase

Represents a National Bureau of Statistics agent within the simulation environment.

Initialization

Initialize a new instance of the InstitutionAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the InstitutionAgent with the provided parameters and sets up necessary internal states.

class agentsociety.agent.agent.GovernmentAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent.InstitutionAgentBase

Represents a government agent within the simulation environment.

Initialization

Initialize a new instance of the InstitutionAgent.

• Args:

  • id (int): The ID of the agent.

  • name (str): The name or identifier of the agent.

  • toolbox (AgentToolbox): The toolbox of the agent.

  • memory (Memory): The memory of the agent.

  • agent_params (Optional[Any]): Additional parameters for the agent. Defaults to None.

  • blocks (Optional[list[Block]]): List of blocks for the agent. Defaults to None.

• Description:

  • Initializes the InstitutionAgent with the provided parameters and sets up necessary internal states.

class agentsociety.agent.agent.SupervisorBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent_base.Agent

abstract async forward(current_round_messages: list[agentsociety.message.Message]) → tuple[dict[agentsociety.message.Message, bool], list[agentsociety.message.Message]]

Process and validate messages from the current round, performing validation and intervention

• Args:

  • current_round_messages (list[Message]): List of messages for the current round, each element is a tuple of (sender_id, receiver_id, content).

• Returns:

  • tuple[dict[Message, bool], list[Message]]: A tuple containing:

    • validation_dict: Dictionary of message validation results, key is message tuple, value is whether validation passed.

    • persuasion_messages: List of persuasion messages.

class agentsociety.agent.agent.IndividualAgentBase(id: int, name: str, toolbox: agentsociety.agent.agent_base.AgentToolbox, memory: agentsociety.memory.Memory, agent_params: Optional[Any] = None, blocks: Optional[list[agentsociety.agent.block.Block]] = None)

Bases: agentsociety.agent.agent_base.Agent

async run(task: agentsociety.taskloader.Task) → Any

Unified entry point for executing the agent’s logic.

• Description:

  • It calls the forward method to execute the agent’s behavior logic.

  • Acts as the main control flow for the agent, coordinating when and how the agent performs its actions.

abstract async forward(task_context: dict[str, Any]) → Any

Process and validate messages from the current round, performing validation and intervention. The task context is a dictionary of the context of the task.

• Args:

  • task_context (dict[str, Any]): The context of the task.

• Returns:

  • Any: The result of the task.