拆解 OpenHands（11）--- Runtime主要组件

本篇继续对 runtime 的解读主要介绍 插件、执行系统和环境这三个组件。因为本系列借鉴的文章过多可能在参考文献中有遗漏的文章如果有还请大家指出。0x01 三大组件本篇要介绍的几个组件如下ActionExecutor在 Runtime 中执行动作的核心组件ActionExecutor 初始化时会根据配置加载指定的插件。插件注册到 ActionExecutor 的插件字典。当接收到动作请求时ActionExecutor 会调用相应的方法执行动作。对于浏览动作ActionExecutor 会使用 BrowserEnv 来处理。如果涉及插件ActionExecutor 会通过插件系统处理AgentSkillsPlugin提供智能体技能功能的插件AgentSkillsPlugin 是一个插件继承自 Plugin 基类。Runtime 初始化时插件会被加载到插件字典中。 插件通过 PluginRequirement 机制被注册到系统中。特定动作触发时调用相应插件功能。BrowserEnv浏览器环境封装使用 BrowserGym 库。ActionExecutor 在初始化时根据配置决定是否启用浏览器环境。当需要执行浏览相关的动作时ActionExecutor 会调用 BrowserEnv 的方法。BrowserEnv 运行在一个独立的多进程环境中。0x02 数据流Runtime 的数据流如下Runtime 会发起动作请求 → ActionExecutor.run_action()ActionExecutor 根据动作类型调用相应的处理方法如果涉及插件通过插件系统处理如果涉及浏览器调用 BrowserEnv 处理返回观察结果给智能体。0x03 插件系统Runtime会遇到如下问题新增模块如自定义工具、新 LLM 模型时需修改核心代码扩展性差多任务并发执行时模块间交互频繁易出现性能瓶颈框架部署与运维复杂难以适配不同环境本地、云端、边缘端。因此业界大多采用微服务架构或插件化设计模块间通过标准化接口通信新增功能只需开发插件并注册。3.1 sandbox_pluginssandbox_plugins 在 OpenHands 的 CodeActAgent 中起到了关键作用主要用于定义和配置代理在沙箱环境中可以使用的工具和功能。这些插件是代理能够与环境交互并完成任务的基础工具集。sandbox_plugins 的定义和作用在 CodeActAgent 类中sandbox_plugins 是一个类属性定义了代理在沙箱环境中需要的插件sandbox_plugins: list[PluginRequirement] [ AgentSkillsRequirement(), JupyterRequirement(), ]这些插件为代理提供了在沙箱环境中执行任务所需的工具和功能。具体插件功能AgentSkillsRequirement 和 JupyterRequirement 是两个插件需求类。AgentSkillsRequirement提供了一系列 Python 函数和工具使代理能够执行各种操作包括文件操作、目录浏览、代码执行等基本技能。需要在 JupyterRequirement 之前初始化因为 Jupyter 需要使用这些函数。JupyterRequirement提供了交互式 Python 解释器环境允许代理执行 Python 代码依赖于 AgentSkillsRequirement 提供的函数。插件在系统中的使用从代码中可以看出这些插件在多个地方被使用在 Runtime 初始化时# 在 agent_session.py 中 self.runtime runtime_cls( pluginsagent.sandbox_plugins, )在 Runtime 中设置插件# 在 base.py 中 self.plugins copy.deepcopy(plugins) if plugins is not None and len(plugins) 0 else []这些插件为代理提供了以下能力执行 Bash 命令通过 AgentSkills 中的命令执行功能执行 Python 代码通过 Jupyter 插件提供 IPython 环境文件系统操作读取、写入、编辑文件目录浏览查看和导航文件系统其他实用工具各种辅助函数和工具我们接下来具体分析基类PluginAgentSkillsRequirement 和 JupyterPlugin3.2 Plugin 基类class Plugin: Base class for a plugin. This will be initialized by the runtime client, which will run inside docker. name: str abstractmethod async def initialize(self, username: str) - None: Initialize the plugin. pass abstractmethod async def run(self, action: Action) - Observation: Run the plugin for a given action. pass dataclass class PluginRequirement: Requirement for a plugin. name: str插件为ALL_PLUGINS { jupyter: JupyterPlugin, agent_skills: AgentSkillsPlugin, vscode: VSCodePlugin, }3.3 JupyterPluginJupyterPlugin是 OpenHands 框架中的 Jupyter 内核插件基于Plugin基类实现核心职责是启动 Jupyter Kernel Gateway内核网关服务提供 IPython 代码单元格的异步执行能力支持代码运行、输出捕获文本 / 图片及 Python 解释器路径获取是框架中集成交互式数据分析、代码调试等 Jupyter 相关功能的核心组件。核心特色跨平台适配兼容 Windows、Linux、macOS 系统针对不同系统采用差异化的进程启动方式Windows 用subprocess.Popen类 Unix 用asyncio.create_subprocess_shell。灵活的运行时支持区分本地运行时LocalRuntime与非本地运行时适配不同部署场景如沙箱环境、本地开发环境自动处理工作目录与环境变量配置。端口自动分配在 40000-49999 端口范围内自动查找可用 TCP 端口避免端口冲突。异步代码执行基于JupyterKernel封装异步代码执行逻辑支持超时控制能捕获文本输出与图片 URL 等结构化结果。环境隔离与兼容通过micromamba虚拟环境或本地环境变量确保依赖一致性支持 Poetry 项目的路径配置适配 OpenHands 框架的工程化部署。流程图代码dataclass class JupyterRequirement(PluginRequirement): Jupyter插件的依赖声明类用于框架识别插件依赖。 name: str jupyter # 依赖名称固定为jupyter class JupyterPlugin(Plugin): Jupyter插件提供Jupyter Kernel Gateway启动与IPython代码执行能力。 name: str jupyter # 插件名称固定为jupyter kernel_gateway_port: int # Jupyter Kernel Gateway服务端口 kernel_id: str # Jupyter内核ID gateway_process: asyncio.subprocess.Process | subprocess.Popen # 内核网关进程对象 python_interpreter_path: str # Python解释器路径 async def initialize( self, username: str, kernel_id: str openhands-default ) - None: 初始化Jupyter插件启动Kernel Gateway服务配置运行环境。 参数: username: 执行用户名称非本地运行时使用 kernel_id: Jupyter内核ID默认openhands-default # 在40000-49999端口范围内查找可用TCP端口避免冲突 self.kernel_gateway_port find_available_tcp_port(40000, 49999) self.kernel_id kernel_id # 判断是否为本地运行时通过环境变量LOCAL_RUNTIME_MODE标记 is_local_runtime os.environ.get(LOCAL_RUNTIME_MODE) 1 # 判断是否为Windows系统 is_windows sys.platform win32 if not is_local_runtime: # 非本地运行时配置用户切换前缀与Poetry虚拟环境 # 若启用SU_TO_USER则添加su - 用户名 -s 前缀切换用户执行命令 prefix fsu - {username} -s if SU_TO_USER else # 命令前缀切换到代码仓库目录配置环境变量使用micromamba虚拟环境 poetry_prefix ( cd /openhands/code\n export POETRY_VIRTUALENVS_PATH/openhands/poetry;\n export PYTHONPATH/openhands/code:$PYTHONPATH;\n export MAMBA_ROOT_PREFIX/openhands/micromamba;\n /openhands/micromamba/bin/micromamba run -n openhands ) else: # 本地运行时无需用户切换直接使用本地环境 prefix # 从环境变量获取代码仓库路径本地运行时必须配置 code_repo_path os.environ.get(OPENHANDS_REPO_PATH) if not code_repo_path: raise ValueError( OPENHANDS_REPO_PATH environment variable is not set. This is required for the jupyter plugin to work with LocalRuntime. ) # 命令前缀切换到代码仓库目录本地环境依赖PATH确保环境正确 poetry_prefix fcd {code_repo_path}\n if is_windows: # Windows系统构建CMD格式的启动命令 jupyter_launch_command ( fcd /d {code_repo_path} # 切换到代码仓库目录/d参数支持跨盘符切换 f{sys.executable} -m jupyter kernelgateway # 启动Jupyter Kernel Gateway --KernelGatewayApp.ip0.0.0.0 # 绑定所有网络接口 f--KernelGatewayApp.port{self.kernel_gateway_port} # 指定端口 ) # Windows系统使用同步subprocess.Popen启动进程asyncio在Windows有兼容性限制 self.gateway_process subprocess.Popen( # type: ignore[ASYNC101] # noqa: ASYNC101 jupyter_launch_command, stdoutsubprocess.PIPE, # 捕获标准输出 stderrsubprocess.STDOUT, # 标准错误重定向到标准输出 shellTrue, # 使用shell执行命令 textTrue, # 输出以文本模式返回 ) # Windows系统同步等待Kernel Gateway启动读取输出直到包含at字符标识服务就绪 output while should_continue(): if self.gateway_process.stdout is None: time.sleep(1) # 无输出时等待1秒 continue line self.gateway_process.stdout.readline() # 读取一行输出 if not line: time.sleep(1) continue output line if at in line: # 服务启动成功的标识输出含at如Listening at... break time.sleep(1) else: # 类Unix系统Linux/macOS构建Bash格式的启动命令 jupyter_launch_command ( f{prefix}/bin/bash EOF\n # 切换到bash执行EOF避免变量解析 f{poetry_prefix} # 环境配置前缀虚拟环境/工作目录 f{sys.executable} -m jupyter kernelgateway # 启动Kernel Gateway --KernelGatewayApp.ip0.0.0.0 # 绑定所有网络接口 f--KernelGatewayApp.port{self.kernel_gateway_port}\n # 指定端口 EOF ) # 类Unix系统使用asyncio创建异步子进程避免阻塞事件循环 self.gateway_process await asyncio.create_subprocess_shell( jupyter_launch_command, stderrasyncio.subprocess.STDOUT, # 标准错误重定向到标准输出 stdoutasyncio.subprocess.PIPE, # 捕获标准输出 ) # 异步等待Kernel Gateway启动读取输出直到包含at字符 output while should_continue() and self.gateway_process.stdout is not None: line_bytes await self.gateway_process.stdout.readline() # 异步读取一行输出 line line_bytes.decode(utf-8) # 字节转字符串 output line if at in line: break await asyncio.sleep(1) # 等待1秒 # 执行测试代码获取当前Python解释器路径验证环境正确性 _obs await self.run( IPythonRunCellAction(codeimport sys; print(sys.executable)) ) self.python_interpreter_path _obs.content.strip() # 提取并保存解释器路径 async def _run(self, action: Action) - IPythonRunCellObservation: 内部方法在Jupyter内核中执行代码单元格。 参数: action: 待执行的动作必须是IPythonRunCellAction类型 返回: IPythonRunCellObservation: 代码执行结果的观察值含文本内容、图片URL等 # 校验动作类型仅支持IPythonRunCellAction if not isinstance(action, IPythonRunCellAction): raise ValueError( fJupyter plugin only supports IPythonRunCellAction, but got {action} ) # 初始化JupyterKernel若未初始化 if not hasattr(self, kernel): self.kernel JupyterKernel( flocalhost:{self.kernel_gateway_port}, # 内核网关地址本地端口 self.kernel_id # 内核ID ) # 若内核未初始化执行初始化建立连接 if not self.kernel.initialized: await self.kernel.initialize() # 异步执行代码支持超时控制超时时间从action获取 output await self.kernel.execute(action.code, timeoutaction.timeout) # 从结构化输出中提取文本内容与图片URL text_content output.get(text, ) # 文本输出stdout/stderr image_urls output.get(images, []) # 图片URL列表如matplotlib绘图结果 # 返回封装后的观察结果 return IPythonRunCellObservation( contenttext_content, # 文本内容 codeaction.code, # 执行的代码 image_urlsimage_urls if image_urls else None, # 图片URL无则为None ) async def run(self, action: Action) - IPythonRunCellObservation: 公开接口执行IPython代码动作返回观察结果。 参数: action: 待执行的IPythonRunCellAction动作 返回: IPythonRunCellObservation: 代码执行结果 # 调用内部_run方法执行代码返回结果 obs await self._run(action) return obs3.4 AgentSkillsPlugin功能概述AgentSkillsPlugin是 OpenHands 框架中管理智能体技能Agent Skills的核心插件负责整合文件操作file_ops、文件读取file_reader、代码仓库操作repo_ops等基础技能模块通过动态导入机制将分散的技能函数统一暴露给框架同时提供插件依赖声明与文档自动生成能力是智能体获取文件处理、仓库管理等核心操作能力的关键组件。class AgentSkillsPlugin(Plugin): name: str agent_skills async def initialize(self, username: str) - None: Initialize the plugin. pass async def run(self, action: Action) - Observation: Run the plugin for a given action. raise NotImplementedError(AgentSkillsPlugin does not support run method)核心特色模块化技能整合通过动态导入机制将file_ops、file_reader、repo_ops等独立模块的技能函数统一聚合简化框架对技能的调用与管理。自动文档生成扫描所有导入的技能函数提取函数签名与文档字符串__doc__自动生成标准化文档提升开发可维护性。柔性依赖处理对repo_ops模块采用可选导入策略导入失败时仅跳过该模块不影响其他技能的正常使用增强插件兼容性。极简初始化设计插件初始化逻辑为空实现无需额外配置聚焦于技能函数的聚合与暴露降低使用门槛。明确的接口约束禁用run方法抛出未实现异常明确该插件的核心作用是技能聚合而非直接执行动作避免误用。AgentSkillsRequirementAgentSkillsRequirement 是一个插件需求类它定义了代理在沙箱环境中运行所需的基本技能集合这些技能主要通过 Python 函数的形式提供使代理能够执行各种操作为代理提供了与文件系统交互的基本能力提供了执行命令和脚本的工具为其他高级插件如 Jupyter提供了基础函数支持确保代理能够在沙箱环境中完成大多数常见的开发任务AgentSkillsRequirement 的主要功能如下文件系统操作提供读取、写入、编辑文件的能力支持目录浏览和文件管理操作允许代理查看和操作工作区中的文件命令执行提供执行 shell 命令的能力允许代理在沙箱环境中运行 bash 命令支持与操作系统交互的各种操作工具函数集合提供一系列实用的 Python 函数这些函数可以被其他插件如 Jupyter使用包括各种辅助功能如字符串处理、数据操作等在 CodeActAgent 中AgentSkillsRequirement 被定义在 sandbox_plugins 列表中sandbox_plugins: list[PluginRequirement] [ AgentSkillsRequirement(), JupyterRequirement(), ]AgentSkillsRequirement 与其他组件的关系与 JupyterRequirement 的关系AgentSkillsRequirement 必须在 JupyterRequirement 之前初始化AgentSkillsRequirement 提供的 Python 函数会被 Jupyter 环境使用这种顺序确保了 Jupyter 可以访问所有必要的工具函数与 Runtime 的关系在 LocalRuntime 和其他运行时环境中这些插件会被加载和初始化总的来说AgentSkillsRequirement 是代理在 OpenHands 环境中执行任务的基础它提供了一套核心函数使代理能够与文件系统、命令行和运行环境进行交互。框架注册与技能发现OpenHands 框架通过「插件注册机制」识别AgentSkillsPlugin并自动发现其聚合的所有具体 Skill 操作步骤如下插件注册与依赖声明AgentSkillsPlugin继承自框架的Plugin基类通过AgentSkillsRequirement声明依赖框架启动时会自动扫描并加载该插件dataclass class AgentSkillsRequirement(PluginRequirement): name: str agent_skills # 插件依赖名称与插件名一致 documentation: str agentskills.DOCUMENTATION class AgentSkillsPlugin(Plugin): name: str agent_skills # 插件名称框架通过该名称识别框架解析技能清单框架加载AgentSkillsPlugin后会读取其__all__变量和全局命名空间提取所有 Skill 函数的关键信息函数名如create_file作为 Skill 的唯一标识函数签名参数、返回值通过inspect.signature解析用于智能体构造调用参数文档字符串__doc__自动生成技能文档供智能体参考使用。技能全局注册框架将解析后的 Skill 信息注册到「全局技能注册表」中形成 key-value 映射keySkill 函数名valueSkill 函数对象 元数据使智能体可通过函数名快速查找并调用对应 Skill。智能体调用具体 Skill 操作智能体Agent通过框架提供的接口从「全局技能注册表」中获取AgentSkillsPlugin聚合的具体 Skill并触发执行logger.debug(Initializing AgentSkills) if agent_skills in self.plugins and jupyter in self.plugins: obs await self.run_ipython( IPythonRunCellAction( codefrom openhands.runtime.plugins.agent_skills.agentskills import *\n ) ) logger.debug(fAgentSkills initialized: {obs})流程图代码dataclass class AgentSkillsRequirement(PluginRequirement): AgentSkillsPlugin 的依赖声明类用于框架识别插件依赖。 name: str agent_skills # 依赖名称固定为agent_skills documentation: str agentskills.DOCUMENTATION # 依赖文档来自agentskills模块 class AgentSkillsPlugin(Plugin): 智能体技能插件负责聚合各类基础技能函数文件操作、仓库操作等。 name: str agent_skills # 插件名称固定为agent_skills async def initialize(self, username: str) - None: 初始化插件空实现无需额外配置。 pass async def run(self, action: Action) - Observation: 执行插件动作禁用该方法。 该插件的核心作用是聚合技能函数而非直接执行动作因此抛出未实现异常。 raise NotImplementedError(AgentSkillsPlugin does not support run method) # 动态导入file_ops模块的所有技能函数添加到当前模块全局命名空间 import_functions( modulefile_ops, # 源模块文件操作模块如创建/删除/修改文件 function_namesfile_ops.__all__, # 导入的函数列表file_ops定义的所有公开函数 target_globalsglobals() # 目标命名空间当前模块全局变量 ) # 动态导入file_reader模块的所有技能函数添加到当前模块全局命名空间 import_functions( modulefile_reader, # 源模块文件读取模块如读取文本文件、解析JSON等 function_namesfile_reader.__all__, # 导入的函数列表 target_globalsglobals() ) # 初始化__all__列表包含已导入的所有技能函数供外部模块导入 __all__ file_ops.__all__ file_reader.__all__ # 可选导入repo_ops模块代码仓库操作如Git克隆、提交等 try: from openhands.runtime.plugins.agent_skills import repo_ops # 动态导入repo_ops模块的所有技能函数 import_functions( modulerepo_ops, function_namesrepo_ops.__all__, target_globalsglobals() ) # 将repo_ops的技能函数添加到__all__ __all__ repo_ops.__all__ except ImportError: # 若repo_ops模块不可用如未安装依赖则跳过导入不影响其他功能 pass # 自动生成所有技能函数的标准化文档 DOCUMENTATION for func_name in __all__: # 从全局命名空间获取技能函数 func globals()[func_name] # 获取函数的文档字符串 cur_doc func.__doc__ # 清理文档字符串去除空行、统一去除每行缩进 cur_doc \n.join(filter(None, map(lambda x: x.strip(), cur_doc.split(\n)))) # 格式化文档每行添加4个空格缩进保证格式统一 cur_doc \n.join(map(lambda x: * 4 x, cur_doc.split(\n))) # 提取函数签名函数名参数列表 fn_signature f{func.__name__} str(signature(func)) # 将函数签名与格式化文档添加到总文档 DOCUMENTATION f{fn_signature}:\n{cur_doc}\n\n # 单独添加file_editor技能函数特殊处理未包含在上述模块中 from openhands.runtime.plugins.agent_skills.file_editor import file_editor # noqa: E402 __all__ [file_editor] # 将file_editor添加到__all__供外部导入0x04 执行系统ActionExecutor是 OpenHands 框架中运行于 Docker 沙箱内的核心动作执行组件负责接收并执行来自后端的各类动作如命令行执行、IPython 代码运行、浏览器操作等生成对应的观察结果Observation同时管理插件生命周期、用户环境、工作目录及资源监控是连接后端指令与沙箱执行环境的关键桥梁。4.1 调用在 OpenHands 项目中ActionExecutor 类的调用与使用流程总结服务端ActionExecutor 在 action_execution_server.py 中作为独立服务运行。客户端各种运行时实现如 LocalRuntime通过 HTTP 请求与 ActionExecutor 通信。执行步骤Runtime实际就是ActionExecutionClient的派生类直接或间接调用 execute_action() 方法class RemoteRuntime(ActionExecutionClient)class LocalRuntime(ActionExecutionClient)class DockerRuntime(ActionExecutionClient)通过 HTTP POST 请求发送到 /execute_action 端点ActionExecutor 接收请求并执行相应操作将观察结果返回给客户端。这种架构把操作执行与主应用分离提供了更好的隔离性和安全性。4.2action_execution_client.pyaction_execution_client.py文件包含ActionExecutionClient类它实现了运行时接口。它是一个抽象实现意味着仍需要通过具体实现来扩展才能使用。ActionExecutionClient 通过HTTP调用与action_execution_server 交互以实际执行运行时操作。ActionExecutionClient 在各种运行时实现中被使用。例如在 LocalRuntime 中操作会通过客户端发送到 ActionExecutor。4.3 action_execution_server.pyActionExecutor 在 openhands/runtime/action_execution_server.py 文件中充当核心组件。该文件将其实例化并作为 FastAPI 应用程序的核心执行器。即action_executor.py文件包含ActionExecutor类负责通过/execute_actionHTTP端点接收的动作。它在HTTP响应中返回观察结果。核心特色如下沙箱内环境管理初始化用户权限与工作目录支持 Windows/Linux 跨平台环境适配确保执行隔离性与安全性。插件化架构支持加载 VSCode、Jupyter 等各类插件通过统一接口管理插件初始化与调用灵活扩展执行能力。多环境协同集成 Bash/Windows PowerShell 命令行环境、浏览器环境BrowserEnv、Jupyter 交互式环境满足多样化动作执行需求。异步初始化优化浏览器环境采用异步初始化避免阻塞主流程提升启动效率。资源与状态监控支持内存上限配置与内存监控同步 Bash 与 Jupyter 工作目录确保执行上下文一致性。异常处理与兼容性针对 Windows 系统、插件缺失等场景做特殊兼容处理抛出明确异常并记录日志。具体会在main中启动if __name__ __main__: logger.debug(fStarting action execution API on port {args.port}) # When LOG_JSON1, provide a JSON log config to Uvicorn so error/access logs are structured log_config None if os.getenv(LOG_JSON, 0) in (1, true, True): log_config get_uvicorn_json_log_config() run(app, host0.0.0.0, portargs.port, log_configlog_config, use_colorsFalse)在 lifespan 函数中初始化asynccontextmanager async def lifespan(app: FastAPI): global client, mcp_proxy_manager logger.info(Initializing ActionExecutor...) client ActionExecutor( plugins_to_load, work_dirargs.working_dir, usernameargs.username, user_idargs.user_id, enable_browserargs.enable_browser, browsergym_eval_envargs.browsergym_eval_env, ) await client.ainit() logger.info(ActionExecutor initialized.) # Check if were on Windows is_windows sys.platform win32 # Initialize and mount MCP Proxy Manager (skip on Windows) if is_windows: logger.info(Skipping MCP Proxy initialization on Windows) mcp_proxy_manager None else: logger.info(Initializing MCP Proxy Manager...) # Create a MCP Proxy Manager mcp_proxy_manager MCPProxyManager( auth_enabledbool(SESSION_API_KEY), api_keySESSION_API_KEY, logger_levellogger.getEffectiveLevel(), ) mcp_proxy_manager.initialize() # Mount the proxy to the app allowed_origins [*] try: await mcp_proxy_manager.mount_to_app(app, allowed_origins) except Exception as e: logger.error(fError mounting MCP Proxy: {e}, exc_infoTrue) raise RuntimeError(fCannot mount MCP Proxy: {e}) yield # Clean up release the resources logger.info(Shutting down MCP Proxy Manager...) if mcp_proxy_manager: del mcp_proxy_manager mcp_proxy_manager None else: logger.info(MCP Proxy Manager instance not found for shutdown.) logger.info(Closing ActionExecutor...) if client: try: client.close() logger.info(ActionExecutor closed successfully.) except Exception as e: logger.error(fError closing ActionExecutor: {e}, exc_infoTrue) else: logger.info(ActionExecutor instance not found for closing.) logger.info(Shutdown complete.)