环境模块¶

本模块提供环境路由与环境模块的基类。

RouterBase¶

class agentsociety2.env.router_base.RouterBase(env_modules, max_steps=10, max_llm_call_retry=10, replay_writer=None, llm_clients_spec=None)[源代码]¶

基类：ABC

环境路由器抽象基类。

Router 用于协调 agent 与环境模块（EnvBase）的交互，职责包括：

聚合并过滤环境工具（函数调用 schema）
接收编排器推送的仿真时钟（set_current_time()），供 env 模块在 agent 阶段读取
提供统一 LLM 调用封装（重试）
生成 world description（可选，用于 PersonAgent 的提示词）

__init__(env_modules, max_steps=10, max_llm_call_retry=10, replay_writer=None, llm_clients_spec=None)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer (ReplayWriter | None) -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec (Dict[str, Any] | None) -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

set_trace_sink(sink)[源代码]¶

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

参数:: 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)[源代码]¶

与环境交互的统一入口（由子类实现具体路由策略）。

参数:

ctx (dict) -- 上下文字典。模板模式下可包含 variables，用于 {var} 替换。
instruction (str) -- 指令文本。模板模式下会进行变量替换后再执行。
readonly (bool) -- 是否只读（只读时应避免改变环境状态）。
template_mode (bool) -- 是否启用模板模式。
trace_id (str | None) -- OTel trace ID（UUID hex），用于跨 agent/env 的追踪关联。
parent_span_id (str | None) -- OTel parent span ID，用于构建 span 层级。

返回:

(ctx, answer)，其中 ctx 可能被环境更新。

返回类型:

Tuple[dict, str]

async init(start_datetime)[源代码]¶

初始化路由器与环境模块。

参数:: start_datetime (datetime) -- 仿真起始时间。
返回:: 是否有 env 模块恢复了 checkpoint（即 resume）。
返回类型:: bool

async step(tick, t)[源代码]¶

推进环境模块一个仿真步。

参数:

tick (int) -- 本步时间跨度（秒）。
t (datetime) -- 本步结束后的仿真时间。

async close()[源代码]¶: 关闭路由器（关闭所有环境模块）。

bind_env_workspaces(root, module_types)[源代码]¶

把每个 env 模块绑定到 root/module_type（幂等）。

module_types 与 env_modules 按位置对应，同时作为目录键（与 env_kwargs 对齐，避免同类多实例冲突）。

async from_workspaces()[源代码]¶

恢复各已绑定模块的状态（存在快照时）。

任一模块恢复失败不会中断其它模块，但会以 ERROR 级汇总告警——否则会出现「部分模块 fresh、部分模块 checkpoint」的静默不一致状态。

返回:: 是否有任一模块加载了快照（即 resume）。
返回类型:: bool

async to_workspaces()[源代码]¶

把每个模块的动态状态写入其 workspace。

用 return_exceptions=True 收集结果，单个模块落盘失败不会让本步其它模块的持久化全部丢失（否则一次序列化异常会导致整步 checkpoint 落后）。

get_tool_call_history()[源代码]¶

汇总所有环境模块的工具调用历史（按时间排序）。

返回:: 调用记录列表（按 timestamp 升序）。
返回类型:: List[Dict[str, Any]]

reset_tool_call_history()[源代码]¶: 清空所有环境模块的工具调用历史。

set_trace_context(trace_id, parent_span_id)[源代码]¶

为所有环境模块设置当前 OTel trace 上下文。

Router 子类应在 ask 入口处调用此方法，使 @tool 装饰器能将 trace_id 写入 tool call history。

参数:

trace_id (str | None) -- OTel trace ID（UUID hex）。
parent_span_id (str | None) -- OTel parent span ID。

clear_trace_context()[源代码]¶

清除所有环境模块的 trace 上下文。

set_replay_writer(writer)[源代码]¶

为所有环境模块设置回放写入器。

参数:: writer (ReplayWriter) -- ReplayWriter 实例。

get_system_prompt()[源代码]¶

构建 router 侧的通用 system prompt（不含具体任务指令）。

返回:: system prompt 文本。
返回类型:: 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

封装 LLM 调度（Dispatcher 负责并发控制与 HTTP 重试）。

参数:

model (Literal['coder', 'summary']) -- coder 或 summary。
messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) -- 消息列表（不自动追加 system prompt；若需要请用 acompletion_with_system_prompt()）。
stream (bool) -- 是否流式返回。
max_retries (int | None) -- 可选。最大重试次数；为空则使用 self.max_llm_call_retry。
base_delay (float) -- 429 类错误的指数退避基准延迟。
max_delay (float) -- 指数退避最大延迟。

返回:

LLM 响应对象（stream=False 时为 litellm.types.utils.ModelResponse）。

抛出:

ValueError -- 超过重试次数仍失败时抛出。

async acompletion_with_system_prompt(model, messages, **kwargs)[源代码]¶

发送补全请求并自动在最前追加 router system prompt。

参数:

model (Literal['coder', 'summary']) -- coder 或 summary。
messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) -- 消息列表。

返回:

LLM 响应对象。

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)[源代码]¶

发送补全并校验为指定 Pydantic 模型（失败时自动反馈并重试）。

参数:

model (Literal['coder', 'summary']) -- coder 或 summary。
model_type (Type[T]) -- 用于校验的 Pydantic 模型类型。
messages (list[ChatCompletionUserMessage | ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionFunctionMessage | ChatCompletionDeveloperMessage]) -- 消息列表（会自动追加 system prompt）。
max_retries (int) -- 最大重试次数（不含首次尝试）。
base_delay (float) -- 传给 LLM dispatcher 的指数退避基准延迟。
max_delay (float) -- 传给 LLM dispatcher 的指数退避最大延迟。
error_feedback_prompt (str | None) -- 可选。自定义错误反馈模板（需包含 {error_message} 与 {model_schema} 占位符）。

返回:

校验通过的 Pydantic 实例。

抛出:

ValueError -- 解析/校验在重试后仍失败时抛出。

返回类型:

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.

返回:: 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
返回类型:: Dict[str, str]

async get_world_description()[源代码]¶

Get the world description.

async generate_world_description_from_tools(max_retries=None)[源代码]¶

根据环境路由器的模块信息生成简短世界描述。

world_description 只承担“这个仿真世界大概是什么、如何通过 ask_env 与环境交互”的轻量背景说明。模块级细节应由 env skill、工具 schema 或 router 自身处理，避免 world section 挤占 prompt 中 env skill 的空间。

返回:: str: 简短世界描述文本，可直接作为 PersonAgent 的 world section
返回类型:: str

get_collected_pydantic_models()[源代码]¶

获取已收集的所有 Pydantic BaseModel 类型及其源代码。

这些模型是在调用 _collect_tools_info() 时从工具函数的参数类型和返回值类型中收集的。

返回:: 字典，key 是 BaseModel 类型，value 是其源代码
返回类型:: Dict[Type[BaseModel], str]

async generate_final_answer(ctx, instruction, results, process_text=None, status='unknown', error=None)[源代码]¶

根据用户输入和Router输出生成最终答案。

这个方法接收用户输入（ctx, instruction）和Router输出（results和过程文本），使用LLM总结为一个answer，并确定执行状态。使用JSON格式返回，通过Pydantic model验证。

参数:

ctx (dict) -- 上下文字典，包含环境变量等信息
instruction (str) -- 用户的原始指令
results (Dict[str, Any]) -- Router执行的结果字典
process_text (str | None) -- 过程文本（可选），描述执行过程的文本信息
status (str) -- 初步的执行状态（可能被LLM更新）
error (str | None) -- 错误信息（如果有）

返回:

Tuple[str, str]: (final_answer, determined_status) - final_answer: LLM生成的最终答案（summary文本） - determined_status: 确定的执行状态（success/in_progress/fail/error）

返回类型:

Tuple[str, str]

EnvBase¶

class agentsociety2.env.base.EnvBase[源代码]¶

基类：object

环境模块基类。

环境模块定义 Agent 可执行的操作和可观察的状态。通过 @tool 装饰器注册方法为可调用工具，供 Router 调用。

工具类型：

常规工具: @tool(readonly=False) — 可修改环境状态
只读工具: @tool(readonly=True) — 仅查询，不修改状态
观察工具: @tool(readonly=True, kind="observe") — 自动调用
统计工具: @tool(readonly=True, kind="statistics") — 统计信息

子类应实现：

使用 @tool 装饰器定义可执行操作
可选：实现 observe() 方法（默认收集 kind="observe" 的工具）

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.

返回:: True if concurrent tool calls are safe.
返回类型:: bool

__init__()[源代码]¶

property name¶: Name of the environment module

classmethod description()[源代码]¶

Return a short module description for router/world orientation.

classmethod init_description()[源代码]¶

Return AI-readable initialization guidance for this environment module.

返回:: Markdown text with class name, constructor kwargs, accepted data shapes, defaults, and a minimal kwargs example when useful.
返回类型:: 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:

Package modules (e.g., mobility_space/__init__.py): Scans <module_dir>/agent_skills/ directory.
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/.

返回:: List of Path objects pointing to directories containing skill subdirectories. Empty list if no skills are provided.
返回类型:: list[Path]

async init(start_datetime)[源代码]¶

初始化环境模块。

参数:: start_datetime (datetime) -- 仿真起始时间。

async step(tick, t)[源代码]¶

推进环境模块一个仿真步。

参数:

tick (int) -- 本步时间跨度（秒）。
t (datetime) -- 本步结束后的仿真时间。

抛出:

NotImplementedError -- 基类不提供默认实现。

async close()[源代码]¶: 关闭环境模块并释放资源（可选重写）。

async to_workspace(workspace_path=None)[源代码]¶

把动态状态写入 workspace（每步由 society 经 router 调用）。

默认 no-op。子类按自选的格式/文件名把需要跨重启保留的状态写到 self._workspace_root``（或传入的 ``workspace_path）下。建议用 atomic_write_text() 原子写。

async restore(workspace_path)[源代码]¶

从 workspace 恢复动态状态（resume 路径）。

默认返回 False。子类读取自己写入的状态并还原；返回是否成功加载（供 router 区分 fresh / resume）。仅在 __init__ 与 init() 之后调用，故可安全覆盖 init() 设置的字段。

async classmethod from_workspace(workspace_path, **init_kwargs)[源代码]¶

从 workspace 重建就绪模块：cls(**init_kwargs) 后 restore。

参数:: init_kwargs (Any) -- 构造该模块用的 kwargs（resume 时由 actor 从 SOCIETY.json 取）。

get_tool_call_history()[源代码]¶

获取工具调用历史（浅拷贝）。

返回:: 调用记录列表。每条包含 function_name、kwargs、return_value、 exception_occurred、exception_info、timestamp。
返回类型:: list[dict[str, Any]]

reset_tool_call_history()[源代码]¶: 清空工具调用历史。

set_replay_writer(writer)[源代码]¶

设置回放写入器（用于自动建表与写入状态）。

参数:: writer (ReplayWriter) -- ReplayWriter 实例。

TokenUsageStats¶

class agentsociety2.env.router_base.TokenUsageStats(*, call_count=0, input_tokens=0, output_tokens=0)[源代码]¶

Token usage statistics for a model.

变量:

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)[源代码]¶

缓存统计信息

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¶: 缓存命中率

property code_execution_success_rate: float¶: 代码执行成功率

property avg_input_tokens: float¶: 平均输入token数

property avg_output_tokens: float¶: 平均输出token数

property avg_retry_count: float¶: 平均代码重试次数

__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)¶

内置路由器¶

ReActRouter¶

class agentsociety2.env.router_react.ReActRouter(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

基类：RouterBase

ReAct模式Router：通过Reasoning-Acting循环，使用function calling直接调用所有模块的所有函数。

工作流程： 1. 收集所有环境模块的工具信息 2. 使用LLM的function calling能力选择并调用工具 3. 执行工具调用，获取结果 4. 将结果反馈给LLM，继续下一轮Reasoning-Acting循环 5. 直到LLM判断任务完成或达到最大步数

__init__(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用ReAct模式处理指令。

参数:

ctx (dict) -- 上下文字典
instruction (str) -- 指令字符串
readonly (bool) -- 是否只读模式
template_mode (bool) -- 模板模式（ReActRouter 不使用，仅为签名兼容）

返回:

(ctx, answer) 元组

返回类型:

Tuple[dict, str]

PlanExecuteRouter¶

class agentsociety2.env.router_plan_execute.PlanExecuteRouter(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

基类：RouterBase

Plan-and-Execute模式Router：先制定计划，然后执行计划中的步骤。

工作流程： 1. 收集所有环境模块的工具信息 2. 第一阶段：使用LLM制定执行计划（不调用工具） 3. 第二阶段：按照计划逐步执行工具调用 4. 根据执行结果生成最终答案

__init__(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用Plan-and-Execute模式处理指令。

参数:

ctx (dict) -- 上下文字典
instruction (str) -- 指令字符串
readonly (bool) -- 是否只读模式
template_mode (bool) -- 模板模式（PlanExecuteRouter 不使用，仅为签名兼容）

返回:

(ctx, answer) 元组

返回类型:

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)[源代码]¶

基类：RouterBase

代码生成式Router：通过生成Python代码的方式调用环境模块中的@tool标记的接口。

工作流程： 1. 收集所有环境模块的工具信息 2. 使用类似 pyi 文件的 Python 代码格式向 LLM 提供环境模块描述和工具信息（含 pydantic BaseModel 与模块类定义） 3. LLM 生成 Python 代码来调用工具 4. 使用 AST 解析检查代码安全性 5. 通过 compile 和 exec 执行代码，捕获打印输出 6. 根据执行结果和打印输出生成最终响应

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)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer (ReplayWriter | None) -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec (Dict[str, Any] | None) -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用代码生成方式处理指令。通过管道-过滤器架构执行。

参数:

ctx (dict) -- 上下文字典（在template模式下，应包含'variables'键）
instruction (str) -- 指令字符串（在template模式下，为模板指令，包含{variable_name}占位符）
readonly (bool) -- 是否只读模式
template_mode (bool) -- 是否启用模板模式（启用后使用缓存机制）

返回:

(ctx, answer) 元组

返回类型:

Tuple[dict, str]

async init(start_datetime)[源代码]¶

Initialize the router with the start datetime and generate code using LLM. 从本地缓存数据库加载当前 env 类型的缓存集，构建 FAISS 索引。

返回:: 是否有 env 模块恢复了 checkpoint（透传自 RouterBase.init）。
返回类型:: bool

async step(tick, t)[源代码]¶

Run forward one step for all simulation modules.

get_cache_stats()[源代码]¶

获取缓存统计信息。

返回:: CacheStats对象，包含所有缓存统计信息
返回类型:: CacheStats

get_cache_stats_summary()[源代码]¶

获取缓存统计信息的摘要字符串。

返回:: 格式化的统计信息字符串
返回类型:: str

async clear_cache()[源代码]¶

清空当前 env 的缓存（内存 + 持久化 DB）

TwoTierReActRouter¶

class agentsociety2.env.router_two_tier_react.TwoTierReActRouter(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

基类：RouterBase

双层ReAct模式Router：先选择Module，然后再调用该模块的函数。

工作流程： 1. 第一层：使用LLM选择合适的环境模块 2. 第二层：使用ReAct模式调用选中模块的工具 3. 如果需要多个模块，可以循环执行

__init__(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用双层ReAct模式处理指令。

参数:

ctx (dict) -- 上下文字典
instruction (str) -- 指令字符串
readonly (bool) -- 是否只读模式
template_mode (bool) -- 模板模式（TwoTierReActRouter 不使用，仅为签名兼容）

返回:

(ctx, answer) 元组

返回类型:

Tuple[dict, str]

TwoTierPlanExecuteRouter¶

class agentsociety2.env.router_two_tier_plan_execute.TwoTierPlanExecuteRouter(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

基类：RouterBase

双层Plan-and-Execute模式Router：先选择Module并制定计划，然后执行计划。

工作流程： 1. 第一层：选择合适的环境模块并制定执行计划 2. 第二层：按照计划执行选中模块的工具调用 3. 如果需要多个模块，可以循环执行

__init__(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用双层Plan-and-Execute模式处理指令。

参数:

ctx (dict) -- 上下文字典
instruction (str) -- 指令字符串
readonly (bool) -- 是否只读模式
template_mode (bool) -- 模板模式（TwoTierPlanExecuteRouter 不使用，仅为签名兼容）

返回:

(ctx, answer) 元组

返回类型:

Tuple[dict, str]

SearchToolRouter¶

class agentsociety2.env.router_search_tool.SearchToolRouter(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

基类：RouterBase

Search Tool as Tool模式Router：增加一个search tool，先搜索合适的模块函数，然后调用候选函数。

工作流程： 1. 提供一个search_tool，可以搜索所有可用的工具 2. 使用LLM的function calling能力，先调用search_tool找到相关工具 3. 然后使用function calling调用搜索到的候选工具 4. 重复直到任务完成

__init__(env_modules, max_steps=10, max_llm_call_retry=10)[源代码]¶

创建路由器实例。

参数:

env_modules (list[EnvBase]) -- 环境模块列表。
max_steps (int) -- 最大执行步数（由具体 router 实现解释）。
max_llm_call_retry (int) -- LLM 调用最大重试次数下限（至少为 1）。
replay_writer -- 可选回放写入器；若提供会自动注入到各 env module。
llm_clients_spec -- 可选。注入的 LLMClient 句柄映射，例如 {"coder": "LLMClient", "default": "LLMClient"}。提供时 router 使用注入的 client；为空时按配置构建本地 client。

async ask(ctx, instruction, readonly=False, template_mode=False, trace_id=None, parent_span_id=None)[源代码]¶

使用Search Tool模式处理指令。

参数:

ctx (dict) -- 上下文字典
instruction (str) -- 指令字符串
readonly (bool) -- 是否只读模式
template_mode (bool) -- 模板模式（SearchToolRouter 不使用，仅为签名兼容）

返回:

(ctx, answer) 元组

返回类型:

Tuple[dict, str]

环境路由 Actor¶

生产环境下环境路由跑在一个专用的 Ray actor 里（由 get_env_router_actor_class 动态创建）， agent 通过 EnvRouterProxy 句柄与之交互。

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()[源代码]¶

转发每步 env 状态持久化到 actor。

async from_workspaces()[源代码]¶

转发 env 状态恢复（resume）到 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.

参数:: 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()).
返回:: A Ray actor class (@ray.remote(max_concurrency=...)).
返回类型:: Any

各路由器的选择与权衡见架构与可扩展性。