agentsociety.agent.agent_base

Module Contents

Classes

AgentParams	Agent parameters
GatherQuery	A model for gather query
AgentType	Agent Type
Agent	Agent base class

Functions

extract_json

Extract JSON substring from a raw string response.

Data

__all__

API

agentsociety.agent.agent_base.__all__

[‘Agent’, ‘AgentType’, ‘AgentParams’, ‘GatherQuery’]

class agentsociety.agent.agent_base.AgentParams

Bases: pydantic.BaseModel

Agent parameters

class agentsociety.agent.agent_base.GatherQuery

Bases: pydantic.BaseModel

A model for gather query

key: str

None

target_agent_ids: list[int]

None

flatten: bool

True

keep_id: bool

True

class agentsociety.agent.agent_base.AgentType(*args, **kwds)

Bases: enum.Enum

Agent Type

• Citizen, Citizen type agent

• Institution, Organization or institution type agent

Initialization

Unspecified

‘Unspecified’

Citizen

‘Citizen’

Institution

‘Institution’

Supervisor

‘Supervisor’

Individual

‘Individual’

agentsociety.agent.agent_base.extract_json(output_str)

Extract JSON substring from a raw string response.

Args: output_str: Raw string output that may contain JSON data.

Returns: Extracted JSON string if valid, otherwise None.

Note: Searches for the first ‘{’ and last ‘}’ to isolate JSON content. Catches JSON decoding errors and logs warnings.

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

Bases: abc.ABC

Agent base class

Initialization

Initialize the Agent.

• Args:

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

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

  • type (AgentType): The type of the agent. Defaults to AgentType.Unspecified.

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

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

ParamsType: type[agentsociety.agent.agent_base.AgentParams]

None

Context: type[agentsociety.agent.context.AgentContext]

None

BlockOutputType: type[agentsociety.agent.block.BlockOutput]

None

StatusAttributes: list[agentsociety.agent.memory_config_generator.MemoryAttribute]

[]

description: str = <Multiline-String>

classmethod default_params()

classmethod default_context()

classmethod __init_subclass__(**kwargs)

async init()

__getstate__()

property id

The Agent’s Simulator ID

property toolbox

The Agent’s Toolbox

property llm

The Agent’s LLM

property environment

The Agent’s Environment

property messager

The Agent’s Messager

property database_writer

The Agent’s Database Writer

property memory

The Agent’s Memory

property status

The Agent’s Status Memory

property stream

The Agent’s Stream Memory

abstract async reset()

Reset the agent.

abstract 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.

async send_message_to_agent(to_agent_id: int, content: str, type: str = 'social')

Send a social or economy message to another agent.

• Args:

  • to_agent_id (int): The ID of the recipient agent.

  • content (str): The content of the message to send.

  • type (str, optional): The type of the message (“social” or “economy”). Defaults to “social”.

• Raises:

  • RuntimeError: If the Messager is not set.

• Description:

  • Validates the message type and logs a warning if it’s invalid.

  • Prepares the message payload with necessary metadata such as sender ID, timestamp, etc.

  • Sends the message asynchronously using _send_message.

  • Optionally records the message in Database if it’s a “social” type message.

_get_gather_query_and_clear()

register_gather_query(key: str, target_agent_ids: list[int], flatten: bool = True, keep_id: bool = True)

get_gather_results(key: str) → Optional[list[Any] | dict[int, Any]]

async register_aoi_message(target_aoi: Union[int, list[int]], content: str)

Register a message to target aoi

• Args:

  • target_aoi (Union[int, list[int]]): The ID of the target aoi.

  • content (str): The content of the message to send.

• Description:

  • Register a message to target aoi.

async cancel_aoi_message(target_aoi: Union[int, list[int]])

Cancel a message to target aoi

abstract async forward() → Any

Define the behavior logic of the agent.

• Raises:

  • NotImplementedError: As this method must be implemented by subclasses.

• Description:

  • This abstract method should contain the core logic for what the agent does at each step of its operation.

  • It is intended to be overridden by subclasses to define specific behaviors.

async status_summary()

Status summary

async close()

Execute when the agent is deleted or the simulation is finished.

async before_forward()

Before forward

async after_forward()

After forward

async before_blocks()

Before blocks

async after_blocks()

After blocks

async run() → 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.

_is_output_type_compatible(block_output_type) → bool

Check if the block output type is compatible with the agent’s expected output type.

• Description:

  • Checks compatibility between block output type and agent’s expected output type.

  • Supports both class inheritance and field compatibility checks.

• Args:

  • block_output_type: The output type class of the block.

• Returns:

  • bool: True if compatible, False otherwise.