Interacting with AgentSociety

This guide introduces how to interact with AgentSociety 2 during experiments.

Overview

AgentSociety 2 provides two main interaction modes:

1. Query Mode (read-only): Ask questions without modifying simulation state

2. Intervention Mode (read-write): Modify agent state or environment variables

These interactions can be performed at:

• During simulation: Between steps or at specific time points

• After simulation: Query final state or collect survey data

Interaction Mode Comparison

Basic Interaction Patterns

ask() Method - Read-only Queries

Use society.ask() for read-only queries that don’t modify the simulation:

# Query about agent state
response = await society.ask("What is Agent 1's current mood?")

# Query about environment
response = await society.ask("What is the current weather?")

# Query about multiple agents
response = await society.ask("List all agents who are unhappy")

intervene() Method - Read-Write Modifications

Use society.intervene() to make changes to the simulation:

# Send a message to agents
result = await society.intervene(
    "Send a message to all agents: 'Severe weather coming, go home!'"
)

# Modify environment variables
result = await society.intervene(
    "Change the weather to rainy and temperature to 15°C"
)

# Modify agent states
result = await society.intervene(
    "Set all agents' happiness to 0.8"
)

Simulation Workflow

Running with Step Control

from datetime import datetime

# Create and initialize society（agent_specs + agent_class_name + env_router）
society = AgentSociety(
    agent_specs=agent_specs,
    agent_class_name="PersonAgent",
    env_router=env_router,
    start_t=datetime.now(),
    run_dir=__import__("pathlib").Path("run"),
)
await society.init()

# Run for specific number of steps
for step_num in range(10):
    # Query before step
    state = await society.ask("What's happening?")
    print(f"Step {step_num}: {state}")

    # 执行一步（tick 为本步时长，秒）；当前仿真时间由编排器内部维护
    await society.step(tick=3600)

    # Intervene based on conditions
    if "emergency" in state.lower():
        await society.intervene("Broadcast emergency alert")

await society.close()

Data Collection

Collecting Agent Responses

# Collect responses from all agents（按 spec 的 id 提问）
for spec in agent_specs:
    response = await society.ask(
        f"Agent {spec['id']}, how do you feel about the current situation?"
    )
    print(f"Agent {spec['id']}: {response}")
    # Store for analysis

# Collect survey responses
survey_questions = [
    "How satisfied are you with your current situation? (1-5)",
    "What would improve your quality of life?",
]

for spec in agent_specs:
    for question in survey_questions:
        answer = await society.ask(f"Agent {spec['id']}: {question}")
        # Save answer to database or file

Data Collection with ReplayWriter

from pathlib import Path

# replay 默认开启，写入 run_dir/replay/；环境 replay dataset 自动记录
society = AgentSociety(
    agent_specs=agent_specs,
    agent_class_name="PersonAgent",
    env_router=env_router,
    start_t=datetime.now(),
    run_dir=Path("run"),
    enable_replay=True,
)
await society.init()

# Run simulation - environment replay datasets are recorded to sharded JSONL
await society.run(num_steps=10, tick=3600)

# Query replay catalog or environment tables later with ReplayReader
# ReplayReader("run/replay").load_dataset_catalog()

await society.close()

Common Interaction Scenarios

Scenario 1: Event Intervention

# Normal simulation
await society.run(num_steps=5, tick=3600)

# Event occurs (e.g., hurricane)
await society.intervene(
    "Broadcast: 'Hurricane warning! Seek shelter immediately!'"
)

# Continue simulation to observe reactions
await society.run(num_steps=5, tick=3600)

# Collect impact data
impact = await society.ask("How did the hurricane affect everyone?")
print(impact)

Scenario 2: Policy Experiments

# Control group - no policy（以 spec 声明，不实例化 agent）
control_specs = [{"id": i, "profile": {...}, "config": {}} for i in range(1, 11)]
control_society = AgentSociety(agent_specs=control_specs, agent_class_name="PersonAgent", ...)
await control_society.init()
await control_society.run(num_steps=10, tick=3600)

# Treatment group - with policy intervention
treatment_specs = [{"id": i + 10, "profile": {...}, "config": {}} for i in range(10)]
treatment_society = AgentSociety(agent_specs=treatment_specs, agent_class_name="PersonAgent", ...)
await treatment_society.init()

# Implement policy
await treatment_society.intervene(
    "Implement UBI policy: everyone receives $1000 monthly"
)

await treatment_society.run(num_steps=10, tick=3600)

# Compare outcomes
control_outcome = await control_society.ask("What's the average happiness?")
treatment_outcome = await treatment_society.ask("What's the average happiness?")

Scenario 3: Data Collection at Multiple Time Points

# Baseline data
baseline = await society.ask("Record everyone's baseline mood")

# Intervention
await society.intervene("Announce new community program")

# Short-term effects
await society.run(num_steps=3, tick=3600)
short_term = await society.ask("How is everyone feeling now?")

# Long-term effects
await society.run(num_steps=10, tick=3600)
long_term = await society.ask("How is everyone feeling now?")

# Analyze change over time

Best Practices

1. Use ask() for queries: Always use ask() when you only need information

2. Use intervene() for changes: Only use intervene() when you want to modify state

3. Use ReplayWriter: analyze experiments with environment replay datasets; for local agent debugging, inspect workspace files under run/agents/agent_xxxx/

4. Query specific agents: Ask specific agents for targeted responses

5. Intervene at appropriate times: Intervene at appropriate simulation times for realistic effects