agentscope.app.workspace_managerWorkspaceManagerBase
class WorkspaceManagerBase(ABC)agentscope/app/workspace_manager/_base.py:28–205工作区管理器基类，承载智能体的部署、调度与服务化运行。位于 agentscope.app.workspace_manager 包，继承自 ABC。核心成员包括 bind_storage、assign_workspace_id、get_workspace、close、close_all。
Subclasses are expected to be used as async context managers — entering
the context activates any background machinery the subclass needs (e.g.
a TTL sweeper task) and exiting it tears that machinery down and closes
every cached workspace via :meth:close_all.
The default __aenter__ / __aexit__ cover the common case where a
subclass has no background machinery: enter is a no-op, exit just calls
:meth:close_all. Subclasses that own background tasks should override
both.
构造器摘要Constructor Summary
| 构造器Constructor | 说明Description |
|---|---|
def __init__(*, isolation: IsolationPolicy = IsolationPolicy.PER_AGENT) -> None | 初始化 `WorkspaceManagerBase` 实例。 Bind the isolation policy for :meth:`assign_workspace_id`. |
方法摘要Method Summary
| 方法Method | 说明Description |
|---|---|
def bind_storage(storage: 'StorageBase') -> None | 设置该参数。 Hand the manager the backend holding workspace bindings. |
async def assign_workspace_id(*, user_id: str, agent_id: str, session_id: str) -> strasync | 执行 `assign_workspace_id` 方法。 Mint a workspace id under :attr:`_isolation`. |
async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str | None = None) -> WorkspaceBaseasyncabstractabstractmethod | 返回已初始化工作区。 Return an initialized workspace. |
async def close(workspace_id: str) -> Noneasyncabstractabstractmethod | 关闭连接并释放相关资源。 Close and evict a single workspace from the cache. |
async def close_all() -> Noneasyncabstractabstractmethod | 关闭所有。 Close every cached workspace. |
async def __aenter__() -> Selfasync | 进入异步上下文并返回自身。 Enter the manager's lifetime. Default is a no-op. |
async def __aexit__(*exc: object) -> Noneasync | 退出异步上下文并清理资源。 Exit the manager's lifetime — closes all cached workspaces. |
构造器详情Constructor Detail
WorkspaceManagerBase
def __init__(*, isolation: IsolationPolicy = IsolationPolicy.PER_AGENT) -> None初始化 WorkspaceManagerBase 实例。
Subclasses MUST forward isolation here via super().__init__(isolation=isolation).
| 名称Name | 类型Type | 说明Description |
|---|---|---|
isolationkw默认default IsolationPolicy.PER_AGENT | IsolationPolicy | 用于管理器的隔离粒度。Isolation grain for the manager. |
agentscope/app/workspace_manager/_base.py:42–68方法详情Method Detail
bind_storage
def bind_storage(storage: 'StorageBase') -> None设置该参数。
Wired by :func:agentscope.app.create_app rather than taken as a constructor argument, because the application author builds the manager before the app exists to supply a storage backend.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
storage | 'StorageBase' | 存储（类型 `'StorageBase'`）。The application's storage backend. |
agentscope/app/workspace_manager/_base.py:70–81assign_workspace_id
async def assign_workspace_id(*, user_id: str, agent_id: str, session_id: str) -> str执行 assign_workspace_id 方法。
Called by the session-creation flow when the caller did not supply an explicit workspace_id.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
user_idkw | str | 用户标识（类型 `str`）。The owning user id. |
agent_idkw | str | 智能体标识（类型 `str`）。The agent the session belongs to. |
session_idkw | str | 会话标识（类型 `str`）。The session id being provisioned (only used by the per-session grain to underline its randomness). |
agentscope/app/workspace_manager/_base.py:83–153get_workspace
async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str | None = None) -> WorkspaceBase返回已初始化工作区。
Return an initialized workspace.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
user_id | str | 用户标识。The user id. |
agent_id | str | 智能体标识。The agent id. |
session_id | str | 会话标识。The session id. |
workspace_id默认default None | str | None | 工作区标识（类型 `str | None`）。Explicit workspace binding. An empty id or `None` triggers the :meth:`assign_workspace_id` fallback — expected only for callers without a persisted binding. |
agentscope/app/workspace_manager/_base.py:164–184close
async def close(workspace_id: str) -> None关闭连接并释放相关资源。
Close and evict a single workspace from the cache.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
workspace_id | str | 工作区标识（类型 `str`）。The workspace id. |
agentscope/app/workspace_manager/_base.py:187–188close_all
async def close_all() -> None关闭所有。
Pure "close all currently tracked workspaces" semantics — does not imply the manager itself is being torn down.
agentscope/app/workspace_manager/_base.py:191–197__aenter__
async def __aenter__() -> Self进入异步上下文并返回自身。
Enter the manager's lifetime.
agentscope/app/workspace_manager/_base.py:199–201__aexit__
async def __aexit__(*exc: object) -> None退出异步上下文并清理资源。
Exit the manager's lifetime — closes all cached workspaces.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
exc | object | exc（类型 `object`）。The exc. |
agentscope/app/workspace_manager/_base.py:203–205