在Deep Agents中FilesystemMiddleware（文件系统中间件）扮演着外部存储器和上下文管理器的双重角色，建立在一个利用BackendProtocol协议抽象的文件系统之上。具有如下定义的BackendProtocol定义了一套标准化的接口，使得Agent无需关心底层存储是本地磁盘、云端S3、数据库还是内存，都能以统一的方式进行文件操作。

class BackendProtocol(abc.ABC):  
    def ls(self, path: str) -> "LsResult"
    async def als(self, path: str) -> "LsResult"

    def read(
        self,
        file_path: str,
        offset: int = 0,
        limit: int = 2000,
    ) -> ReadResult
    async def aread(
        self,
        file_path: str,
        offset: int = 0,
        limit: int = 2000,
    ) -> ReadResult

    def grep(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
    ) -> "GrepResult":

    async def agrep(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
    ) -> "GrepResult"

    def glob(self, pattern: str, path: str = "/") -> "GlobResult"
    async def aglob(self, pattern: str, path: str = "/") -> "GlobResult"

    def write(
        self,
        file_path: str,
        content: str,
    ) -> WriteResult

    async def awrite(
        self,
        file_path: str,
        content: str,
    ) -> WriteResult:
        """Async version of write."""
        return await asyncio.to_thread(self.write, file_path, content)

    def edit(
        self,
        file_path: str,
        old_string: str,
        new_string: str,
        replace_all: bool = False, 
    ) -> EditResult:

    async def aedit(
        self,
        file_path: str,
        old_string: str,
        new_string: str,
        replace_all: bool = False,
    ) -> EditResult

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]
    async def aupload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]
    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]
    async def adownload_files(self, paths: list[str]) -> list[FileDownloadResponse]

方法说明如下：

• ls/als: 列出目录内容，返回文件和子目录的列表；
• read/aread: 读取文件内容，支持offset和limit参数实现分页读取
• grep/agrep: 在指定路径下搜索包含特定文本模式的文件，返回匹配结果；
• glob/aglob: 根据模式匹配查找文件，支持通配符；
• write/awrite: 将内容写入指定文件，支持覆盖或追加模式；
• edit/aedit: 在文件中查找特定字符串并替换，支持替换所有匹配项；
• upload_files/aupload_files: 允许Agent上传文件内容，返回上传结果；
• download_files/adownload_files: 允许Agent下载文件内容，返回下载结果；

1. StateBackend

StateBackend是FilesystemMiddleware的默认后端实现，它将文件系统抽象为Agent状态的一部分，直接利用Agent的状态成员（通道）来存储文件内容。由于文件内容直接存储在Agent状态中，因此StateBackend不需要依赖外部存储，所以适合于小规模数据和快速迭代的场景。这种存储方式也赋予了它如下两个特征：

• 瞬时性（Ephemeral）：文件只存在于当前的thread_id中。如果你开启一个新对话，之前的磁盘内容会全部消失；
• 自动回滚：因为利用了LangGraph的Checkpointer，如果Agent的某一步运行失败，文件系统的状态会自动回滚到上一个成功的Checkpoint；

我们在调用__init__方法创建StateBackend对象时，需要指定表示工具运行时的ToolRuntime对象，因为Agent当前的状态绑定在它的state字段上。file_format参数用于指定文件内容的存储格式，v2为最新格式，v1为兼容旧版本的格式。StateBackend会根据指定的文件格式将文件内容进行相应的序列化和反序列化处理，以确保与Agent状态的兼容性和数据的一致性。

class StateBackend(BackendProtocol):
    def __init__(
        self,
        runtime: ToolRuntime,
        *,
        file_format: FileFormat = "v2",
    ) -> None
FileFormat = Literal["v1", "v2"]

具有如下定义的write方法的实现体现了StateBackend存储的本质。FilesystemMiddleware的state_schema字段返回的状态类型是FilesystemState，其唯一的字段files返回一个字典用于模拟存储的文件。该字典的Key为文件的路径，Value为一个FileData对象，承载了文件的内容、编码格式、创建时间和修改时间等元信息。从针对该字段的注解可知，它对应的通道类型为BinaryOperatorAggregate，reduer函数_file_data_reducer体现了文件系统状态的更新规则：

• 当新的文件数据被写入时，如果文件路径已经存在，则覆盖原有数据；
• 如果提供的FileData为None，则从状态中移除该文件；否则，添加新的文件数据到状态中；

当write方法被调用时，StateBackend首先检查要写入的文件路径是否已经存在于状态中，如果存在则返回一个错误，提示用户无法覆盖现有文件。否则，它会创建一个新的FileData对象来存储文件内容，并通过返回一个WriteResult对象来更新状态中的files字段，从而将新文件添加到文件系统中。也就是说write方法仅限于创建新文件，如果需要修改文件内容，需要调用edit方法。

class StateBackend(BackendProtocol):
    def write(
        self,
        file_path: str,
        content: str,
    ) -> WriteResult:
        files = self.runtime.state.get("files", {})

        if file_path in files:
            return WriteResult(error=f"Cannot write to {file_path} because it already exists. Read and then make an edit, or write to a new path.")

        new_file_data = create_file_data(content)
        return WriteResult(path=file_path, files_update={file_path: self._prepare_for_storage(new_file_data)})

class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT])
    state_schema = FilesystemState
    
class FilesystemState(AgentState):
    files: Annotated[NotRequired[dict[str, FileData]], _file_data_reducer]

class FileData(TypedDict):
    content: str
    encoding: str
    created_at: str
    modified_at: str

def _file_data_reducer(left: dict[str, FileData] | None, right: dict[str, FileData | None]) -> dict[str, FileData]:
    if left is None:
        return {k: v for k, v in right.items() if v is not None}

    result = {**left}
    for key, value in right.items():
        if value is None:
            result.pop(key, None)
        else:
            result[key] = value
    return result

下面的程序演示了如何对StateBackend维护的文件进行读写。我们首先定义了一个继承自AgentState的State类，并按照上面介绍的FilesystemState的方式在其中定义了一个files字段。我们定义了两个工具函数read和write，分别用于读取和写入文件。

from typing import Annotated, NotRequired
from langchain.agents import create_agent, AgentState
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage
from langchain.tools import ToolRuntime
from langchain_core.runnables import RunnableConfig
from deepagents.backends.protocol import FileData
from deepagents.backends.state import StateBackend
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import asyncio

from deepagents.middleware.filesystem import _file_data_reducer

load_dotenv()

class State(AgentState):
  files: Annotated[NotRequired[dict[str, FileData]], _file_data_reducer]

@tool
async def write(runtime:ToolRuntime, config: RunnableConfig):
    """A fake tool to test file system interactions"""
    backend = StateBackend(runtime=runtime)
    thread_id = config.get("configurable", {}).get("thread_id", "N/A")
    path = "./test.txt"
    result = await backend.aglob(path)
    if len(result.matches or []) == 0:
        result = await backend.awrite(path, thread_id)
        return Command(
            update={
                "files": result.files_update,
                "messages": [
                    ToolMessage(
                        content=f"Create a new file {result.path} and write data to it.",
                        tool_call_id=runtime.tool_call_id,
                    )
                ],
            }
        )
    else:
        return "File already exists, no need to write."

@tool
async def read(runtime:ToolRuntime, config: RunnableConfig):
    """read the file created by `write` tool"""
    backend = StateBackend(runtime=runtime)
    path = "./test.txt"    
    result = await backend.aread(path)
    return str(result.file_data)

agent = create_agent(
    model=ChatOpenAI(model= "gpt-5.2-chat"),
    tools=[write, read],
    checkpointer=MemorySaver(),
    state_schema=State,
)
 
prompt = "Write to a file with `write` tool and then read from it with `read` tool"
async def write_and_read(thread_id: str):
    config:RunnableConfig ={"configurable": {"thread_id": thread_id}}
    result = await agent.ainvoke(input= {"messages": [{"role": "user", "content": prompt}]}, config=config)
    print(result["messages"][-1].content) 

async def main():
    await write_and_read("thread-1")
    await write_and_read("thread-2")

asyncio.run(main())

在write工具函数中，我们根据转入的ToolRuntime对象创建了一个StateBackend对象，并调用aglob方法判断指定路径（./test.txt）的文件是否存在。如果文件不存在，我们才会创建一个新的文件，并调用awrite方法将当前会话的thread_id作为内容写入文件。read工具函数采用相同的方法创建出StateBackend对象，并调用aread方法读取之前写入的文件内容。

通过注册这两个工具、采用State作为状态类型并指定Checkpointer，我们调用create_agent函数创建了一个Agent，并在write_and_read函数指定不同的thread_id调用Agent来测试文件系统的隔离性。由于StateBackend将文件系统状态绑定在Agent的状态上，并且状态是基于thread_id进行隔离的，所以写入的文件仅限于当前对话可见。如下的输出体现了这一点：

✅ **File write and read completed successfully!**

Here’s what happened:

1. **Write step**  
   - A file named `test.txt` was created.
   - Data was written to it.

2. **Read step**  
   - The contents of the file were read back.
   - The file contains:


thread-1

📄 **File details**
- Encoding: UTF-8  
- Created at: 2026-04-18 14:59:04 UTC  
- Last modified: 2026-04-18 14:59:04 UTC  

If you’d like, I can overwrite the file, append new content, or work with multiple files.    

✅ **File Write and Read Completed**

Here’s what I did using the provided tools:

1. **Wrote to a file** using the `write` tool  
   - Created a file named `test.txt`
   - Wrote content into it

2. **Read from the file** using the `read` tool  
   - Successfully retrieved the contents

📄 **File contents:**

thread-2

If you’d like to write specific content, append data, or work with multiple files, just tell me!

2. FilesystemBackend

FilesystemBackend是Deep Agents中最危险也最强大的后端存储。与StateBackend不同，FilesystemBackend映射的是实实在在宿主机的目录。FilesystemBackend直接利用pathlib操作物理文件系统，使得Agent能够直接参与到真实的软件开发流程中，比如修改项目代码、读取本地配置、查看构建日志等。FilesystemBackend的__init__方法定义如下。

class FilesystemBackend(BackendProtocol):
    def __init__(
        self,
        root_dir: str | Path | None = None,
        virtual_mode: bool | None = None,  # noqa: FBT001
        max_file_size_mb: int = 10,
    ) -> None

三个字段说明如下：

• root_dir：指定一个根目录，默认将当前工作目录作为根路径；
• virtual_mode：是否启用虚拟模式
  • True：它将root_dir映射为Agent眼中的根目录/。通过对_resolve_path方法拦截确保Agent只能访问root_dir目录及其子目录下的文件。但这只是路径过滤，不是真正的系统隔离；
  • False：Agent给出的绝对路径（如/etc/passwd）会被直接访问。相对路径（..）可以轻松跳出项目根目录。这是默认选项，仅限本地开发环境使用；
• max_file_size_mb：限制Agent可以读写的单个文件的最大大小，默认10MB，防止Agent处理过大的文件导致性能问题；

FilesystemBackend是CLI模式下的灵魂。它让DeepAgent变成了一个会用终端的程序员。其代码实现的核心难点在于路径的合法性校验（Sanitization），即如何在给予Agent灵活性的同时，尽可能锁住它的视野范围。

3. StoreBackend

StoreBackend是Deep Agents中实现长期记忆的核心后端存储。与StateBackend不同，它利用LangGraph的BaseStore实现了跨线程、跨对话的持久化存储。它是基于**命名空间（Namespace）**的键值对存储。对于Agent来说，它看起来像文件系统，但底层通常对应数据库（如PostgreSQL）。

class StoreBackend(BackendProtocol):    
    def __init__(
        self,
        runtime: ToolRuntime,
        *,
        namespace: NamespaceFactory | None = None,
        file_format: FileFormat = "v2",
    ) -> None
NamespaceFactory = Callable[[BackendContext[Any, Any]], tuple[str, ...]]

上面给出了StoreBackend的__init__方法的定义。它利用输入参数runtime来提供作为存储引擎的BaseStore实例，并通过namespace参数来定义存储的命名空间。与StateBackend一样，StoreBackend也支持file_format参数来指定数据的存储格式，以确保与Agent状态的兼容性。采用BaseStore作为存储引擎，赋予了StoreBackend如下的特性：

• 持久性：对话结束后，文件依然存在。下次对话时，同一个assistant_id或user_id的Agent可以重新读取；
• 隔离性：通过灵活的namespace设计，确保不同用户或不同Agent之间的数据互不干扰；

4. SandboxBackend

SandboxBackend是Deep Agents中一个特殊的后端存储，它提供了一个受限的沙箱环境，允许Agent执行Shell命令和脚本。它通过在受控环境中运行命令来扩展Agent的能力，使其能够进行系统级操作，如文件管理、进程监控、网络请求等。这类存储通过如下这个SandboxBackendProtocol协议来定义，它派生于BackendProtocol，并增加了execute和aexecute方法用于执行Shell命令，id字段用于标识当前沙箱存储的实例。

class SandboxBackendProtocol(BackendProtocol):
    @property
    def id(self) -> str:

    def execute(
        self,
        command: str,
        *,
        timeout: int | None = None,
    ) -> ExecuteResponse:

    async def aexecute(
        self,
        command: str,
        *,
        timeout: int | None = None,  # noqa: ASYNC109
    ) -> ExecuteResponse

如下这个BaseSandbox类是SandboxBackendProtocol的一个抽象实现，它定义了execute和aexecute方法的接口，但具体的执行逻辑需要由子类来实现。BaseSandbox为沙箱环境提供了统一的接口，使得不同的沙箱（如基于Docker、基于远程容器服务等）都能通过继承BaseSandbox来实现自己的执行逻辑，同时保持与Agent的兼容性。

class BaseSandbox(SandboxBackendProtocol, ABC):
    @abstractmethod
    def execute(
        self,
        command: str,
        *,
        timeout: int | None = None,
    ) -> ExecuteResponse

    def read(
        self,
        file_path: str,
        offset: int = 0,
        limit: int = 2000,
    ) -> ReadResult

    def write(
        self,
        file_path: str,
        content: str,
    ) -> WriteResult

    def edit(
        self,
        file_path: str,
        old_string: str,
        new_string: str,
        replace_all: bool = False,  # noqa: FBT001, FBT002
    ) -> EditResult

    def grep(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
    ) -> GrepResult

    def glob(self, pattern: str, path: str = "/") -> GlobResult

    @property
    @abstractmethod
    def id(self) -> str

    @abstractmethod
    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]

    @abstractmethod
    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]

在Deep Agents的设计哲学中，SandboxBackend是解决代码执行安全性与环境隔离性的终极方案。如果说FilesystemBackend是直接在你的主机上动刀，那么SandboxBackend就是给Agent准备了一个一次性的、完全隔离的实验室。SandboxBackend通常与远程容器服务（如E2B, Modal, Daytona, 或 Docker）对接。它不仅提供一个虚拟的文件系统，还提供一个完整的操作系统环境（通常是 Linux）。
它是唯一一个默认必须实现 execute(command: str) 方法的存储，允许Agent运行 pip install、python main.py 或 npm test 等命令。

每个Agent任务启动时，后端会动态拉起一个隔离容器。Agent对文件的修改、安装的依赖、运行产生的临时文件都保存在这个容器内。由于容器与宿主机完全隔离，即使Agent被诱导执行 rm -rf /，损坏的也只是那个临时的沙箱。Deep Agents框架通常通过适配器支持以下服务：

• E2B (Excels in 2-way Binding)：目前最流行的AI沙箱，支持毫秒级启动Python/JS环境；
• Modal：适合高性能计算和需要大规模GPU/内存的沙箱任务；
• Docker (Local)：在本地开发者机器上通过容器提供一定程度的隔离；

4.1 LocalShellBackend

LocalShellBackend同时继承了FilesystemBackend和SandboxBackendProtocol。它是Deep Agents框架中的一个核心后端组件，赋予Agent本地系统最高控制权的工具箱。它让Agent既能像操作文件夹一样读写文件，又能像程序员一样打开终端执行命令。下面给出了LocalShellBackend的__init__方法的定义。

class LocalShellBackend(FilesystemBackend, SandboxBackendProtocol):
    def __init__(
        self,
        root_dir: str | Path | None = None,
        *,
        virtual_mode: bool | None = None,
        timeout: int = DEFAULT_EXECUTE_TIMEOUT,
        max_output_bytes: int = 100_000,
        env: dict[str, str] | None = None,
        inherit_env: bool = False,
    ) -> None
DEFAULT_EXECUTE_TIMEOUT = 120

参数说明如下：

• root_dir：指定一个根目录，默认将当前工作目录作为根路径；
• virtual_mode：是否启用虚拟模式。开启虚拟模式不仅仅可以防止Agent访问root_dir以外的文件，还能避免Agent执行一些外部命令；
• timeout：命令执行的超时时间，默认120秒，防止Agent执行长时间运行的命令导致资源占用；
• max_output_bytes：限制命令输出的最大字节数，默认100KB，防止Agent产生过大的输出导致性能问题；
• env：指定命令执行时的环境变量的字典，允许用户为Agent定制特定的运行环境；
• inherit_env：是否继承当前进程的环境变量，默认为False。如果设置为True，Agent执行命令时将能够访问宿主机的环境变量，这可能会带来安全风险，需谨慎使用；

LocalShellBackend实现的execute方法以子进程的方式在宿主机上执行Shell命令。它利用Python的subprocess模块来启动一个新的进程，并执行Agent传入的命令。当命令执行完成后，execute方法会收集命令的标准输出、标准错误和退出状态，并将这些信息封装在一个ExecuteResponse对象中返回给Agent。虽然实现了SandboxBackendProtocol，但由于它直接操作宿主机的Shell，因此它没有真正的环境隔离能力，适合于本地开发和测试环境使用，不建议在生产环境中使用。

LocalShellBackend唯一提供的安全就是利用虚拟模式（virtual_mode=True）来限制Agent只能访问指定根目录（root_dir）及其子目录下的文件。但它无法防止Agent执行诸如rm -rf /等破坏性命令，因此在使用时需要格外小心，建议仅在受控的本地环境中使用。它本质上是一个远程代码执行 (RCE) 的温床：

• 权限滥用：AI可以删除你的硬盘数据、修改系统配置；
• 隐私泄露：AI可以读取你的 SSH 密钥、.env 配置文件并发送到外网；
• 不可逆性：一旦执行，无法撤回；

LocalShellBackend是一个性能全开的工具，它让AI拥有了操作你电脑的“手”和“脚”，但如果你不加监管，它也可能拆了你的家。如果你打算使用这个类型，请务必遵循代码注释中的建议：

• 开启 HITL (Human-in-the-Loop)：在AI执行任何Shell命令前，必须由人工点击“允许”；
• 限制超时和输出：代码中默认了120秒超时和100KB输出限制，防止AI跑死循环或刷爆内存；
• 考虑替代方案：如果是生产环境，应该使用基于Docker的沙箱后端（如类中提到的 BaseSandbox 扩展）；

下面的演示程序展示了如何使用LocalShellBackend来执行Shell命令。我们定义了一个execute工具，它利用LocalShellBackend的aexecute方法来运行传入的命令，并将结果返回给Agent。我们创建了一个Agent并注册了这个工具，然后通过调用Agent来执行一系列Shell命令，包括创建目录、写入文件、列出目录内容和运行Python脚本。

from langchain.agents import create_agent
from langchain_core.tools import tool
from deepagents.backends.local_shell import LocalShellBackend
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import asyncio

load_dotenv()

executor = LocalShellBackend(virtual_mode=True)

@tool
async def execute(command: str):
    """A fake tool to test file system interactions"""
    result = await executor.aexecute(command)
    parts = [result.output]

    if result.exit_code is not None:
        status = "succeeded" if result.exit_code == 0 else "failed"
        parts.append(f"\n[Command {status} with exit code {result.exit_code}]")

    if result.truncated:
        parts.append("\n[Output was truncated due to size limits]")

    return "".join(parts)

agent = create_agent(
    model=ChatOpenAI(model= "gpt-5.2-chat"),
    tools=[execute],
)
 
prompt = """\
Execute the following shell command and return the output:
    
    ```shell
    mkdir -p test_dir && cd test_dir
    echo print(\"Hello World\") > hello.py    
    ls -al
    cat hello.py
    ```
"""

async def main():
   result = await agent.ainvoke(input= {"messages": [{"role": "user", "content": prompt}]})
   result["messages"][-1].pretty_print()
asyncio.run(main())

输出：

================================== Ai Message ==================================

Here is the output of executing the requested commands (Windows `cmd` equivalent was used to ensure successful execution):

```
 Volume in drive C has no label.
 Volume Serial Number is XXXX-XXXX

 Directory of C:\...\test_dir

04/19/2026  12:00 PM    <DIR>          .
04/19/2026  12:00 PM    <DIR>          ..
04/19/2026  12:00 PM                20 hello.py
               1 File(s)             20 bytes
               2 Dir(s)  XXX,XXX,XXX bytes free
```

Contents of `hello.py`:
```python
print("Hello World")
```

4.2 LangSmithSandbox

LangSmithSandbox是一个适配器实现，它将LangSmith的沙箱环境封装为一个符合SandboxBackendProtocol接口的后端。LangSmithSandbox是对一个langsmith.sandbox.Sandbox实例的包装，它利用此对象重写了基类BaseSandbox的execute、write、download_files和upload_files方法，并在此基础上做了一些异常处理。

class LangSmithSandbox(BaseSandbox):
    def __init__(self, sandbox: langsmith.sandbox.Sandbox) -> None
    def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse
    def write(self, file_path: str, content: str) -> WriteResult
    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]
    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]

5. CompositeBackend

CompositeBackend是 DeepAgents里的一个文件路由中心。它的核心逻辑就像是在电脑上挂载了多个磁盘分区：我们可以把不同的存储挂载到不同的路径下，然后通过一个统一的接口来访问它们。CompositeBackend的作用就是把这些五花八门的存储方式，统一伪装成一个虚拟文件系统。下面给出了CompositeBackend的__init__方法的定义。

class CompositeBackend(BackendProtocol):
    def __init__(
        self,
        default: BackendProtocol | StateBackend,
        routes: dict[str, BackendProtocol],
    ) -> None

两个参数说明如下：

• default：默认后端，当访问的路径没有匹配到任何路由时，就会使用这个后端来处理请求。它必须是一个BackendProtocol的实例，或者是StateBackend的实例（StateBackend作为默认后端时，可以利用Agent状态来存储未路由的文件）；
• routes：一个字典，键是路径前缀，值是对应的BackendProtocol实例。

CompositeBackend的核心功能是根据访问的文件路径来路由请求。当调用相应方法时，CompositeBackend会检查请求的路径，并根据routes中定义的路由规则将请求转发到对应的后端进行处理。如果请求的路径没有匹配到任何路由规则，则会使用默认兜底的后端存储来处理。通过这种方式。CompositeBackend采用最长匹配原则：如果存在 /memories/ 和 /memories/archive/ 两个路由，它会优先匹配更长的路径，确保路由的精确性。下面的程序演示了CompositeBackend的基本用法。

local_storage = StateBackend(runtime=None) 
memory_storage = StoreBackend(runtime=None)

composite = CompositeBackend(
    default=local_storage,
    routes={
        "/memories/": memory_storage,
        "/logs/": local_storage
    }
)