Environment Modules

This module provides the base class for environment routing and environment modules.

RouterBase

class agentsociety2.env.router_base.RouterBase(env_modules, max_steps=10, max_llm_call_retry=10, replay_writer=None, llm_clients_spec=None)

Bases: ABC

Environment router abstract base class.

Router is used to coordinate the interaction between the agent and the environment module (EnvBase). Its responsibilities include:

• Aggregate and filter environment tools (function call schema)

• Receive the simulation clock pushed by the orchestrator (set_current_time()) so env modules can read it during the agent phase.

• Provide unified LLM call encapsulation (retry and token statistics)

• Generate world description (optional, used as the prompt for PersonAgent)

__init__(env_modules, max_steps=10, max_llm_call_retry=10, replay_writer=None, llm_clients_spec=None)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer (ReplayWriter | None) – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec (Dict[str, Any] | None) – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

set_trace_sink(sink)

Inject a sync sharded-writer adapter for env-side trace spans.

Parameters:

sink (Any) – An object exposing append_record(record_dict) (and optionally flush()), e.g. the adapter built by build_local_sink(). Pass None to disable tracing.

set_current_time(t)

Push the society’s current clock to the router and env modules.

This is NOT time advancement — the society advances its own clock in step() and calls this so that, during the agent phase (which runs before step()), env modules observe the correct simulation time. Several env modules read self.t inside @tool methods (e.g. event scheduling, social-media post timestamps), so the time must be set before agents start querying the environment.

abstractmethod async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

A unified entrance to interact with the environment (specific routing strategies are implemented by subclasses).

Parameters:

• ctx (dict) – Context dictionary. Template mode can contain variables for {var} replacement.

• instruction (str) – Command text. In template mode, variables will be replaced before execution.

• readonly (bool) – Whether it is read-only (when reading-only, avoid changing the environment state).

• template_mode (bool) – Whether to enable template mode.

• trace_id (str | None) – OTel trace ID (UUID hex) used to correlate tracing across agents and environments.

• parent_span_id (str | None) – OTel parent span ID used to build the span hierarchy.

Returns:

(ctx, answer), where ctx may be updated by the environment.

Return type:

Tuple[dict, str]

async init(start_datetime)

Initialize router and environment modules.

Parameters:

start_datetime (datetime) – Simulation start time.

Returns:

Whether any env module restored a checkpoint, i.e. whether this is a resume.

Return type:

bool

async step(tick, t)

Advance the environment module one simulation step.

Parameters:

• tick (int) – Time span of this step (seconds).

• t (datetime) – The simulation time after the end of this step.

async close()

Shut down the router (turn off all environment modules).

bind_env_workspaces(root, module_types)

Bind each env module to root/module_type idempotently.

module_types and env_modules correspond by position and also act as directory keys. They align with env_kwargs to avoid conflicts between multiple instances of the same class.

async from_workspaces()

Restore the state of each bound module when a snapshot exists.

A restore failure in one module does not stop other modules, but the failures are summarized at ERROR level. Otherwise the run could silently enter an inconsistent state where some modules are fresh and others came from checkpoints.

Returns:

Whether any module loaded a snapshot, i.e. whether this is a resume.

Return type:

bool

async to_workspaces()

Write each module’s dynamic state to its workspace.

Collect results with return_exceptions=True so one module’s write failure does not discard persistence for every other module in this step. Without this, a single serialization error could leave the whole step’s checkpoint behind.

get_tool_call_history()

Summarize tool call history for all environment modules (sorted by time).

Returns:

Recall the record list (in ascending order by timestamp).

Return type:

List[Dict[str, Any]]

reset_tool_call_history()

Clear the tool call history of all environment modules.

set_trace_context(trace_id, parent_span_id)

Set the current OTel trace context for all environment modules.

Router subclasses should call this method at the ask entry point so the @tool decorator can write trace_id into tool-call history.

Parameters:

• trace_id (str | None) – OTel trace ID（UUID hex）。

• parent_span_id (str | None) – OTel parent span ID。

clear_trace_context()

Clear trace context for all environment modules.

set_replay_writer(writer)

Set replay writers for all environment modules.

Parameters:

writer (ReplayWriter) – ReplayWriter Example.

get_system_prompt()

Build a general system prompt on the router side (without specific task instructions).

Returns:

system prompt text.

Return type:

str

async acompletion(model: Literal['coder', 'summary'], messages: list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage], stream: Literal[False] = False, **kwargs: Any) → ModelResponse

async acompletion(model: Literal['coder', 'summary'], messages: list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage], stream: Literal[True] = True, **kwargs: Any) → CustomStreamWrapper

Wrap LLM dispatch; the dispatcher handles concurrency control and HTTP retries.

Parameters:

• model (Literal['coder', 'summary']) – coder or summary.

• messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) – Message list (system prompt is not automatically appended; please use acompletion_with_system_prompt() if necessary).

• stream (bool) – Whether to stream returns.

• max_retries (int | None) – Optional. Maximum number of retries; if empty, use self.max_llm_call_retry.

• base_delay (float) – Exponential backoff benchmark latency for class 429 errors.

• max_delay (float) – Exponential backoff maximum delay.

Returns:

LLM response object (litellm.types.utils.ModelResponse when stream=False).

Raises:

ValueError – Thrown when the number of retries is exceeded and still fails.

async acompletion_with_system_prompt(model, messages, **kwargs)

Send a completion request and automatically append the router system prompt at the front.

Parameters:

• model (Literal['coder', 'summary']) – coder or summary.

• messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) – Message list.

Returns:

LLM response object.

async acompletion_with_pydantic_validation(model, model_type, messages, max_retries=10, base_delay=1.0, max_delay=60.0, error_feedback_prompt=None, **kwargs)

Send completion and verify to the specified Pydantic model (automatic feedback and retry on failure).

Parameters:

• model (Literal['coder', 'summary']) – coder or summary.

• model_type (Type[T]) – Pydantic model type used for validation.

• messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) – Message list (system prompt will be automatically appended).

• max_retries (int) – Maximum number of retries (excluding first attempt).

• base_delay (float) – Base exponential-backoff delay passed to the LLM dispatcher.

• max_delay (float) – Maximum exponential-backoff delay passed to the LLM dispatcher.

• error_feedback_prompt (str | None) – Optional. Customize error feedback template (need to contain {error_message} and {model_schema} placeholders).

Returns:

A Pydantic instance that passes validation.

Raises:

ValueError – Thrown when parsing/verification fails after retries.

Return type:

T

get_token_usages()

Return aggregated token usage from this router’s LLM clients.

Each dispatcher (LLMClient) tracks its own per-loop usage; we snapshot them here (without clearing). Returns {model: TokenUsageStats}.

reset_token_usages()

Reset aggregated token usage on this router’s LLM clients (drain).

static get_status_descriptions()

Get standard status descriptions.

The status indicates whether the user’s instruction has been effectively completed in the environment modules, or needs to wait for some time before the user actively checks completion.

Returns:

Dictionary mapping status values to their descriptions: - success: The task has been completed successfully. All required operations finished without errors. - in_progress: The task is still being executed or more steps are needed. The agent need to check whether it is done in the next steps. - fail: The task could not be completed (e.g., unsupported instruction, missing data, invalid input). Include detailed reason in results. - error: An error occurred during code execution. Must include error details in results[‘error’]. - unknown: Execution status unknown

Return type:

Dict[str, str]

async get_world_description()

Get the world description.

async generate_world_description_from_tools(max_retries=None)

Generate a short world description from the environment router module information.

world_description only provides lightweight background: what this simulation world roughly is and how to interact with it through ask_env. Module-level details should be handled by env skills, tool schemas, or the router itself, so the world section does not crowd out env-skill space in the prompt.

Returns:

str: The generated world description text, which can be directly used as the world_description parameter of PersonAgent

Return type:

str

get_collected_pydantic_models()

Get all collected Pydantic BaseModel types and their source code.

These models are collected from the parameter types and return value types of the tool functions when calling _collect_tools_info().

Returns:

Dictionary, key is BaseModel type, value is its source code

Return type:

Dict[Type[BaseModel], str]

async generate_final_answer(ctx, instruction, results, process_text=None, status='unknown', error=None)

Generate the final answer based on user input and Router output.

This method receives user input (ctx, instruction) and Router output (results and process text), summarizes them into an answer using LLM, and determines the execution status. Returned in JSON format and verified through Pydantic model.

Parameters:

• ctx (dict) – Context dictionary, containing information such as environment variables

• instruction (str) – User’s original instruction

• results (Dict[str, Any]) – Dictionary of results executed by Router

• process_text (str | None) – Process text (optional), text information describing the execution process

• status (str) – Preliminary execution status (may be updated by LLM)

• error (str | None) – Error message (if any)

Returns:

Tuple[str, str]: (final_answer, determined_status) - final_answer: the final answer generated by LLM (summary text) - determined_status: determined execution status (success/in_progress/fail/error)

Return type:

Tuple[str, str]

EnvBase

class agentsociety2.env.base.EnvBase

Bases: object

Environment module base class.

The environment module defines the actions and observable states that the Agent can perform. Register the method as a callable tool through the @tool decorator for Router to call.

Tool type:

• Regular tools: @tool(readonly=False) — Can modify environment state

• Read-only tools: @tool(readonly=True) — Query only, does not modify state

• Observation tools: @tool(readonly=True, kind="observe") — Automatically called

• Statistics tools: @tool(readonly=True, kind="statistics") — Statistical information

Subclasses should implement:

• Use the @tool decorator to define executable operations

• Optional: Implement the observe() method (tools that collect kind=”observe” by default)

classmethod is_concurrency_safe()

Return whether this module’s tools may be called concurrently.

Concurrency-safe modules do not mutate shared (cross-agent) state in their tools, so the router can execute their generated code without the global _execute_lock and the env Ray actor can fan out parallel ask calls. Default is False (conservative); modules that keep only per-agent state (keyed by agent_id, no shared mutable structures) override this to True.

Returns:

True if concurrent tool calls are safe.

Return type:

bool

__init__()

property name

Name of the environment module

classmethod description()

Return a short module description for router/world orientation.

classmethod init_description()

Return directories containing agent skills provided by this environment module.

Returns:

Markdown text with class name, constructor kwargs, accepted data shapes, defaults, and a minimal kwargs example when useful.

Return type:

str

classmethod skill_dirs()

Return directories containing agent skills provided by this environment module.

This method enables environment modules to bundle specialized skills that agents should use when operating within that environment. Skills are discovered by scanning the returned directories for SKILL.md files.

Default Discovery Convention:

The base implementation automatically discovers skills from:

1. Package modules (e.g., mobility_space/__init__.py): Scans <module_dir>/agent_skills/ directory.

2. Single-file modules (e.g., economy_space.py): Scans <module_dir>/<stem>_agent_skills/ directory.

Override for Custom Paths:

Subclasses can override this method to specify custom skill directories:

@classmethod
def skill_dirs(cls) -> list[Path]:
    from pathlib import Path
    skills_dir = Path(__file__).parent.parent / "skills"
    return [skills_dir] if skills_dir.is_dir() else []

Skill Directory Structure:

Each skill directory should contain subdirectories with SKILL.md files, for example agent_skills/navigation/SKILL.md and optional agent_skills/navigation/scripts/.

Returns:

List of Path objects pointing to directories containing skill subdirectories. Empty list if no skills are provided.

Return type:

list[Path]

async init(start_datetime)

Initialize the environment module.

Parameters:

start_datetime (datetime) – Simulation start time.

async step(tick, t)

Advance the environment module one simulation step.

Parameters:

• tick (int) – Time span of this step (seconds).

• t (datetime) – The simulation time after the end of this step.

Raises:

NotImplementedError – The base class does not provide a default implementation.

async close()

Closes the environment module and releases resources (optional override).

async to_workspace(workspace_path=None)

Write dynamic state to the workspace; called by society through the router on each step.

The default implementation is a no-op. Subclasses write restart-persistent state under self._workspace_root or the provided workspace_path using their chosen format and file name. Use atomic_write_text() for atomic writes.

async restore(workspace_path)

Restore dynamic state from the workspace on the resume path.

The default implementation returns False. Subclasses read and restore the state they wrote, and return whether a snapshot was loaded so the router can distinguish fresh startup from resume. This is called only after __init__ and init(), so it can safely override fields set by init().

async classmethod from_workspace(workspace_path, **init_kwargs)

Rebuild a ready module from a workspace: cls(**init_kwargs) followed by restore.

Parameters:

init_kwargs (Any) – Keyword arguments used to construct the module; during resume the actor reads them from SOCIETY.json.

get_tool_call_history()

Get tool call history (shallow copy).

Returns:

Call record list. Each entry contains function_name, kwargs, return_value, exception_occurred, exception_info, timestamp.

Return type:

list[dict[str, Any]]

reset_tool_call_history()

Clear tool call history.

set_replay_writer(writer)

Set the playback writer (for automatic table creation and writing status).

Parameters:

writer (ReplayWriter) – ReplayWriter Example.

TokenUsageStats

class agentsociety2.env.router_base.TokenUsageStats(*, call_count=0, input_tokens=0, output_tokens=0)

Token usage statistics for a model.

Variables:

• call_count – Number of API calls made.

• input_tokens – Total number of input tokens consumed.

• output_tokens – Total number of output tokens consumed.

call_count: int

input_tokens: int

output_tokens: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

CacheStats

class agentsociety2.env.router_codegen.CacheStats(request_count=0, predefined_hit_count=0, cache_hit_count=0, cache_miss_count=0, total_input_tokens=0, total_output_tokens=0, code_execution_success_count=0, code_execution_failure_count=0, total_code_retry_count=0)

Cache statistics

request_count: int = 0

predefined_hit_count: int = 0

cache_hit_count: int = 0

cache_miss_count: int = 0

total_input_tokens: int = 0

total_output_tokens: int = 0

code_execution_success_count: int = 0

code_execution_failure_count: int = 0

total_code_retry_count: int = 0

property cache_hit_rate: float

Cache hit rate

property code_execution_success_rate: float

Code execution success rate

property avg_input_tokens: float

Average input token count

property avg_output_tokens: float

Average output token count

property avg_retry_count: float

Average code retry count

__init__(request_count=0, predefined_hit_count=0, cache_hit_count=0, cache_miss_count=0, total_input_tokens=0, total_output_tokens=0, code_execution_success_count=0, code_execution_failure_count=0, total_code_retry_count=0)

Built-in router

ReActRouter

class agentsociety2.env.router_react.ReActRouter(env_modules, max_steps=10, max_llm_call_retry=10)

Bases: RouterBase

ReAct mode Router: Through the Reasoning-Acting loop, use function calling to directly call all functions of all modules.

Workflow: 1. Collect tool information of all environment modules 2. Use LLM’s function calling capability to select and call tools 3. Execute tool calls and obtain results 4. Feed back the results to LLM and continue the next Reasoning-Acting cycle 5. Until LLM determines that the task is completed or the maximum number of steps is reached

__init__(env_modules, max_steps=10, max_llm_call_retry=10)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use ReAct mode to process instructions.

Parameters:

• ctx (dict) – context dictionary

• instruction (str) – command string

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Template mode (not used by ReActRouter, only compatible with signatures)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

PlanExecuteRouter

class agentsociety2.env.router_plan_execute.PlanExecuteRouter(env_modules, max_steps=10, max_llm_call_retry=10)

Bases: RouterBase

Plan-and-Execute mode Router: first make a plan, and then execute the steps in the plan.

Workflow: 1. Collect tool information from all environment modules 2. Phase 1: Use LLM to develop an execution plan (without calling tools) 3. Phase 2: Execute tool calls step by step according to the plan 4. Generate the final answer based on the execution results

__init__(env_modules, max_steps=10, max_llm_call_retry=10)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use Plan-and-Execute mode to process instructions.

Parameters:

• ctx (dict) – context dictionary

• instruction (str) – command string

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Template mode (not used by PlanExecuteRouter, only compatible with signatures)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

CodeGenRouter

class agentsociety2.env.router_codegen.CodeGenRouter(env_modules, max_body_code_lines=15, max_steps=10, max_llm_call_retry=10, replay_writer=None, final_summary_enabled=False, code_format='raw_code', template_cache_enabled=True, template_cache_similarity_threshold=0.85, template_cache_max_size=1000, template_cache_dir=None, llm_clients_spec=None)

Bases: RouterBase

Code-generating Router: Calls the @tool-marked interface in the environment module by generating Python code.

Workflow: 1. Collect tool information of all environment modules 2. Provide environment module description and tool information (including pydantic BaseModel and module class definition) to LLM using a Python code format similar to pyi files 3. LLM generates Python code to call tools 4. Use AST parsing to check code safety 5. Execute code through compile and exec, capture printout 6. Generate final response based on execution results and printout

ALLOWED_BUILTINS: ClassVar[set[str]] = {'abs', 'all', 'any', 'bool', 'dict', 'dir', 'enumerate', 'float', 'getattr', 'hasattr', 'int', 'isinstance', 'len', 'list', 'max', 'min', 'print', 'range', 'reversed', 'round', 'set', 'sorted', 'str', 'sum', 'tuple', 'type', 'zip'}

FORBIDDEN_AST_NODES: ClassVar[set[type]] = {<class 'ast.Assert'>, <class 'ast.AsyncWith'>, <class 'ast.ClassDef'>, <class 'ast.Delete'>, <class 'ast.Global'>, <class 'ast.Nonlocal'>, <class 'ast.With'>}

ALLOWED_MODULES: ClassVar[set[str]] = {'collections', 'copy', 'datetime', 'decimal', 'fractions', 'functools', 'itertools', 'json', 'math', 'np', 'numpy', 'operator', 'random', 're', 'statistics', 'string'}

DANGEROUS_MODULES: ClassVar[set[str]] = {'__builtin__', '__builtins__', 'builtins', 'ctypes', 'ftplib', 'http', 'marshal', 'os', 'pickle', 'shutil', 'smtplib', 'socket', 'subprocess', 'sys', 'urllib'}

__init__(env_modules, max_body_code_lines=15, max_steps=10, max_llm_call_retry=10, replay_writer=None, final_summary_enabled=False, code_format='raw_code', template_cache_enabled=True, template_cache_similarity_threshold=0.85, template_cache_max_size=1000, template_cache_dir=None, llm_clients_spec=None)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer (ReplayWriter | None) – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec (Dict[str, Any] | None) – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use code generation to process instructions. Execution via pipe-filter architecture.

Parameters:

• ctx (dict) – Context dictionary (in template mode, should contain the ‘variables’ key)

• instruction (str) – Instruction string (in template mode, it is a template instruction, including {variable_name} placeholder)

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Whether to enable template mode (use caching mechanism after enabling)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

async init(start_datetime)

Initialize the router with the start datetime and generate code using LLM. Load the cache set of the current env type from the local cache database and build the FAISS index.

Returns:

Whether any env module restored a checkpoint; forwarded from RouterBase.init.

Return type:

bool

async step(tick, t)

Run forward one step for all simulation modules.

get_cache_stats()

Get cache statistics.

Returns:

CacheStats object, containing all cache statistics

Return type:

CacheStats

get_cache_stats_summary()

Get a summary string of cache statistics.

Returns:

Formatted statistics string

Return type:

str

async clear_cache()

Clear the cache of the current env (memory + persistent DB)

ReActRouter

class agentsociety2.env.router_two_tier_react.TwoTierReActRouter(env_modules, max_steps=10, max_llm_call_retry=10)

Bases: RouterBase

Two-tier ReAct router: first select a module, then call functions in that module.

Workflow: 1. First tier: use the LLM to choose the appropriate environment module. 2. Second tier: use ReAct mode to call tools in the selected module. 3. If multiple modules are needed, repeat in a loop.

__init__(env_modules, max_steps=10, max_llm_call_retry=10)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use ReAct mode to process instructions.

Parameters:

• ctx (dict) – context dictionary

• instruction (str) – command string

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Template mode (not used by ReActRouter, only compatible with signatures)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

PlanExecuteRouter

class agentsociety2.env.router_two_tier_plan_execute.TwoTierPlanExecuteRouter(env_modules, max_steps=10, max_llm_call_retry=10)

Bases: RouterBase

Plan-and-Execute mode Router: first make a plan, and then execute the steps in the plan.

Workflow: 1. First tier: choose the appropriate environment module and create an execution plan. 2. Second tier: execute tool calls in the selected module according to the plan. 3. If multiple modules are needed, repeat in a loop.

__init__(env_modules, max_steps=10, max_llm_call_retry=10)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use Plan-and-Execute mode to process instructions.

Parameters:

• ctx (dict) – context dictionary

• instruction (str) – command string

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Template mode (not used by PlanExecuteRouter, only compatible with signatures)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

ReActRouter

class agentsociety2.env.router_search_tool.SearchToolRouter(env_modules, max_steps=10, max_llm_call_retry=10)

Bases: RouterBase

Search Tool as Tool router: adds a search tool, searches for suitable module functions first, then calls candidate functions.

Workflow: 1. Provide a search_tool that can search all available tools. 2. Use the LLM function-calling capability to call search_tool and find relevant tools. 3. Use function calling to call the selected candidate tools. 4. Repeat until the task is complete.

__init__(env_modules, max_steps=10, max_llm_call_retry=10)

Create a router instance.

Parameters:

• env_modules (list[EnvBase]) – List of environment modules.

• max_steps (int) – Maximum number of execution steps (interpreted by specific router implementation).

• max_llm_call_retry (int) – Lower limit on the maximum number of retries for LLM calls (at least 1).

• replay_writer – Optional replay writer; when provided, it is automatically injected into each env module.

• llm_clients_spec – Optional injected LLMClient handle mapping, such as {"coder": "LLMClient", "default": "LLMClient"}. When provided, the router uses the injected clients; when empty, it builds local clients from configuration.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Use ReAct mode to process instructions.

Parameters:

• ctx (dict) – context dictionary

• instruction (str) – command string

• readonly (bool) – Whether to read-only mode

• template_mode (bool) – Template mode (not used by ReActRouter, only compatible with signatures)

Returns:

(ctx, answer) tuple

Return type:

Tuple[dict, str]

Environment Router Actor

In production, environment routing runs in a dedicated Ray actor, dynamically created by get_env_router_actor_class; agents interact with it through an EnvRouterProxy handle.

EnvRouterProxy

class agentsociety2.env.env_router_proxy.EnvRouterProxy(actor_handle, *, run_dir=None, env_module_types=None)

Agent/society-facing handle delegating to a Ray env-router actor.

Unlike the in-process router, the proxy does NOT hold concrete env-module instances (they live in the actor’s process), so it cannot expose env_modules. To keep env-provided skills discoverable from the agent side, the proxy carries env_skill_dirs – the union of every configured env module’s skill_dirs() resolved from the registry at construction time, each paired with the module’s class name (used as the env:{ClassName} namespace). AgentBase.discover_skill_sources reads this attribute in addition to env_modules, so env skills are visible regardless of whether the agent sees a real router or this proxy. The attribute is plain JSON-serializable data (list[(str, str)]), so the proxy still crosses the Ray boundary cleanly.

__init__(actor_handle, *, run_dir=None, env_module_types=None)

set_current_time(t)

Forward the society clock to the actor (fire-and-forget).

See RouterBase.set_current_time() — this pushes the society’s current time so env modules observe it during the agent phase.

set_replay_writer(writer)

Forward a replay writer/proxy to the actor (fire-and-forget).

The env actor normally receives its replay proxy at construction (so env modules and agents share the same replay directory). This method lets a caller additionally/alternatively inject a writer after construction; it is forwarded to the actor via set_replay_writer. A ReplayProxy is serializable, so it travels across the Ray boundary cleanly.

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)

Forward an env ask to the actor and await the (ctx, answer) result.

async get_world_description()

async init(start_datetime)

async step(tick, t)

async to_workspaces()

Forward per-step env state persistence to the actor.

async from_workspaces()

Forward env state restoration (resume) to the actor.

async close()

agentsociety2.env.env_router_actor.get_env_router_actor_class(max_concurrency=1)

Return (creating once per max_concurrency) the Ray env-router actor class.

Parameters:

max_concurrency (int) – Ray actor concurrency. 1 = serialize ask calls (default when any env module is not concurrency-safe); >1 = allow parallel async ask calls (only safe when all env modules declare is_concurrency_safe()).

Returns:

A Ray actor class (@ray.remote(max_concurrency=...)).

Return type:

Any

For router choices and tradeoffs, see Architecture and Scalability.