Agent 模块

本模块提供智能体的核心类和数据模型。

核心类

AgentBase

class agentsociety2.agent.base.AgentBase

基类：ABC

Abstract base class for all agents.

Directly owns the generic agent runtime (workspace binding, skill runtime, ReAct tool dispatch) plus identity, service slots, and the abstract contract. Construction is via create / from_workspace; __init__ is arg-less.

Subclasses must implement (forced abstracts):

• create()

• from_workspace()

• to_workspace()

• ask()

• step()

__init__()

Initialize an unbound agent (arg-less).

Only empty slots are set; no profile parsing, no id required. The real initialization happens in restore() (called by from_workspace()).

classmethod create(workspace_path, profile, config)

Create the initial agent workspace (static, write-once config).

Writes config.json (static, never rewritten) + initial AGENT.json + the standard empty directories. Does NOT return an agent instance; use from_workspace() to reconstruct.

参数:

• workspace_path (Path) -- Agent workspace root (created if missing).

• profile (dict) -- Agent profile dict (must include id).

• config (dict) -- Static config dict written verbatim to config.json.

async classmethod from_workspace(workspace_path, service_proxy)

Reconstruct a ready agent from its workspace.

agent = cls() (arg-less); await agent.restore(ws, proxy); return the agent.

参数:

• workspace_path (Path) -- Agent workspace root.

• service_proxy (ServiceProxy) -- Shared service container (env / llm / trace / replay).

返回:

Ready agent instance.

返回类型:

AgentBase

async restore(workspace_path, service_proxy)

The real initialization (called by from_workspace).

Reads config.json + AGENT.json; sets _id/_profile/ _name/_config; calls _bind_services + _bind_workspace; restores visible/activated skills + counters.

Subclasses override this to ALSO restore person-specific state after await super().restore(...).

参数:

• workspace_path (Path) -- Agent workspace root.

• service_proxy (ServiceProxy) -- Shared service container.

abstractmethod async to_workspace(workspace_path)

Write current dynamic state back to the workspace.

abstractmethod async ask(message, readonly=True, *, t=None)

Answer an external question through the agent's reasoning flow.

abstractmethod async step(tick, t)

Run one simulation step.

property id: int

Agent unique identifier.

property name: str

Agent display name.

property logger: Logger

Agent-scoped logger.

env_ask_env_ctx_overlay()

Generate the ask_env / CodeGenRouter.ask context overlay.

Returns the stable identity keys (id, agent_id, person_id) provided by the framework, independent of any skill.

返回:

dict with id, agent_id, person_id.

返回类型:

dict[str, Any]

async ask_env(ctx, message, readonly, template_mode=False, trace_id=None, parent_span_id=None)

Send a request to the environment router.

Wraps interaction with the simulation environment, supporting template mode and context-variable substitution.

参数:

• ctx (dict) -- Context dict (may include a variables key for template mode).

• message (str) -- Request message.

• readonly (bool) -- Whether this is a read-only request.

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

• trace_id (str | None) -- OTel trace ID for cross-agent/env correlation.

• parent_span_id (str | None) -- OTel parent span ID.

返回:

Tuple (ctx, answer).

async acompletion(messages, stream=False, **kwargs)

Single LLM completion via the bound default-role dispatcher.

Convenience for subclasses that need a one-shot LLM call (the full multi-turn tool loop is run_react_loop()). Uses the agent's bound dispatcher + model.

参数:

• messages (list[dict[str, Any]]) -- Chat messages.

• stream (bool) -- Streaming flag (False by default; the local LLM dispatcher does not support streaming).

• **kwargs (Any) -- Extra litellm kwargs (timeout, tools, ...).

返回:

The LLM response (litellm ModelResponse).

返回类型:

Any

async close()

Close the agent and release resources. Subclasses may override.

get_profile()

Return the agent profile as a dict.

返回:

Profile dict (raw dict, model_dump, or {"raw": str}).

返回类型:

Dict[str, Any]

classmethod description()

Return a short agent description for module lists.

classmethod init_description()

Return AI-readable initialization guidance for this agent class.

workspace_root_path()

Return the agent-owned workspace root.

抛出:

RuntimeError -- When the workspace is not initialized.

trace_span(name, *, trace_id=None, parent_span_id=None, attributes=None, end_attributes=None)

Create an agent-owned trace span context.

build_agent_json(*, tick, t)

Build the consolidated self-description file for the agent.

Subclasses may override to add extra fields (e.g. memory summary).

参数:

• tick (int | None) -- Current simulation tick, if available.

• t (Any | None) -- Current simulation time, if available.

返回:

Serializable AGENT.json dictionary.

返回类型:

dict[str, Any]

persist_agent_json(*, tick=None, t=None)

Persist consolidated agent self-description to AGENT.json.

参数:

• tick (int | None) -- Current simulation tick, if available.

• t (Any | None) -- Current simulation time, if available.

返回:

Data written to AGENT.json.

返回类型:

dict[str, Any]

discover_skill_sources(env, *, extra_skill_paths=None)

Discover custom and environment-provided skill sources.

Scans the default <root>/custom/skills locations (env.run_dir and this agent's workspace root), then any extra skill directories merged from two sources:

• declarative: extra_skill_paths in config.json (resolved in restore());

• programmatic: the extra_skill_paths argument here.

Each extra entry is a directory containing skill subdirectories (same layout as custom/skills). Relative entries resolve against the workspace root. The default scan is unchanged when neither is given, so agents can load custom skills stored anywhere via either entry point.

参数:

• env (RouterBase) -- Environment router used to locate run directories and modules.

• extra_skill_paths (Iterable[Path | str] | None) -- Additional skill directories to scan now (each a root of skill subdirectories).

返回:

Mapping from discovery source labels to added skill IDs.

返回类型:

dict[str, list[str]]

async run_lifecycle_hooks(hook_type, *, tick, t)

Run lifecycle hooks for currently activated skills.

参数:

• hook_type (str) -- Lifecycle hook name, such as pre_step or post_step.

• tick (int) -- Current simulation tick.

• t (Any) -- Current simulation time.

返回:

Hook execution summaries.

返回类型:

list[dict[str, Any]]

async dispatch_react_tool(action, args, *, readonly=False)

Dispatch one ReAct tool call.

Handles workspace file tools, skill tools, and ask_env. Person-specific tools (memory_*, todo_*) are handled by the subclass before this method is reached (or via a subclass override).

参数:

• action (str) -- Tool name.

• args (dict[str, Any]) -- Tool argument dictionary.

• readonly (bool) -- Whether mutation tools should be blocked.

返回:

Tool execution result.

返回类型:

ReactToolResult

async run_react_loop(*, tick, t, observations=None, question=None, readonly=False, skill_hooks=None)

Run the ReAct loop until finish or turn limit.

参数:

• tick (int) -- Current simulation tick.

• t (datetime) -- Current simulation time.

• observations (list[dict[str, Any]] | None) -- Initial observation list (appended in-place).

• question (str | None) -- Optional external question (ask mode).

• readonly (bool) -- Whether mutation tools are blocked.

• skill_hooks (list[dict[str, Any]] | None) -- Optional pre_step hook outputs for the prompt.

返回:

Final result string from the loop.

返回类型:

str

build_react_messages(*, tick, t, observations, question=None, readonly=False, skill_hooks=None)

Build ReAct prompt messages (subclass prompt hook).

The base implementation raises NotImplementedError. Subclasses (e.g. PersonAgent) override this to assemble their agent-specific prompt and return OpenAI-style chat messages.

参数:

• tick (int) -- Current simulation tick.

• t (datetime) -- Current simulation time.

• observations (list[dict[str, Any]]) -- Recent observation list.

• question (str | None) -- Optional external question (ask mode).

• readonly (bool) -- Whether mutation tools are blocked.

• skill_hooks (list[dict[str, Any]] | None) -- Optional pre_step hook outputs for the prompt.

返回:

OpenAI-style chat messages.

抛出:

NotImplementedError -- Always, in the base class.

返回类型:

list[dict[str, str]]

dispatch_todo_tool(action, args)

Dispatch built-in TODO tools.

参数:

• action (str) -- TODO tool action name.

• args (dict[str, Any]) -- Tool argument dictionary.

返回:

Tool execution result.

返回类型:

ReactToolResult

PersonAgent

class agentsociety2.agent.person.PersonAgent

基类：AgentBase

Workspace-backed simulated person agent.

Person-specific responsibilities:

• Memory runtime (PersonMemoryRuntime): MEMORY.md + episodes, plus the memory_* tool dispatch.

• todo_* tool dispatch (the generic TODO state store + helpers live on AgentBase; PersonAgent routes the todo_* prefix to them).

• Person prompt building (build_react_messages) — the prompt hook the generic ReAct loop calls.

• step / ask orchestration (observe + react loop + lifecycle hooks).

Generic agent machinery (workspace binding, skill discovery, file/skill/ask_env tool dispatch, the ReAct loop, LLM dispatch, response parsing, TODO state, trace spans, AGENT.json persistence, construction model) is inherited from AgentBase.

classmethod description()

Return a short registry description.

classmethod init_description()

Return initialization guidance for generated configs.

async restore(workspace_path, service_proxy)

Restore base state, then build person-specific runtime (memory).

参数:

• workspace_path (Path) -- Agent workspace root.

• service_proxy (ServiceProxy) -- Shared service container.

async to_workspace(workspace_path)

Write current dynamic state back to the workspace.

Writes AGENT.json (profile / step_count / current_time / skills / initialized_at). config.json is NOT rewritten. Memory (MEMORY.md + episodes) is persisted via the memory runtime's after_step path during step().

参数:

workspace_path (Path) -- Agent workspace root.

build_agent_json(*, tick, t)

Build AGENT.json with person-specific memory / skill fields.

async dispatch_react_tool(action, args, *, readonly=False)

Dispatch person-specific prefixes, then delegate the rest to base.

Person-specific prefixes handled here:

• memory_* → _dispatch_memory_tool() (person memory runtime).

• todo_* → dispatch_todo_tool() (inherited from AgentBase; the generic TODO store + arg normalization).

All other actions (workspace / skill / ask_env) are delegated to await super().dispatch_react_tool(action, args, readonly=readonly).

build_react_messages(*, tick, t, observations, question=None, readonly=False, skill_hooks=None)

Build ReAct prompt messages (person-specific).

async step(tick, t)

Run one simulation step.

async ask(message, readonly=True, *, t=None)

Answer an external question through the ReAct loop.

AgentBase 直接拥有 workspace 绑定、技能运行时、ReAct 工具循环、TODO 状态、trace 与 AGENT.json 持久化等通用能力。PersonAgent 是其上的薄编排器，只实现面向人物的行为逻辑。 智能体不再通过 __init__ 传参构造，而是经 AgentBase.create() 写一次 workspace、再经 await AgentBase.from_workspace(path, service_proxy) 重建（详见 使用智能体）。

如果只是扩展人物行为，优先新增或修改 Agent Skill（见 Agent Skills（智能体技能），API 参考见 Skills 模块）；只有需要改变智能体生命周期、状态机或外部系统调用方式时，才继承 AgentBase / PersonAgent 创建新类。