
langchain-daytona 实战指南用 Daytona 云端沙箱为 Deep Agents 提供隔离执行后端【免费下载链接】deepagentsThe batteries-included agent harness.项目地址: https://gitcode.com/GitHub_Trending/de/deepagents本指南聚焦 Deep Agents 官方沙箱合作伙伴集成 langchain-daytona位于仓库libs/partners/daytona完整讲解如何安装依赖、创建DaytonaSandbox后端、执行 Shell 命令以及上传/下载文件并深入源码剖析其基于 Daytona 会话session的命令轮询执行链路、可自定义的超时与轮询策略以及deepagents.backends.sandbox.BaseSandbox派生的文件操作能力。读完本文你将能够把 Daytona 云沙箱无缝接入 Deep Agents 的 agent 运行时并具备排查超时、轮询与文件传输问题的源码级认知。一、定位Deep Agents 的 Daytona 云沙箱后端langchain-daytona是 Daytona sandbox integration for Deep Agents它把 Daytona 扩展了BackendProtocol额外要求提供 Shell 命令执行能力execute()/aexecute()与唯一标识id其设计目标正是面向运行在隔离环境容器、虚拟机、远程主机中的后端。DaytonaSandbox正是这一协议的 Daytona 实现它继承自BaseSandbox后者已经基于execute()与upload_files()派生出了ls、glob、grep、read、write、edit等一整套文件操作。也就是说只要实现了execute()与upload_files()其余文件能力会自动获得——DaytonaSandbox的 docstring 明确写到“This implementation inherits all file operation methods from BaseSandbox and only implements the execute() method using Daytonas API.”因此理解langchain-daytona的关键就落在三个点上如何包装 Daytona SDK 的Sandbox实例、如何执行命令并轮询结果、如何传输文件。二、快速安装与最小可运行示例包以langchain-daytona为名发布在 PyPI 上使用 uv 一行安装uv add langchain-daytona安装完成后README 给出的最小示例即可运行from daytona import Daytona from langchain_daytona import DaytonaSandbox sandbox Daytona().create() backend DaytonaSandbox( sandboxsandbox, timeout300, sync_polling_interval0.25, ) result backend.execute(echo hello) print(result.output)这段代码依次完成四件事通过daytona.Daytona().create()创建一个 Daytona 沙箱实例需要先按 Daytona SDK 的规范完成 API Key 等配置将沙箱包装成DaytonaSandbox后端并显式指定默认命令超时timeout300秒与同步轮询间隔sync_polling_interval0.25秒调用backend.execute(echo hello)在沙箱内执行 Shell 命令打印result.output即命令的合并输出。DaytonaSandbox由包的__init__.py导出langchain_daytona/__init__.py中__all__ [DaytonaSandbox]因此from langchain_daytona import DaytonaSandbox是唯一的官方导入方式。三、DaytonaSandbox 构造参数详解DaytonaSandbox的构造函数签名见 sandbox.py只接受关键字参数def __init__( self, *, sandbox: daytona.Sandbox, timeout: int 30 * 60, sync_polling_interval: SyncPollingInterval 0.1, ) - None:各参数的作用如下参数类型默认值说明sandboxdaytona.Sandbox无必填需要包装的、已创建的 Daytona 沙箱实例通常由Daytona().create()返回timeoutint30 * 601800 秒调用execute()且未显式传timeout时使用的默认命令超时单位秒sync_polling_intervalfloat \| Callable[[float], float]0.1同步执行路径上轮询 Daytona 命令完成状态的间隔既可以是固定秒数也可以是一个接收“已执行时长秒”、返回下一次轮询间隔的函数自适应策略值得注意的两个设计点超时的特殊语义execute()的 docstring 明确说明在 Daytona 实现中timeout0表示“无限期等待”。这一语义被单元测试test_execute_polls_with_callable_interval所覆盖见 tests/unit_tests/test_import.py其中以timeout0调用execute。轮询间隔的双形态SyncPollingInterval类型定义为float | Callable[[float], float]。构造函数会把float包装成一个忽略elapsed、恒定返回该数值的内部函数如果传入的是可调用对象则原样作为轮询策略使用见 sandbox.py。另外构造时传入的daytona.Sandbox实例会被保存在内部后端实例的id属性直接透传沙箱的idsandbox.py可据此唯一标识后端实例。四、execute()基于会话日志的完整执行链路execute(command, *, timeoutNone)是后端最核心的入口。其执行逻辑并不直接“跑一条命令、拿一个结果”而是围绕 Daytona 的**会话session**机制构建了“创建会话 → 异步启动命令 → 轮询完成 → 拉取日志 → 清理会话”的完整生命周期。整体实现在_execute_via_session_logs()中sandbox.pysession_id str(uuid4()) self._sandbox.process.create_session(session_id) try: started_at time.monotonic() result self._sandbox.process.execute_session_command( session_id, SessionExecuteRequest(commandcommand, run_asyncTrue), timeouttimeout, ) while True: if timeout ! 0 and time.monotonic() - started_at timeout: msg fCommand timed out after {timeout} seconds return ExecuteResponse(outputmsg, exit_code124, truncatedFalse) command_result self._sandbox.process.get_session_command( session_id, result.cmd_id, ) if command_result.exit_code is not None: break elapsed time.monotonic() - started_at time.sleep(self._sync_polling_interval(elapsed)) logs self._sandbox.process.get_session_command_logs( session_id, result.cmd_id, ) finally: self._sandbox.process.delete_session(session_id)逐步拆解这条链路创建会话为每次执行生成新的uuid4()作为session_id调用process.create_session创建会话保证每次命令在独立会话中执行、互不干扰。异步启动命令通过execute_session_command下发SessionExecuteRequest(commandcommand, run_asyncTrue)——run_asyncTrue是关键命令立即返回带cmd_id而不是阻塞等待执行完毕。轮询完成状态循环调用get_session_command(session_id, cmd_id)读取命令状态。只要exit_code is None尚未结束就根据已执行时长elapsed计算本次轮询间隔并time.sleep直到拿到非None的exit_code才跳出循环。超时熔断timeout ! 0时若monotonic()已流逝时间达到timeout立即返回一个特殊的ExecuteResponseoutput为Command timed out after N secondsexit_code124对应 GNU timeout 的退出码truncatedFalse。拉取日志命令结束后调用get_session_command_logs获取 stdout/stderr。确保清理finally中调用delete_session删除会话即使命令超时返回也不会泄漏会话资源。execute()的返回值类型是ExecuteResponse包含三个字段protocol.pyoutput合并后的 stdoutstderr 文本、exit_code0 为成功非 0 为失败None表示无法确定、truncated输出是否因后端限制被截断。注意 Daytona 实现目前恒为truncatedFalse。关于 stderr 的处理有一个值得一提的细节sandbox.py当logs.stderr非空时会以\nstderr.../stderr的形式拼接在 stdout 之后而不是混入普通文本便于调用方区分标准输出与错误输出。单元测试 test_execute_returns_stdout 验证了这条链路的基本契约execute(echo hello world)返回output hello world、exit_code 0、truncated is False并且create_session与delete_session各恰好被调用一次——印证了“每次执行创建并销毁一个会话”的实现。而 test_execute_timeout 则验证了超时路径模拟monotonic()从 0 跳到 11 秒后execute(sleep 999, timeout10)返回exit_code 124且输出包含timed out。五、轮询策略固定间隔与自适应 Callable轮询间隔支持两种形态且可以直接从单元测试中看到它们各自的行为模式固定间隔。传入float时每次轮询固定休眠该时长。测试 test_execute_polls_with_fixed_interval 以sync_polling_interval0.25运行前两次get_session_command返回exit_codeNone、第三次返回exit_code0最终断言sleep被调用两次且参数均为0.25。自适应 Callable。传入Callable[[float], float]时每次轮询前会把“当前已执行秒数”传进去由其返回下一次等待间隔。测试 test_execute_polls_with_callable_interval 给出了一个典型的分段退避示例def adaptive_interval(elapsed: float) - float: interval ( 0.1 if elapsed ADAPTIVE_POLLING_FAST_THRESHOLD # 1 秒内0.1s else 0.2 if elapsed ADAPTIVE_POLLING_MEDIUM_THRESHOLD # 10 秒内0.2s else 1.0 # 超过 10 秒1.0s ) return interval在模拟的monotonic()序列[0.0, 0.0, 0.5, 5.0, 65.0]下断言得到的实际休眠序列为[0.1, 0.1, 0.2]——即命令刚启动时高频轮询随着执行时间变长逐步降低轮询频率。这种策略非常适合“短命令多、偶发长命令”的 agent 工作负载既保证短命令的低延迟感知又避免长时间轮询对 Daytona API 造成无谓压力。从源码注释看这类自适应策略“最终只会保留在同步路径上”未来异步实现就绪后会有独立的优化但当前接口形态稳定可用。六、文件传输download_files 与 upload_filesDaytonaSandbox在execute()之外还显式实现了两个文件操作download_files()与upload_files()sandbox.py。二者都遵循统一的路径前置校验逻辑只接受绝对路径以/开头不满足的路径直接返回errorinvalid_path不会发起任何 Daytona API 调用合法路径批量打包成请求后通过sandbox.fs.download_files(FileDownloadRequest(sourcepath))/sandbox.fs.upload_files(FileUpload(sourcecontent, destinationpath))一次性传输返回值按请求路径的顺序一一对应方便调用方按索引匹配结果。具体到download_files()其错误映射有两层Daytona 返回result is None文件不存在/读取失败时映射为errorfile_not_found对每个请求路径若结果序列不足极端异常情况也会兜底补一个errorfile_not_found文件内容content直接透传 Daytona SDK 返回的字节数据代码注释说明 Daytona SDK 返回bytes。这两个方法的返回类型分别是FileDownloadResponsepath/content/error与FileUploadResponsepath/error定义于 protocol.py。需要强调的是除此之外的其余文件能力全部由BaseSandbox自动派生。以 BaseSandbox 的模块注释为准ls、glob、grep、read通过execute()执行沙箱内的 Shell 命令如python3内嵌脚本完成write通过upload_files()传输内容edit在 payload 小于_EDIT_INLINE_MAX_BYTES50000 字节时走服务端内联替换更大时则退化为“上传临时文件 服务端替换脚本”的路径。也就是说只要实现了execute()与upload_files()agent 在沙箱内“读、写、改、搜、列”的完整能力就齐备了。七、测试与验证标准集成测试 针对性单元测试langchain-daytona的测试分为两层均可从仓库直接查看与复跑。单元测试tests/unit_tests/test_import.py使用MagicMock模拟 Daytona SDK不依赖真实云环境覆盖了test_import_daytona包可正常导入test_execute_returns_stdout正常输出、退出码、会话创建/销毁次数test_execute_polls_with_fixed_interval固定间隔轮询行为test_execute_polls_with_callable_interval自适应轮询 timeout0无限等待语义test_execute_timeout超时熔断退出码 124、输出含 timed out。集成测试tests/integration_tests/test_integration.py则直接复用 LangChain 生态的langchain_tests.integration_tests.SandboxIntegrationTests标准套件类级 fixture 中完成“创建 → 包装 → 用后删除”的完整生命周期pytest.fixture(scopeclass) def sandbox(self) - Iterator[SandboxBackendProtocol]: sdk daytona.Daytona() sandbox sdk.create() backend DaytonaSandbox(sandboxsandbox) try: yield backend finally: sandbox.delete()这一测试模式的价值在于SandboxIntegrationTests是 LangChain 为“沙箱后端协议”预置的统一契约测试集DaytonaSandbox只需通过一个 fixture 注入即可验证其对协议的全部约定执行、文件读写、编辑、搜索等无需为每个后端重复编写。这也解释了为什么包级测试还保留了一个轻量的 tests/test_import.py 仅验证导入。八、工程细节版本、依赖与发布配置从 pyproject.toml 可以确认以下事实包名与版本langchain-daytona当前版本0.0.8MIT 协议Python 版本要求3.11,4.0明确支持的 CPython 为 3.11、3.12、3.13、3.14运行时依赖deepagents0.7.0,0.8.0与daytonaSDK。其中deepagents在[tool.uv.sources]中被声明为指向仓库内../../deepagents的可编辑路径editable即仓库内联合开发时直接使用本地源码构建后端hatchlingwheel 仅打包langchain_daytona包目录测试依赖pytest 系列含 pytest-timeout、pytest-asyncio、pytest-xdist、ruff、ty 以及langchain-tests1.1.9提供SandboxIntegrationTests测试配置pytest 以--strict-markers --strict-config运行filterwarnings将警告视为错误并针对 Python 3.14 上第三方库的已知弃用警告做了定向豁免如daytonaSDK 中asyncio.iscoroutinefunction的弃用调用。仓库中的libs/partners/目录还收录了 modal、quickjs、runloop、vercel 等同类合作伙伴沙箱集成langchain-daytona与它们共享同一套SandboxBackendProtocol后端抽象——理解 Daytona 这一家的实现也就理解了这套“云沙箱即 Deep Agents 后端”的通用接入范式。九、常见问题与使用建议基于以上源码事实归纳几条实际使用建议不要忘记删除沙箱DaytonaSandbox只负责包装沙箱生命周期创建/删除由调用方管理。集成测试中sandbox.delete()在finally中执行生产代码也应遵循同样模式避免云端资源泄漏。合理设置默认超时默认timeout30*60对大多数命令足够宽裕对长任务可以显式传timeout需要无限等待时使用timeout0务必确认这是你真正想要的语义。长命令用自适应轮询当工作负载存在“秒级短命令 分钟级长命令”混跑时为sync_polling_interval传入分段退避 Callable可在延迟与 API 压力之间取得平衡固定小间隔如0.1会带来更频繁的get_session_command轮询。路径必须为绝对路径upload_files/download_files只接受以/开头的绝对路径相对路径会以errorinvalid_path被拒绝。依赖版本保持匹配langchain-daytona对deepagents的依赖约束为0.7.0,0.8.0升级 Deep Agents 主包时需留意其版本区间避免接口漂移。至此从安装示例到会话执行链路、轮询策略、文件传输与测试体系langchain-daytona作为 Deep Agents 云端沙箱后端的完整面貌已经清晰。需要进一步了解 Deep Agents 后端协议的通用约定可继续阅读 BaseSandbox 实现 与 SandboxBackendProtocol 定义如需扩展自己的沙箱后端这两个文件是最直接的参考蓝本。【免费下载链接】deepagentsThe batteries-included agent harness.项目地址: https://gitcode.com/GitHub_Trending/de/deepagents创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考