Society 模块

本模块提供仿真的核心编排器和辅助工具。

AgentSociety

class agentsociety2.society.AgentSociety(agent_specs, agent_class_name, env_router, start_t, run_dir=None, *, service_proxy=None, batch_size=None, enable_replay=True, env_module_types=None, env_kwargs=None)

基类：object

Record-based 仿真编排器（不持有 agent 对象）。

AgentSociety 是框架的核心类，负责管理仿真生命周期：

• 持有 agent_specs**（元数据：``{"id","profile","config"}``）与 **agent_ids， 绝不在主进程常驻 agent 对象。

• step 每 tick 切批提交 step_agent_batch Ray Task，跨 worker 并行；批内顺序执行 （每个 agent 是 LLM-bound，顺序无妨）。

• ask/intervene/run_questionnaire/dump 为低频外部查询，按需在主进程 from_workspace 重建目标 agent（workspace 在本地磁盘）。

Example:

from datetime import datetime
from pathlib import Path
from agentsociety2.society import AgentSociety

society = AgentSociety(
    agent_specs=[{"id":1,"profile":{...},"config":{...}}, ...],
    agent_class_name="PersonAgent",
    env_router=env_proxy,
    service_proxy=proxy,
    start_t=datetime.now(),
    run_dir=Path("./run"),
)

async with society:
    await society.run(num_steps=100, tick=3600)

__init__(agent_specs, agent_class_name, env_router, start_t, run_dir=None, *, service_proxy=None, batch_size=None, enable_replay=True, env_module_types=None, env_kwargs=None)

创建 record-based 仿真编排器。

参数:

• agent_specs (list[dict]) -- agent 元数据列表，每个形如 {"id": int, "profile": dict, "config": dict}。仅元数据，不在构造时实例化。

• agent_class_name (str) -- agent 类名（经 registry 解析，如 "PersonAgent"）。

• env_router (RouterBase) -- 环境路由器（in-process 或 EnvRouterProxy）。

• start_t (datetime) -- 仿真开始时间。

• run_dir (Optional[Path]) -- 运行目录（run_dir/agents 为 agent workspaces 根）。

• service_proxy (Optional['ServiceProxy']) -- 共享服务句柄（env / llm / trace / replay）。流式任务 处理必需——runner 任务通过它访问 LLM clients / env actor。提供时 init() 直接复用；close 关闭其中 trace/replay actor。

• batch_size (Optional[int]) -- 每 step_agent_batch Ray Task 处理的 agent 数； None 时回落到 Config.BATCH_SIZE``（环境变量 ``AGENTSOCIETY_BATCH_SIZE，缺省 256）。

• enable_replay (bool) -- 是否启用回放记录。

property current_time: datetime

返回当前仿真时间。

property step_count: int

返回已执行的仿真步数。

property agent_ids: list[int]

返回所有 agent id（不暴露 agent 对象）。

property agent_specs: list[dict]

返回 agent specs（元数据快照）。

async init()

初始化：Ray / LLM dispatcher → service_proxy → replay → 批式创建 workspaces → env init。

async close()

关闭：env router + service_proxy 的 trace/replay 句柄 + LLM dispatcher。

async to_workspace(*, tick=None)

持久化 society + env 状态（每步调用，无状态 resume）。

让 env router 落盘各模块状态，并写 run_dir/SOCIETY_STEP.json``（仅少量 标量：当前时间 / 步数 / 已完成前置步数 / terminated）。不可变的大字段 （agent_specs 等）只在 :meth:`init` 写一次到 ``SOCIETY.json，避免每步 重序列化。run_dir 为空时 no-op；env 落盘失败只告警，不中断 step。

参数:

tick (int | None) -- 保留以兼容旧调用签名（不再单独持久化）。

mark_step_completed(step_idx)

标记第 step_idx 个顶层 step 已完成并立即持久化进度。

由 CLI 在每个顶层 step（含 Ask/Intervene/Questionnaire）成功后调用—— 否则连续的非 RunStep 完成后若崩溃，resume 会把它们当作未执行而重跑。

async classmethod from_workspace(run_dir, *, env_router, service_proxy=None)

从 run_dir 重建 society（resume）。

读 SOCIETY.json``（不可变：agent specs、env 模块类型+kwargs）+ ``SOCIETY_STEP.json``（每步标量：当前时间、步数、已完成前置步数）。 ``env_router 必须由调用方（CLI）预先构造（actor 不可在此重建）； env 模块状态由 actor 在自身 init() 中恢复。

返回的 society 的 init() 会跳过 agent workspace 创建与 profile 重写 （原 run 已存在），也不会覆盖已有的 SOCIETY.json。

抛出:

• FileNotFoundError -- run_dir/SOCIETY.json 不存在。

• ValueError -- schema 不兼容或 checkpoint 损坏。

async step(tick)

推进一次仿真步：sync clock → 切批 step_agent_batch → env.step → 前进时钟。

参数:

tick (int) -- 本步时间跨度（秒）。

async run(num_steps, tick)

运行多步仿真。

参数:

• num_steps (int) -- 运行步数上限。

• tick (int) -- 每步时间跨度（秒）。

async ask(question)

向仿真系统提问（由 helper 协调 agents/env 作答）。

参数:

question (str) -- 问题文本。

返回:

答案文本。

返回类型:

str

async intervene(instruction)

对仿真进行干预（由 helper 协调执行）。

参数:

instruction (str) -- 干预指令文本。

返回:

执行结果/反馈文本。

返回类型:

str

async run_questionnaire(questionnaire, target_agent_ids=None)

向目标 agents 发放问卷并返回结构化结果。

目标 agent 按 id 切批，每批提交一个 questionnaire_agent_batch Ray Task （与 step 同款分布式批处理）：worker 进程内 from_workspace 重建本批 agents、并发跑完整个问卷、to_workspace 落盘，再返回结构化结果。失败的单个 agent 被隔离（不中断整批），与 step_agent_batch 的容错语义一致。

参数:

• questionnaire (Questionnaire) -- 问卷定义。

• target_agent_ids (list[int] | None) -- 目标 agent id（默认全部）。

返回:

汇总后的问卷响应。

返回类型:

QuestionnaireResponse

AgentSocietyHelper

class agentsociety2.society.helper.AgentSocietyHelper(env_router, society, max_steps=8, max_replans=2, max_llm_call_retry=10)

基类：object

Plan-and-Execute helper that answers external questions or executes interventions by first creating a plan, then executing steps, with support for dynamic replanning.

__init__(env_router, society, max_steps=8, max_replans=2, max_llm_call_retry=10)

Create a plan-and-execute helper bound to a record-based society.

The helper holds the AgentSociety handle and reconstructs target agents on demand via society._reconstruct_agent / society._reconstruct_agents (low-volume external queries; workspaces are on local disk). Filter-by-profile over the full population works on the society's specs (no reconstruction) when the profile field is directly readable from the spec.

参数:

• env_router (RouterBase) -- Environment router (in-process or EnvRouterProxy).

• society (AgentSociety) -- The record-based AgentSociety (holds specs/ids + reconstruction callbacks). Required.

• max_steps (int) -- Max plan steps per ask/intervene.

• max_replans (int) -- Max replan attempts on failure.

• max_llm_call_retry (int) -- Max LLM retries per planning/summary call.

async ask(question)

async intervene(instruction)

PlanStep

class agentsociety2.society.helper.PlanStep(*, description, tool, args, expected_output)

A single step in the execution plan.

description: str

Clear description of what this step does.

tool: str

The name of the tool to use.

args: Dict[str, Any]

Arguments to pass to the tool.

expected_output: str

What information or result this step should produce.

model_config = {}

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