工作区管理 - AgentScope

工作区管理器（Workspace Manager）持有隔离策略，并管理智能体服务中每个智能体的 Workspace 实例的生命周期。它是 agentscope.workspace 模块在服务侧的对应实现 —— 每一种 WorkspaceBase 子类都有一个配套的 WorkspaceManagerBase 的子类来负责创建、缓存与删除。它的特点：

可配置的隔离粒度 —— PER_AGENT（默认）、PER_SESSION 或 PER_USER —— 决定 workspace 在 (user_id, agent_id, session_id) 三个维度上如何共享或隔离，让同一套部署可以干净地服务多个租户。
可插拔的沙箱后端 —— 同一份生命周期契约既可以包裹本地目录、Docker 容器、E2B 云沙箱、OpenSandbox 沙箱，也可以对应 Kubernetes Pod。切换后端只需在 create_app 中改一行。

可选的实现

类	说明
`LocalWorkspaceManager`	直接落在宿主目录下的裸机式 workspace。零基础设施、零沙箱化 —— 智能体直接跑在开发者的宿主文件系统上。
`DockerWorkspaceManager`	每个 workspace 对应一个 Docker 容器，宿主目录以 bind mount 的方式挂进去做持久化。需要一个可达的 Docker daemon。
`E2BWorkspaceManager`	通过 E2B 托管的云沙箱。空闲时自动挂起，下一次访问时恢复。
`OpenSandboxWorkspaceManager`	通过 OpenSandbox 托管的云沙箱。沙箱在暂停/恢复之间保留自己的文件系统；重新挂接依据沙箱 metadata 进行，因此任意服务副本都能重连。
`K8sWorkspaceManager`	在 Kubernetes 集群上为每个 workspace 起一个 Pod + PVC。适合已经在跑其他负载的生产集群。

想深入了解 workspace 究竟是什么（文件系统布局、gateway、MCP 接线、内置工具），请看 Workspace 章节。

接入智能体服务

在 create_app 入口处设置对应的 WorkspaceManager 实例即可。下面是五种内置实现的示例：

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    IsolationPolicy,
    LocalWorkspaceManager,
)

workspace_manager = LocalWorkspaceManager(
    # 宿主根目录；每个 agent 的工作目录落在 `<basedir>/<user_id>/<agent_id>` 下。
    basedir="/data/workspaces",
    # `PER_AGENT`（默认）：同一 agent 的所有 session 共享一个 workspace。
    # `PER_SESSION` / `PER_USER` 改变粒度，详见下文。
    isolation=IsolationPolicy.PER_AGENT,
)

app = create_app(
    # ...existing code...
    workspace_manager=workspace_manager,
)

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    DockerWorkspaceManager,
    IsolationPolicy,
)

workspace_manager = DockerWorkspaceManager(
    # 宿主根目录，会 bind mount 到每个容器内的 `/workspace`。
    basedir="/data/workspaces",
    isolation=IsolationPolicy.PER_AGENT,
    # 基础镜像，必须自带 `python3`。镜像按内容哈希，
    # 只有 Dockerfile 输入变化时才会重新构建。
    base_image="python:3.11-slim",
    # 镜像里内置的 Node.js 主版本号（基于 npx 的 MCP 会用到）。
    node_version="20",
    # 空闲容器在被 sweeper 清掉前的缓存时间（秒）。
    ttl=3600.0,
)

app = create_app(
    # ...existing code...
    workspace_manager=workspace_manager,
)

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    E2BWorkspaceManager,
    IsolationPolicy,
)

workspace_manager = E2BWorkspaceManager(
    isolation=IsolationPolicy.PER_AGENT,
    # E2B template id，模板里要自带 agent 所需的运行时。
    template="base",
    # 传 `""` 时会回退去读环境变量 `E2B_API_KEY`。
    api_key="",
    # E2B 侧的沙箱 keep-alive 超时时间。
    timeout_seconds=300,
    ttl=3600.0,
)

app = create_app(
    # ...existing code...
    workspace_manager=workspace_manager,
)

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    IsolationPolicy,
    OpenSandboxWorkspaceManager,
)

workspace_manager = OpenSandboxWorkspaceManager(
    isolation=IsolationPolicy.PER_AGENT,
    # 沙箱启动所用的镜像；gateway 的 venv 会在其之上引导安装。
    image="python:3.11-slim",
    # 传 `""` 时会让 `opensandbox` SDK 回退到基于环境变量的配置。
    api_key="",
    # 可选的服务端 domain，用于自托管 / 本地部署。
    domain="",
    protocol="http",
    # OpenSandbox 侧的沙箱 keep-alive 超时时间。
    timeout_seconds=300,
    ttl=3600.0,
)

app = create_app(
    # ...existing code...
    workspace_manager=workspace_manager,
)

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    IsolationPolicy,
    K8sWorkspaceManager,
)

workspace_manager = K8sWorkspaceManager(
    isolation=IsolationPolicy.PER_AGENT,
    # 存放 workspace Pod 与 PVC 的 K8s namespace。
    namespace="agentscope",
    # 传 `None` 用集群内配置；集群外使用时传一个 kubeconfig 路径。
    kubeconfig=None,
    # 每个 Pod 的容器镜像。
    image="python:3.11-slim",
    # 为 workspace 文件系统兜底的 PVC 大小。
    storage_size="1Gi",
    ttl=3600.0,
)

app = create_app(
    # ...existing code...
    workspace_manager=workspace_manager,
)

隔离粒度

isolation 决定 workspace 在 (user_id, agent_id, session_id) 三元组上如何共享或隔离。Manager 会在 session 创建时按所选策略生成一个 workspace_id；后续所有带着同一个 workspace_id 的请求都会落到同一个底层 workspace 上。

取值	共享规则	典型场景
`PER_AGENT`（默认）	同一 `(user_id, agent_id)` 下的所有 session 共享一个 workspace	让每个 agent 拥有一份持久化的、按用户区分的工作目录 —— 文件、skill、MCP 注册跨 session 也能保留下来。
`PER_SESSION`	每个 session 一个独立的 workspace	需要保证 session 之间互不泄露状态的场景 —— 例如一次性沙箱评测、短生命周期的自动化任务。
`PER_USER`	同一 `user_id` 下的所有 session 共享一个 workspace，不区分是哪个 agent	同一用户的多个 agent 在同一份文件系统上协作（少见，谨慎使用）。

在 session 创建请求中显式指定 workspace_id 永远会覆盖策略。这就是内置团队工具（AgentCreate / AgentInvite）用来让子 agent 与团队 leader 共用同一个 workspace 的机制。

在智能体服务中的工作原理

工作区管理器是一个应用级单例资源：一个智能体服务进程只有一份实例，所有请求共享。它的职责有两个：（a）在配置的 isolation 策略下判断一次请求属于哪个 workspace；（b）把该 workspace 作为一个廉价依赖交给需要它的路由和服务。

接线方式

开发者传给 create_app(workspace_manager=...) 的实例会被挂到 app.state 上，并通过一个 FastAPI 依赖对外暴露：

# agentscope/app/_app.py
app.state.workspace_manager = workspace_manager

# agentscope/app/deps.py
async def get_workspace_manager(request: Request) -> WorkspaceManagerBase:
    return request.app.state.workspace_manager

所有会碰到 workspace 的 router、service 与内置工具都通过这个依赖来消费它（服务类则通过构造函数注入）：

Router —— /session、/workspace/*（MCP、skill）。它们调用 Depends(get_workspace_manager)，然后用 manager.get_workspace(...) 解析出调用方对应的 workspace。
ChatService —— 在 lifespan 里通过构造函数拿到 manager，每次 chat 运行时调用 get_workspace 拿到 agent 将要作用其中的 workspace。
内置团队工具（AgentCreate / AgentInvite）—— 拿到 manager 后在派生子 agent session 时调用 assign_workspace_id(...)，让子 agent 落到与 leader 相同的 workspace 里。

生命周期

Manager 是一个异步上下文管理器。应用的 lifespan 会在启动时进入它，在关闭时通过一个 AsyncExitStack 退出它，所以它所拥有的任何后台机制（例如 DockerWorkspaceManager / E2BWorkspaceManager / OpenSandboxWorkspaceManager / K8sWorkspaceManager 的空闲 TTL sweeper）都会与服务的其他部分同步启停：

# agentscope/app/_lifespan.py（节选）
workspace_manager = app.state.workspace_manager
async with AsyncExitStack() as stack:
    await stack.enter_async_context(storage)
    await stack.enter_async_context(message_bus)
    await stack.enter_async_context(workspace_manager)  # ← 就是这里
    ...
    yield
# ← 退出时，`__aexit__` 会调用 `close_all()` 拆掉每一个缓存中的 workspace。

端到端请求流程

服务跑起来之后，一次典型的交互长这样：

创建 session —— /session router 只计算一次 workspace id 并把它写到 session 记录上：
```
# agentscope/app/_router/_session.py（节选）
resolved_workspace_id = body.workspace_id or (
    workspace_manager.assign_workspace_id(
        user_id=user_id,
        agent_id=agent_id,
        session_id=session_id,
    )
)
```
显式传入的 body.workspace_id 永远优先（团队工具就是靠它让子 agent 共享 leader 的 workspace）；否则由 manager 按配置的 isolation 策略生成一个。assign_workspace_id 是一个纯函数 —— 不做 I/O，不查缓存 —— 所以在热路径上调用非常便宜。
后续每个请求带着这个 session —— 无论是 chat 运行、MCP 注册还是 skill 上传 —— 都会从 session 记录里读出保存好的 workspace_id，然后请求 manager 把它落到具体的 workspace 上：
```
workspace = await workspace_manager.get_workspace(
    user_id, agent_id, session_id, session_record.config.workspace_id,
)
```
get_workspace 是缓存命中的路径。它在 manager 内部的 dict 里查 workspace_id，刷新一下最后访问时间，就在 O(1) 内返回已初始化好的 Workspace。只有 miss 的时候才会抢一把 asyncio.Lock，把后端准备好（构建 Docker 镜像、启动 E2B 或 OpenSandbox 沙箱、创建 K8s Pod……），再把结果缓存起来 —— 这把锁能防止多个请求同时冲进来给同一个 id 起两份后端。
空闲驱逐 —— 沙箱类的 manager 会跑一个后台 sweeper（在 __aenter__ 里启动），周期性遍历缓存，把最后访问时间超过 ttl 的 workspace 关闭并踢出。下一次访问该 id 时会透明地重新准备一份。
关闭 —— lifespan 退出 manager，manager 调用 close_all() 拆掉所有缓存中的 workspace 及其后台 sweeper。

由于 workspace_id 只在 session 创建时生成一次并持久化，运行中的部署改动 isolation 策略并不会对现存 session 重新分组 —— 它们会继续保留当初分配的 id。只有新 session 才会走新策略。

核心 API

每个 manager 都要实现来自 WorkspaceManagerBase 的以下契约。自定义子类只需要补齐这些抽象方法即可；assign_workspace_id 里的隔离逻辑由基类提供，并由 isolation 构造参数驱动。

方法	用途
`assign_workspace_id(*, user_id, agent_id, session_id) -> str`	在配置的 `isolation` 策略下为一个新 session 生成 workspace id。纯函数，无 I/O。当客户端在 session 创建请求中省略了 `workspace_id` 时会被调用。
`get_workspace(user_id, agent_id, session_id, workspace_id=None) -> WorkspaceBase`	返回一个绑定到 `workspace_id` 的已初始化 workspace。热路径上的缓存命中路径；miss 时 manager 会准备底层后端并缓存结果。`workspace_id=None` 会回退到 `assign_workspace_id`。
`create_workspace(user_id, agent_id, session_id) -> WorkspaceBase`	全新准备一个 workspace 并纳入管理。调用方还没有可复用的 id 时使用。
`close(workspace_id)`	把某一个 workspace 从缓存中踢掉，并拆掉它的后端。
`close_all()`	踢掉所有缓存的 workspace —— 服务关闭时调用。
`async with manager: ...`	进入 / 退出 manager 的生命周期。进入时启动后台机制（例如 TTL sweeper）；退出时调用 `close_all`。

自定义 Manager

任何基于 WorkspaceBase 构建的 workspace 类，都可以通过配一个 WorkspaceManagerBase 子类的方式接到智能体服务里。大多数情况下只需要写一小段准备 + 缓存的样板代码 —— 隔离策略由基类自动继承。

自定义 Manager

import asyncio
import time
from typing import Self

from agentscope.app.workspace_manager import (
    IsolationPolicy,
    WorkspaceManagerBase,
)
from agentscope.workspace import WorkspaceBase


class MyWorkspaceManager(WorkspaceManagerBase):
    """封装自定义 `MyWorkspace` 后端的轻量 manager。"""

    def __init__(
        self,
        *,
        isolation: IsolationPolicy = IsolationPolicy.PER_AGENT,
        ttl: float = 3600.0,
    ) -> None:
        super().__init__(isolation=isolation)
        self._ttl = ttl
        # workspace_id -> (workspace, last_access_monotonic)
        self._cache: dict[str, tuple[WorkspaceBase, float]] = {}
        self._lock = asyncio.Lock()

    async def get_workspace(
        self,
        user_id: str,
        agent_id: str,
        session_id: str,
        workspace_id: str | None = None,
    ) -> WorkspaceBase:
        # 当此 session 没有持久化的显式绑定时，
        # 回退到 manager 的 isolation 策略。
        if workspace_id is None:
            workspace_id = self.assign_workspace_id(
                user_id=user_id,
                agent_id=agent_id,
                session_id=session_id,
            )
        async with self._lock:
            hit = self._cache.get(workspace_id)
            if hit is not None:
                ws, _ = hit
                self._cache[workspace_id] = (ws, time.monotonic())
                return ws
            ws = MyWorkspace(workspace_id=workspace_id)  # 你的后端
            await ws.initialize()
            self._cache[workspace_id] = (ws, time.monotonic())
            return ws

    async def create_workspace(
        self,
        user_id: str,
        agent_id: str,
        session_id: str,
    ) -> WorkspaceBase:
        ws = MyWorkspace()  # 让 workspace 自己生成 id
        await ws.initialize()
        async with self._lock:
            self._cache[ws.workspace_id] = (ws, time.monotonic())
        return ws

    async def close(self, workspace_id: str) -> None:
        async with self._lock:
            entry = self._cache.pop(workspace_id, None)
        if entry is not None:
            await entry[0].close()

    async def close_all(self) -> None:
        async with self._lock:
            entries = list(self._cache.values())
            self._cache.clear()
        await asyncio.gather(
            *(ws.close() for ws, _ in entries),
            return_exceptions=True,
        )

把一个 MyWorkspaceManager 实例传给 create_app(workspace_manager=...)，服务就会把它当作内置 manager 一样使用。

如果后端需要一个后台 sweeper 来做空闲驱逐，或者需要跑长时间的准备逻辑，重写 __aenter__ / __aexit__ 来启停这些机制 —— DockerWorkspaceManager 的源码就是一个完整示例。

分布式部署

LocalWorkspaceManager 与 DockerWorkspaceManager 都是单节点的 —— workspace 状态活在跑服务进程的那台机器上。如果做水平扩容（负载均衡后挂多个智能体服务 worker 节点），同一个 workspace_id 的请求可能落到任意一个节点，而这两个 manager 都够不到其他节点上准备好的 workspace。分布式部署要用云托管的沙箱后端：

E2BWorkspaceManager —— 沙箱可以通过 metadata 跨节点寻址；缓存 miss 时服务直接重连已存在的沙箱即可。
OpenSandboxWorkspaceManager —— 沙箱携带 agentscope.workspace.id 这个 metadata 标签；缓存 miss 时任意服务副本都可以通过对 list_sandbox_infos(...) 过滤来重新挂接，完全不依赖宿主本地状态。
K8sWorkspaceManager —— Pod 与 PVC 都是集群级资源；集群内任意服务副本都可以按 workspace-id 派生出的名字重连到同一个 Pod。

本地开发用 LocalWorkspaceManager，单机生产用 DockerWorkspaceManager，一旦要横向扩容就切到 E2B、OpenSandbox 或 K8s。

​可选的实现

​接入智能体服务

​隔离粒度

​在智能体服务中的工作原理

​接线方式

​生命周期

​端到端请求流程

​核心 API

​自定义 Manager

​分布式部署

可选的实现

接入智能体服务

隔离粒度

在智能体服务中的工作原理

接线方式

生命周期

端到端请求流程

核心 API

自定义 Manager

分布式部署