LangChain Deep Agents Execute 工具与沙箱环境深度解析
本文深入分析了LangChain Deep Agents框架中的execute工具实现原理。该工具通过FilesystemMiddleware定义，支持执行Shell命令并返回结果。系统采用分层协议设计，将基础文件操作与沙箱命令执行能力分离。底层实现基于Python的subprocess.run()，支持超时控制、输出截断保护等特性，并通过环境变量和工作目录配置确保执行安全性。文章详细解析了工具参
·
前言
在 AI Agent 框架中，命令执行能力是实现自动化任务的关键特性。LangChain Deep Agents 提供了 execute 工具，使 Agent 能够在运行环境中执行 Shell 命令。本文将从源码层面深入分析 execute 工具的实现原理、后端架构设计以及沙箱环境的概念与实现方式。
一、Execute 工具概述
1.1 工具定义
execute 工具定义在 FilesystemMiddleware 中，其核心功能是执行 Shell 命令并返回执行结果。该工具的定义如下：
def _create_execute_tool(self) -> BaseTool:
"""创建 execute 工具"""
def sync_execute(
command: Annotated[str, "Shell command to execute."],
runtime: ToolRuntime[None, FilesystemState],
timeout: Annotated[int | None, "Optional timeout in seconds."] = None,
) -> str:
resolved_backend = self._get_backend(runtime)
result = resolved_backend.execute(command, timeout=timeout)
return result.output
return StructuredTool.from_function(
name="execute",
description=EXECUTE_TOOL_DESCRIPTION,
func=sync_execute,
)
1.2 工具参数说明
| 参数 | 类型 | 必需 | 说明 |
|---|---|---|---|
command |
str |
是 | 要执行的 Shell 命令字符串 |
timeout |
int | None |
否 | 命令执行超时时间（秒），默认使用后端配置 |
1.3 返回值结构
execute 工具的底层返回 ExecuteResponse 数据结构：
class ExecuteResponse(TypedDict):
"""命令执行响应结构"""
output: str
"""命令输出内容（stdout 和 stderr 合并）"""
exit_code: int
"""进程退出码（0 表示成功，非 0 表示失败）"""
truncated: bool
"""输出是否因超出大小限制而被截断"""
二、后端协议架构
2.1 协议层次设计
Deep Agents 采用分层协议设计，将文件操作与命令执行能力进行抽象：
┌─────────────────────────────────────────────────────────────┐
│ BackendProtocol（基础协议） │
│ ┌─────────────────────────────────────────────────────────┐│
│ │ 文件操作：read, write, ls, grep, glob, download_files ││
│ └─────────────────────────────────────────────────────────┘│
├─────────────────────────────────────────────────────────────┤
│ SandboxBackendProtocol（沙箱协议） │
│ ┌─────────────────────────────────────────────────────────┐│
│ │ 继承 BackendProtocol + execute() 方法 ││
│ └─────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────┘
2.2 协议定义
class BackendProtocol(abc.ABC):
"""可插拔存储后端协议（基础文件操作）"""
def read(self, path: str, *, offset: int = 0, limit: int = 100) -> str:
"""读取文件内容"""
raise NotImplementedError
def write(self, path: str, content: str, *, mode: str = "overwrite") -> str:
"""写入文件内容"""
raise NotImplementedError
def ls_info(self, path: str) -> list[FileInfo]:
"""列出目录内容"""
raise NotImplementedError
# ... 其他文件操作方法
class SandboxBackendProtocol(BackendProtocol):
"""沙箱后端协议（支持命令执行）"""
def execute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
"""执行 Shell 命令"""
raise NotImplementedError
三、Execute 底层实现原理
3.1 核心实现：subprocess.run()
execute 工具的底层实现基于 Python 标准库的 subprocess.run() 函数。以 LocalShellBackend 为例：
def execute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
"""在主机系统上直接执行 Shell 命令"""
effective_timeout = timeout if timeout is not None else self._default_timeout
try:
result = subprocess.run(
command,
check=False,
shell=True, # 使用系统 Shell 执行
capture_output=True,
text=True,
timeout=effective_timeout,
env=self._env,
cwd=str(self.cwd), # 工作目录
)
# 合并 stdout 和 stderr
output_parts = []
if result.stdout:
output_parts.append(result.stdout)
if result.stderr:
stderr_lines = result.stderr.strip().split("\n")
output_parts.extend(f"[stderr] {line}" for line in stderr_lines)
output = "\n".join(output_parts) if output_parts else "<no output>"
# 输出截断保护
truncated = False
if len(output) > self._max_output_bytes:
output = output[: self._max_output_bytes]
output += f"\n\n... Output truncated at {self._max_output_bytes} bytes."
truncated = True
return ExecuteResponse(
output=output,
exit_code=result.returncode,
truncated=truncated,
)
except subprocess.TimeoutExpired:
return ExecuteResponse(
output=f"Error: Command timed out after {effective_timeout} seconds.",
exit_code=124, # 标准超时退出码
truncated=False,
)
3.2 关键参数解析
| 参数 | 值 | 说明 |
|---|---|---|
shell |
True |
通过系统 Shell（如 /bin/sh）执行命令 |
capture_output |
True |
捕获 stdout 和 stderr |
text |
True |
以文本模式返回输出（而非字节） |
cwd |
self.cwd |
命令执行的工作目录 |
env |
self._env |
环境变量配置 |
3.3 执行流程图
┌─────────────────────────────────────────────────────────────┐
│ Agent 调用流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Agent │
│ │ │
│ │ execute(command="ls -la") │
│ ▼ │
│ FilesystemMiddleware │
│ │ │
│ │ _get_backend() → 获取后端实例 │
│ ▼ │
│ Backend.execute() │
│ │ │
│ │ subprocess.run(command, shell=True, ...) │
│ ▼ │
│ 系统 Shell (/bin/sh) │
│ │ │
│ │ 执行: /bin/sh -c "ls -la" │
│ ▼ │
│ 返回 ExecuteResponse │
│ │ │
│ │ output: "total 48\ndrwxr-xr-x ..." │
│ │ exit_code: 0 │
│ │ truncated: False │
│ ▼ │
│ Agent 获得执行结果 │
│ │
└─────────────────────────────────────────────────────────────┘
四、沙箱环境概念与实现
4.1 什么是沙箱环境
沙箱（Sandbox）是一种隔离的执行环境，用于在受控条件下运行不受信任的代码或命令。其核心目的是保护主机系统免受潜在的恶意操作影响。
沙箱环境的主要特征包括：
- 隔离性：与主机系统隔离，无法直接访问主机资源
- 可控性：可以限制资源使用（CPU、内存、网络等）
- 可恢复性：执行完成后可以轻松重置到初始状态
- 安全性：即使执行恶意命令，也不会影响主机系统
4.2 沙箱的常见实现方式
| 实现方式 | 隔离级别 | 性能开销 | 典型应用 |
|---|---|---|---|
| Docker 容器 | 进程级 | 低 | 开发环境、CI/CD |
| 虚拟机（VM） | 硬件级 | 高 | 高安全性场景 |
| 云沙箱服务 | 完全隔离 | 中 | 生产环境、多租户 |
| chroot | 文件系统级 | 极低 | 简单隔离 |
4.3 Deep Agents 后端类型对比
Deep Agents 提供了多种后端实现，以适应不同的安全需求和部署场景：
┌─────────────────────────────────────────────────────────────┐
│ 后端类型对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ LocalShellBackend│ │ BaseSandbox │ │ StateBackend│ │
│ │ (本地执行) │ │ (沙箱基类) │ │ (纯内存) │ │
│ └────────┬────────┘ └────────┬────────┘ └──────┬──────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 无沙箱隔离 │ │ 完全隔离 │ │ 不支持执行 │ │
│ │ 直接在主机 │ │ 在容器/VM中 │ │ 仅文件操作 │ │
│ │ 执行命令 │ │ 执行命令 │ │ 基于状态存储 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ 适用场景: 适用场景: 适用场景: │
│ - 本地开发 - 生产环境 - 测试/模拟 │
│ - 可信环境 - 多租户 - 无执行需求 │
│ - 快速原型 - 高安全性 - 纯文件操作 │
│ │
└─────────────────────────────────────────────────────────────┘
4.3.1 LocalShellBackend（本地 Shell 后端）
LocalShellBackend 直接在主机系统上执行命令，无任何沙箱隔离：
class LocalShellBackend(FilesystemBackend, SandboxBackendProtocol):
"""直接在主机系统上执行命令的后端
警告：命令使用 subprocess.run() 和 shell=True 直接执行，
没有沙箱、隔离或安全限制。命令以当前用户的完整权限运行。
"""
def __init__(
self,
root_dir: str | Path | None = None,
*,
default_timeout: int = 120,
max_output_bytes: int = 100_000,
env: dict[str, str] | None = None,
) -> None:
super().__init__(root_dir=root_dir)
self._default_timeout = default_timeout
self._max_output_bytes = max_output_bytes
self._env = {**os.environ, **(env or {})}
安全警告：使用 LocalShellBackend 时，Agent 可以执行任意命令，包括：
- 访问文件系统中的任何文件
- 执行任何程序或脚本
- 建立网络连接
- 修改系统配置
- 安装软件包
因此，官方强烈建议在使用 LocalShellBackend 时启用 Human-in-the-Loop（HITL）中间件。
4.3.2 BaseSandbox（沙箱基类）
BaseSandbox 是一个精心设计的抽象基类，其核心理念是：只需实现 execute() 方法，即可自动获得所有文件操作能力。
class BaseSandbox(SandboxBackendProtocol):
"""沙箱后端基类
设计理念：所有文件操作都通过 execute() 实现，
子类只需实现 execute() 方法即可获得完整的文件系统操作能力。
"""
@abc.abstractmethod
def execute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
"""执行 Shell 命令（子类必须实现）"""
raise NotImplementedError
def read(self, path: str, *, offset: int = 0, limit: int = 100) -> str:
"""通过 execute 实现文件读取"""
# 使用 sed 命令实现分页读取
start_line = offset + 1
end_line = offset + limit
command = f"sed -n '{start_line},{end_line}p' {shlex.quote(path)}"
result = self.execute(command)
return result.output
def write(self, path: str, content: str, *, mode: str = "overwrite") -> str:
"""通过 execute 实现文件写入"""
escaped_content = content.replace("'", "'\\''")
if mode == "overwrite":
command = f"echo '{escaped_content}' > {shlex.quote(path)}"
else: # append
command = f"echo '{escaped_content}' >> {shlex.quote(path)}"
result = self.execute(command)
return "OK" if result.exit_code == 0 else result.output
def ls(self, path: str = ".") -> list[str]:
"""通过 execute 实现目录列表"""
command = f"ls -1 {shlex.quote(path)}"
result = self.execute(command)
return result.output.strip().split("\n") if result.output.strip() else []
def grep(self, pattern: str, path: str, *, context: int = 2) -> str:
"""通过 execute 实现文本搜索"""
command = f"grep -n -C {context} {shlex.quote(pattern)} {shlex.quote(path)}"
result = self.execute(command)
return result.output
这种设计的优势在于：
- 最小实现原则：子类只需实现一个方法
- 一致性保证：所有文件操作都通过相同的执行路径
- 灵活性：可以轻松适配不同的沙箱实现（Docker、VM、云服务等）
4.3.3 StateBackend（状态后端）
StateBackend 是一个纯内存实现，不支持命令执行，仅提供基于 LangGraph 状态的文件操作：
class StateBackend(BackendProtocol):
"""基于 LangGraph 状态的后端（不支持 execute）"""
# 注意：不继承 SandboxBackendProtocol，因此没有 execute 方法
4.4 沙箱环境是否需要安装 Python？
这是一个常见的误解。答案是：沙箱环境不需要安装 Python。
需要区分两个不同的环境：
┌─────────────────────────────────────────────────────────────┐
│ 主机环境（运行 Deep Agents） │
│ │
│ Python 进程 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ subprocess.run("ls -la", shell=True) │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ /bin/sh -c "ls -la" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 这里需要 Python（运行 Agent 框架） │
└─────────────────────────────────────────────────────────────┘
│
│ 如果是远程沙箱
▼
┌─────────────────────────────────────────────────────────────┐
│ 沙箱环境（Docker/VM/云服务） │
│ │
│ 只需要 Shell（/bin/sh 或 /bin/bash） │
│ 不需要 Python！ │
│ │
│ 可以执行： │
│ - ls, cat, grep, find（系统命令） │
│ - node script.js（如果安装了 Node.js） │
│ - python script.py（如果安装了 Python） │
│ - 任何 Shell 命令 │
└─────────────────────────────────────────────────────────────┘
关键理解：
| 环境 | 是否需要 Python | 原因 |
|---|---|---|
| 主机（运行 Agent） | 需要 | Deep Agents 是 Python 框架，subprocess.run() 是 Python 标准库 |
| 沙箱（执行命令） | 不需要 | execute 执行的是 Shell 命令，只需要基本的 Shell 环境 |
如果需要在沙箱中执行 Python 脚本（如 python script.py），则沙箱需要安装 Python。但对于基本的文件操作和系统命令，只需要 Shell 即可。
五、自定义沙箱实现示例
5.1 Docker 沙箱实现
以下是一个基于 Docker 的沙箱实现示例：
import docker
from deepagents.backends.sandbox import BaseSandbox
from deepagents.backends.protocol import ExecuteResponse
class DockerSandbox(BaseSandbox):
"""基于 Docker 容器的沙箱实现"""
def __init__(
self,
image: str = "ubuntu:22.04",
*,
timeout: int = 120,
memory_limit: str = "512m",
cpu_quota: int = 50000,
) -> None:
self.client = docker.from_env()
self.image = image
self.timeout = timeout
self.memory_limit = memory_limit
self.cpu_quota = cpu_quota
self._container = None
def _ensure_container(self) -> docker.models.containers.Container:
"""确保容器已启动"""
if self._container is None:
self._container = self.client.containers.run(
self.image,
command="sleep infinity",
detach=True,
mem_limit=self.memory_limit,
cpu_quota=self.cpu_quota,
remove=True,
)
return self._container
def execute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
"""在 Docker 容器中执行命令"""
container = self._ensure_container()
effective_timeout = timeout or self.timeout
try:
exit_code, output = container.exec_run(
cmd=["sh", "-c", command],
demux=False,
)
return ExecuteResponse(
output=output.decode("utf-8") if output else "<no output>",
exit_code=exit_code,
truncated=False,
)
except Exception as e:
return ExecuteResponse(
output=f"Error: {e!s}",
exit_code=1,
truncated=False,
)
def cleanup(self) -> None:
"""清理容器资源"""
if self._container:
self._container.stop()
self._container = None
5.2 使用自定义沙箱
from deepagents import create_deep_agent
# 创建 Docker 沙箱实例
sandbox = DockerSandbox(
image="python:3.11-slim", # 如果需要执行 Python 脚本
memory_limit="1g",
timeout=300,
)
# 创建使用沙箱的 Agent
agent = create_deep_agent(
model="gpt-4",
backend=sandbox,
)
# Agent 现在可以安全地执行命令
# 所有命令都在隔离的 Docker 容器中运行
5.3 云沙箱服务集成
Deep Agents 还支持与云沙箱服务集成，如 Modal、Daytona、Runloop 等：
# Modal 集成示例
from deepagents.integrations.modal import ModalSandbox
sandbox = ModalSandbox(
app_name="my-agent-sandbox",
image="python:3.11",
)
# Daytona 集成示例
from deepagents.integrations.daytona import DaytonaSandbox
sandbox = DaytonaSandbox(
workspace_id="my-workspace",
project_id="my-project",
)
六、安全性考虑
6.1 LocalShellBackend 的风险
使用 LocalShellBackend 时，Agent 具有与运行用户相同的权限，存在以下风险：
| 风险类型 | 示例 | 影响 |
|---|---|---|
| 文件系统访问 | cat /etc/passwd |
读取敏感系统文件 |
| 数据删除 | rm -rf / |
删除重要数据 |
| 网络访问 | curl malicious.com |
数据泄露、恶意下载 |
| 权限提升 | sudo ... |
获取更高权限 |
| 资源耗尽 | `:(){ : | :& };:` |
6.2 安全最佳实践
6.2.1 启用 Human-in-the-Loop（HITL）
from deepagents import create_deep_agent
from deepagents.middleware import HITLMiddleware
agent = create_deep_agent(
model="gpt-4",
middleware=[
HITLMiddleware(
tools=["execute"], # 对 execute 工具启用人工审批
),
],
)
6.2.2 使用命令白名单
class AllowlistBackend(LocalShellBackend):
"""带命令白名单的后端"""
ALLOWED_COMMANDS = {"ls", "cat", "grep", "find", "head", "tail"}
def execute(self, command: str, **kwargs) -> ExecuteResponse:
# 提取命令名称
cmd_name = command.split()[0] if command else ""
if cmd_name not in self.ALLOWED_COMMANDS:
return ExecuteResponse(
output=f"Error: Command '{cmd_name}' is not allowed.",
exit_code=1,
truncated=False,
)
return super().execute(command, **kwargs)
6.2.3 使用沙箱环境
对于生产环境，强烈建议使用沙箱后端：
# 推荐：使用 Docker 沙箱
agent = create_deep_agent(
model="gpt-4",
backend=DockerSandbox(image="ubuntu:22.04"),
)
# 不推荐：直接使用 LocalShellBackend（仅限开发环境）
agent = create_deep_agent(
model="gpt-4",
backend=LocalShellBackend(),
)
6.3 输出保护机制
LocalShellBackend 实现了多层输出保护：
# 1. 输出大小限制
if len(output) > self._max_output_bytes:
output = output[: self._max_output_bytes]
output += f"\n\n... Output truncated at {self._max_output_bytes} bytes."
truncated = True
# 2. 超时保护
try:
result = subprocess.run(..., timeout=effective_timeout)
except subprocess.TimeoutExpired:
return ExecuteResponse(
output=f"Error: Command timed out after {effective_timeout} seconds.",
exit_code=124,
truncated=False,
)
# 3. stderr 标记
if result.stderr:
stderr_lines = result.stderr.strip().split("\n")
output_parts.extend(f"[stderr] {line}" for line in stderr_lines)
七、FilesystemMiddleware 工具集
execute 工具是 FilesystemMiddleware 提供的工具之一。该中间件还提供了一系列文件操作工具：
7.1 工具列表
| 工具名称 | 功能 | 依赖后端 |
|---|---|---|
read_file |
读取文件内容（支持分页） | BackendProtocol |
write_file |
写入文件内容 | BackendProtocol |
list_directory |
列出目录内容 | BackendProtocol |
grep |
在文件中搜索文本 | BackendProtocol |
glob |
按模式匹配文件 | BackendProtocol |
execute |
执行 Shell 命令 | SandboxBackendProtocol |
7.2 工具与后端的关系
┌─────────────────────────────────────────────────────────────┐
│ FilesystemMiddleware │
│ ┌─────────────────────────────────────────────────────────┐│
│ │ 工具层 ││
│ │ read_file | write_file | list_directory | grep | glob ││
│ │ execute ││
│ └─────────────────────────────────────────────────────────┘│
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐│
│ │ 后端层 ││
│ │ ││
│ │ BackendProtocol SandboxBackendProtocol ││
│ │ (文件操作) (文件操作 + execute) ││
│ │ │ │ ││
│ │ ▼ ▼ ││
│ │ StateBackend LocalShellBackend ││
│ │ FilesystemBackend BaseSandbox 子类 ││
│ │ StoreBackend (Docker/VM/云服务) ││
│ └─────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────┘
7.3 条件性工具注册
FilesystemMiddleware 根据后端类型条件性地注册 execute 工具：
class FilesystemMiddleware(AgentMiddleware):
"""文件系统中间件"""
def get_tools(self) -> list[BaseTool]:
"""获取可用工具列表"""
tools = [
self._create_read_file_tool(),
self._create_write_file_tool(),
self._create_list_directory_tool(),
self._create_grep_tool(),
self._create_glob_tool(),
]
# 只有当后端支持 execute 时才添加该工具
if isinstance(self._backend, SandboxBackendProtocol):
tools.append(self._create_execute_tool())
return tools
八、总结
本文深入分析了 LangChain Deep Agents 的 execute 工具和沙箱环境实现。核心要点如下：
-
execute 工具底层原理：基于 Python 标准库的
subprocess.run()实现，通过系统 Shell 执行命令 -
后端协议设计：采用分层协议架构，
BackendProtocol提供基础文件操作，SandboxBackendProtocol扩展了命令执行能力 -
沙箱环境：
- 沙箱是隔离的执行环境，用于安全地运行不受信任的命令
- 沙箱环境不需要安装 Python，只需要基本的 Shell 环境
- Deep Agents 支持多种沙箱实现：Docker、VM、云服务等
-
BaseSandbox 设计理念：只需实现
execute()方法，即可自动获得所有文件操作能力 -
安全最佳实践：
- 生产环境应使用沙箱后端
- 启用 Human-in-the-Loop 中间件
- 考虑使用命令白名单
通过合理选择后端类型和安全策略，开发者可以在保证安全性的前提下，充分利用 execute 工具的强大能力。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
13
0- 0
已为社区贡献26条内容
所有评论(0)