ADK Python 环境抽象实战指南BaseEnvironment 接口与 LocalEnvironment 本地子进程实现【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python本文是 ADK PythonAgent Development Kit中环境Environment体系的核心技术指南围绕BaseEnvironment接口与LocalEnvironment本地实现展开从为什么需要这一层抽象出发讲解环境生命周期、ExecutionResult返回约定、配置参数语义、自定义环境编写方法以及多工具集共享同一环境时的注意事项。读完本文你将能够在自己的 Agent 中挂载EnvironmentToolset/SkillToolset让模型在一个受控工作目录中执行 shell 命令、读写文件并能判断何时该切换到 E2B、Daytona 这类远程沙箱环境。为什么要有一层 Environment 抽象ADK 中有两个功能都需要一个能放文件、能跑命令的地方环境工具集EnvironmentToolset为 Agent 提供Execute、ReadFile、WriteFile、EditFile四个工具让模型可以运行脚本、阅读报错、修改文件技能工具集SkillToolset当技能Skill附带可执行脚本时脚本同样需要在某个工作目录中运行。这两者都不关心底层的子进程、沙箱或远程容器细节因此它们统一依赖BaseEnvironment接口只调用三个方法execute、read_file、write_file且全部相对于该环境的working_dir。正是这层间接性构成了该抽象存在的全部理由同一份 Agent 定义开发阶段可以在本地子进程上运行生产阶段可以切换到隔离的远程沙箱两者之间只差一个构造参数。ADK 内置了本地实现LocalEnvironment位于google.adk.environment同时也通过google.adk.integrations提供了 E2B 环境 与 Daytona 环境的远程沙箱实现源码见 e2b/_e2b_environment.py 与 daytona/_daytona_environment.py。接口刻意保持精简只覆盖工作目录 shell 命令 字节级文件读写。它既不是文件系统抽象也不是容器 API设计上也不会往这个方向膨胀——这在BaseEnvironment的类文档中写得很明确src/google/adk/environment/_base_environment.py。快速开始挂载一个本地环境下面这个 Agent 在本机的一个临时目录中工作具备执行命令和编辑文件的能力from google.adk.agents import Agent from google.adk.environment import LocalEnvironment from google.adk.tools.environment import EnvironmentToolset root_agent Agent( namelocal_environment_agent, descriptionRuns shell commands and edits files in a scratch directory., instruction( You have a working directory where you can create files and run shell commands. Write a script before guessing, and read the error output when a command fails. ), tools[EnvironmentToolset(environmentLocalEnvironment())], )EnvironmentToolset会在内部替你调用initialize()——从源码可以看到它在get_tools()首次构建工具列表时以及注入系统指令的process_llm_request()中都会检查并初始化环境src/google/adk/tools/environment/_environment_toolset.py。因此上例中你无需手动管理生命周期。如果你要直接驱动一个环境生命周期管理就是自己的职责了env LocalEnvironment() await env.initialize() await env.write_file(Path(notes/hello.txt), hi there) print(await env.read_file(Path(notes/hello.txt))) # bhi there result await env.execute(cat notes/hello.txt) print(result.exit_code, result.stdout) # 0 hi there await env.close()当你不传working_dir时initialize()会在系统临时目录下创建名为adk_workspace_*的临时目录随后close()会连同其中的一切内容一起删除。这一点在LocalEnvironment.initialize()的实现中可以看到它使用tempfile.mkdtemp(prefixadk_workspace_)创建目录并标记_auto_created Trueclose()时对该标记的目录执行shutil.rmtreesrc/google/adk/environment/_local_environment.py。工作原理三个必懂的核心机制在正式依赖某个环境之前有三件事值得先弄清楚每件都对应下面的一个小节工作目录何时创建、何时销毁execute在命令失败时返回什么LocalEnvironment究竟能保护你免受什么风险。生命周期构造 → initialize → 使用 → close一个环境严格按以下四步运转构造Construct构造函数只保存设置不做任何其他事。此时目录尚未创建进程也尚未启动。initialize()创建工作目录或连接远程工作区。各实现应当保证幂等性重复调用是安全的。execute/read_file/write_file工作阶段。close()释放initialize获取的资源。对LocalEnvironment而言这会删除自动创建的临时目录而你显式提供的目录则原样保留。BaseEnvironment上initialize和close都有默认的空实现因此不需要它们的环境实现可以干脆不重写。而其余四个成员——working_dir、execute、read_file、write_file——都是抽象成员abstractmethod子类若未全部实现则根本无法实例化src/google/adk/environment/_base_environment.py。is_initialized只是一个普通布尔属性由各实现自行维护。框架本身从不检查它请把它当作状态报告而非保护性开关。在LocalEnvironment上若在initialize()之前调用execute会抛出RuntimeError:working_diris not set. Call initialize() first.该错误同样来自read_file/write_file的守卫逻辑见 src/google/adk/environment/_local_environment.py。ExecutionResult失败不是异常而是返回值每次execute调用都会返回一个ExecutionResultdataclass无论命令发生了什么都不会抛异常字段类型默认值说明exit_codeint0进程退出状态码。stdoutstr捕获的标准输出。stderrstr捕获的标准错误。timed_outboolFalse命令是否超出了timeout限制。非零退出码是正常结果不是异常。失败的命令会带着退出码和填满的stderr原样返回注意到这一点是调用方的职责。实际调用方是工具透过工具最终是模型——所以在信任stdout之前务必先检查exit_code。timed_out是唯一必须与另一个字段配合解读的字段。当LocalEnvironment判定命令超时后它会杀死进程因此exit_code返回的是-9SIGKILL的负数而不是命令自己选择的任何值stdout里只保留到那一刻为止已刷新的内容。该 dataclass 定义见 src/google/adk/environment/_base_environment.py。LocalEnvironment 的实现细节LocalEnvironment.execute通过asyncio.create_subprocess_shell运行命令意味着你传入的字符串会交给 shell 解释管道、、重定向都可用shell 对文本能做的一切它都能做。子进程以工作目录作为cwd其环境为os.environ的副本并在其上合并env_varssrc/google/adk/environment/_local_environment.py。同时子进程以start_new_sessionTrue启动拥有独立的进程组这样超时清理时能连同其后代进程一起终止而不误伤其他进程。文件读写方面两个文件方法先将相对路径解析到工作目录再校验解析结果仍位于工作目录内。试图越界如../../etc/passwd的路径会在任何 I/O 发生前抛出ValueError: Path escapes working directory: path实现见_resolve_pathsrc/google/adk/environment/_local_environment.pywrite_file按需创建父目录接受str或bytes文本以 UTF-8 写入且关闭换行符转换newlineread_file则始终返回bytes两个文件方法通过asyncio.to_thread在 worker 线程上执行读写大文件不会阻塞事件循环src/google/adk/environment/_local_environment.py。write_file的换行符关闭行为、父目录自动创建、越界路径拒绝等语义均有对应的单元测试覆盖tests/unittests/environment/test_local_environment.py例如test_write_preserves_explicit_crlf验证 CRLF 序列不被改写test_rejects_relative_path_escape验证目录穿越被拦截。配置选项LocalEnvironment接受两个仅限关键字的参数其中第一个参数的意义比表面看起来更大参数类型默认值说明working_dirPath \| NoneNoneAgent 的工作目录。省略时自动创建临时目录并在关闭时删除。env_varsdict[str, str] \| NoneNone合并进子进程环境的额外变量。working_dir同时决定了归属权与位置传入目录若不存在则创建makedirs(exist_okTrue)close()时原样保留。这正是输出本身才是重点的场景所需比如 Agent 生成的报告、或它编辑过的仓库省略参数你会得到一个位于系统临时位置、随环境一起消失的一次性目录适合作为 scratch 工作区。env_vars合并到父进程环境的副本之上而不是替换它因此子进程会继承宿主进程的一切——包括环境中的凭据。API 上不存在从空环境开始的选项。如果需要在沙箱中彻底隔离环境变量应改用远程沙箱实现如 E2B 的env_vars是设置在沙箱内部的。进阶应用下面两个场景分别解决两类问题用非子进程的后端支撑该接口以及让两个工具集共用同一个工作目录。编写你自己的环境实现一个环境本质就是实现那四个抽象成员。生命周期钩子是可选的唯一值得注意的细节是initialize与close都可能被调用多次框架约定它们幂等。class MemoryEnvironment(BaseEnvironment): Keeps files in a dict; refuses to run commands. def __init__(self): self._files: dict[str, bytes] {} property def working_dir(self) - Path: return Path(/) async def execute(self, command: str, *, timeout: float | None None): return ExecutionResult(exit_code1, stderrThis environment has no shell.) async def read_file(self, path: Path) - bytes: try: return self._files[str(path)] except KeyError as e: raise FileNotFoundError(path) from e async def write_file(self, path: Path, content: str | bytes) - None: self._files[str(path)] ( content.encode(utf-8) if isinstance(content, str) else content )内置工具依赖两条约定即使是自定义环境也应遵守read_file对不存在的文件抛FileNotFoundError而不是返回空字节BaseEnvironment的 docstring 亦明确此约定见 src/google/adk/environment/_base_environment.pyexecute通过exit_code报告失败而不是抛异常。另外注意BaseEnvironment上的working_dir被标注为绝对路径absolute path自定义实现时应遵循这一契约src/google/adk/environment/_base_environment.py。跨对话共享同一个环境把环境只构造一次然后把同一个实例传给所有需要它的地方。EnvironmentToolset和SkillToolset都接受environment参数两者拿到同一对象时Agent 通过一个工具集写入的文件在另一个工具集中立即可见。社区示例 local_env_skill_toolset/agent.py 正是这样做的同一个LocalEnvironment()实例被传入SkillToolset技能脚本与文件读写共享同一工作目录。共享有一个必须提前理解的坑每个工具集都会在自己的close()中关闭环境且互不检查对方是否还在使用。关闭其中一个工具集共享的工作目录就会被删除另一个工具集上的每个工具随后都会失败错误信息统一为working_dir is not set. Call initialize() first.而且它不会自行恢复因为该工具集仍然认为自己已经初始化过了。要么给每个工具集各配一个环境要么在同一时刻关闭两者。环境的生命周期因此等于你构建的那个对象的生命周期与会话无关。放在模块级变量里的环境会被进程的所有用户共享——对单用户命令行 Agent 没问题但对服务器就是错误的设计。局限性务必在选型前了解LocalEnvironment不是沙箱。命令以你进程的用户身份运行携带你进程的环境变量只有文件路径被限制在工作目录内。execute完全不受限——命令可以读取该用户能读的一切、访问网络、写出工作目录。请在信任模型输出的场景使用它不信任时请使用远程沙箱。路径校验只覆盖read_file和write_file。它校验的是解析后的路径能拦下../../etc/passwd但对execute(cat /etc/passwd)无能为力。无流式输出。execute在命令结束后才返回stdout 与 stderr 全部缓存在内存中。长命令在退出前不产生任何输出打印 1GB 数据的命令就会在内存里撑起 1GB。close()不会停止正在运行的工作。它只释放目录Agent 遗留在后台的进程会存活下来。这些类属于实验性 API。请从google.adk.environment导入__init__.py中导出了BaseEnvironment、ExecutionResult、LocalEnvironment三个符号见 src/google/adk/environment/init.py而不要从更深的子模块导入同时预期函数签名未来可能变化。配套示例与延伸阅读仓库中提供了一组可直接运行的配套示例可对照本文逐步实践local_environment/agent.py基于LocalEnvironment挂载环境工具集的最简 Agentlocal_env_skill_toolset/agent.py让技能工具集与文件读写共享同一环境的示例e2b_environment/agent.py保持 Agent 形状完全不变仅把后端指向远程沙箱的对照示例。相关指南方面E2BEnvironment是本接口的远程沙箱实现依赖pip install google-adk[e2b]安装可选依赖默认镜像为 E2B 公共base模板沙箱 TTL 默认 300 秒并在每次操作时续期详见 e2b/_e2b_environment.pySkill、Frontmatter 与 Resources 指南 介绍了技能体系——技能脚本正是运行在本文所述的环境之中。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考