socioverse.abc — the interfaces

A study is complete when it implements the first four; Core implements the rest.

EnvironmentProvider

socioverse.abc.environment

EnvironmentProvider ABC — the dynamic environment E.

Implementations own the world state and expose the two time channels

• advance_to(t): SCHEDULED/exogenous change (numeric events + information broadcasts)
• apply(actions): ENDOGENOUS feedback E_{t+1} = f(E_t, B_t)

and the per-agent view assembly observe_batch() that materializes the 4 quadrants.

EnvironmentProvider

Bases: ABC

reset abstractmethod

reset(seed: int) -> None

Build initial state for t=0 (load geo/info layers, adjacency, dynamic state).

Source code in socioverse/abc/environment.py

@abstractmethod
def reset(self, seed: int) -> None:
    """Build initial state for t=0 (load geo/info layers, adjacency, dynamic state)."""

advance_to abstractmethod

advance_to(t: int) -> list[ScheduledEvent]

Apply scheduled numeric events + activate information broadcasts whose at_step <= t and that haven't fired. Returns the events that fired (for logging).

Source code in socioverse/abc/environment.py

@abstractmethod
def advance_to(self, t: int) -> list[ScheduledEvent]:
    """Apply scheduled numeric events + activate information broadcasts whose
    at_step <= t and that haven't fired. Returns the events that fired (for logging)."""

apply abstractmethod

apply(actions: list[Action]) -> None

Fold agents' behavior back into env state (endogenous feedback).

Source code in socioverse/abc/environment.py

@abstractmethod
def apply(self, actions: list[Action]) -> None:
    """Fold agents' behavior back into env state (endogenous feedback)."""

observe_batch abstractmethod

observe_batch(
    agent_ids: list[str], t: int, round_idx: int = 0
) -> list[Observation]

Assemble the 4-quadrant Observation per agent. Macro layers are shared verbatim; local layers are filtered by each agent's position/network.

Source code in socioverse/abc/environment.py

@abstractmethod
def observe_batch(
    self, agent_ids: list[str], t: int, round_idx: int = 0
) -> list[Observation]:
    """Assemble the 4-quadrant Observation per agent. Macro layers are shared
    verbatim; local layers are filtered by each agent's position/network."""

agent_state abstractmethod

agent_state(agent_id: str) -> dict

Current per-agent state for the panel store (tract_id, satisfaction, ...).

Source code in socioverse/abc/environment.py

@abstractmethod
def agent_state(self, agent_id: str) -> dict:
    """Current per-agent state for the panel store (tract_id, satisfaction, ...)."""

snapshot

snapshot() -> dict

Optional global env state for env-scale metrics. Default empty.

Source code in socioverse/abc/environment.py

def snapshot(self) -> dict:
    """Optional global env state for env-scale metrics. Default empty."""
    return {}

PopulationProvider

socioverse.abc.population

PopulationProvider ABC — the fixed population pool P with persistent ids.

PopulationProvider

Bases: ABC

build abstractmethod

build(seed: int) -> list[Persona]

Materialize the persona pool with PERSISTENT, deterministic agent_ids.

Source code in socioverse/abc/population.py

@abstractmethod
def build(self, seed: int) -> list[Persona]:
    """Materialize the persona pool with PERSISTENT, deterministic agent_ids."""

neighbors

neighbors(agent_id: str) -> list[str]

Agent_ids reachable for local-information propagation. Default: none.

Source code in socioverse/abc/population.py

def neighbors(self, agent_id: str) -> list[str]:
    """Agent_ids reachable for local-information propagation. Default: none."""
    return []

DecisionModel

socioverse.abc.decision

DecisionModel ABC — computes behavior B_t.

decide_batch is the primary method: the engine hands it the WHOLE step (every agent's Observation) at once and expects one Action per agent back. "Batch" describes that interface, not how the LLM is called; the implementation decides the call pattern:

• From-scratch studies (sv-build-model) call the LLM once per agent, sending the step's per-agent calls concurrently on a bounded thread pool, and parse / fall back per agent (see studies/opinion_diffusion/model.py). Do not ask for one JSON reply covering the whole population: long structured replies get truncated and one bad reply fails the step.
• Legacy adapters keep the wrapped engine's own call pattern (e.g. the Chicago model's archetype-grouped calls in its two LLM phases) so its validated dynamics are unchanged.

DecisionModel

Bases: ABC

decide_batch abstractmethod

decide_batch(
    obs: list[Observation], memories: dict[str, Any]
) -> list[Action]

Compute B_t for every agent of the step; return one Action per agent. memories maps agent_id -> AgentMemory (per-agent rolling history) for genuinely longitudinal behavior. How the LLM is called inside is up to the model (see the module docstring).

Source code in socioverse/abc/decision.py

@abstractmethod
def decide_batch(
    self, obs: list[Observation], memories: dict[str, Any]
) -> list[Action]:
    """Compute B_t for every agent of the step; return one Action per agent. `memories`
    maps agent_id -> AgentMemory (per-agent rolling history) for genuinely longitudinal
    behavior. How the LLM is called inside is up to the model (see the module docstring)."""

MetricCollector & TrajectoryStore

socioverse.abc.trajectory

TrajectoryStore + MetricCollector ABCs (the longitudinal panel + metrics).

TrajectoryStore

Bases: ABC

Database-like store for panel rows, metrics, events, and messages.

Implemented over DuckDB (default) so the same file supports later real-time interactive querying via SQL.

record abstractmethod

record(records: list[TrajectoryRecord]) -> None

Append panel rows (one per agent per step).

Source code in socioverse/abc/trajectory.py

@abstractmethod
def record(self, records: list[TrajectoryRecord]) -> None:
    """Append panel rows (one per agent per step)."""

record_metrics abstractmethod

record_metrics(row: dict[str, Any]) -> None

Append one per-step aggregate metrics row (must contain 'step').

Source code in socioverse/abc/trajectory.py

@abstractmethod
def record_metrics(self, row: dict[str, Any]) -> None:
    """Append one per-step aggregate metrics row (must contain 'step')."""

record_events

record_events(step: int, notes: list[str]) -> None

Optional: log fired scheduled events / activated broadcasts at a step.

Source code in socioverse/abc/trajectory.py

def record_events(self, step: int, notes: list[str]) -> None:
    """Optional: log fired scheduled events / activated broadcasts at a step."""

record_messages

record_messages(messages: list[dict[str, Any]]) -> None

Optional: log inter-agent messages (scenario 2).

Source code in socioverse/abc/trajectory.py

def record_messages(self, messages: list[dict[str, Any]]) -> None:
    """Optional: log inter-agent messages (scenario 2)."""

finalize abstractmethod

finalize() -> None

Flush, export parquet, close.

Source code in socioverse/abc/trajectory.py

@abstractmethod
def finalize(self) -> None:
    """Flush, export parquet, close."""

MetricCollector

Bases: ABC

collect abstractmethod

collect(
    env: Any, actions: list[Action], t: int
) -> dict[str, Any]

Return a flat per-step metrics dict (the collector sets/overrides 'step').

Source code in socioverse/abc/trajectory.py

@abstractmethod
def collect(self, env: Any, actions: list[Action], t: int) -> dict[str, Any]:
    """Return a flat per-step metrics dict (the collector sets/overrides 'step')."""

columns

columns() -> list[str]

Declared metric column names (contract for the reporter). Optional.

Source code in socioverse/abc/trajectory.py

def columns(self) -> list[str]:
    """Declared metric column names (contract for the reporter). Optional."""
    return []

Simulator

socioverse.abc.simulator

Simulator ABC — the longitudinal driver.

Simulator

Bases: ABC

run abstractmethod

run() -> MetricsHistory

Execute the E_t -> B_t -> E_{t+1} loop over the study horizon and return the per-step metrics history. Side effect: writes the panel/metrics store.

Source code in socioverse/abc/simulator.py

@abstractmethod
def run(self) -> MetricsHistory:
    """Execute the E_t -> B_t -> E_{t+1} loop over the study horizon and return
    the per-step metrics history. Side effect: writes the panel/metrics store."""

MessageBus

socioverse.abc.messaging

MessageBus ABC — the inter-agent communication medium (scenario 2, HiSim).

Agent communication is mediated, NOT O(N^2) direct calls: a post/reply action lands on the bus (a local-information channel); the next observation delivers it to the author's network neighbours. A simple in-memory implementation lives in engine/messaging.py; large-scale backends (DuckDB/Redis) can subclass this.

MessageBus

Bases: ABC

post abstractmethod

post(message: dict[str, Any]) -> None

Publish a message. Expected keys: author_id, content, step, channel, audience?

Source code in socioverse/abc/messaging.py

@abstractmethod
def post(self, message: dict[str, Any]) -> None:
    """Publish a message. Expected keys: author_id, content, step, channel, audience?"""

fetch abstractmethod

fetch(
    recipient_id: str, t: int, neighbors: list[str]
) -> list[dict[str, Any]]

Return messages visible to recipient_id at step t (from its neighbours).

Source code in socioverse/abc/messaging.py

@abstractmethod
def fetch(self, recipient_id: str, t: int, neighbors: list[str]) -> list[dict[str, Any]]:
    """Return messages visible to `recipient_id` at step t (from its neighbours)."""

drain

drain() -> list[dict[str, Any]]

Return + clear messages posted this round (for persistence). Optional.

Source code in socioverse/abc/messaging.py

def drain(self) -> list[dict[str, Any]]:
    """Return + clear messages posted this round (for persistence). Optional."""
    return []

Reporter

socioverse.abc.reporter

Reporter ABC — turns the trajectory store into figures + report.md.

Reporter

Bases: ABC

render abstractmethod

render(
    study: StudySpec,
    metrics: MetricsHistory,
    store_path: Path,
    out_dir: Path,
) -> None

Read the (DuckDB) store + metrics and write figures + report.md to out_dir.

Source code in socioverse/abc/reporter.py

@abstractmethod
def render(
    self,
    study: StudySpec,
    metrics: MetricsHistory,
    store_path: Path,
    out_dir: Path,
) -> None:
    """Read the (DuckDB) store + metrics and write figures + report.md to out_dir."""